better-claude-config/tests/test_catalog_review.py

"""Tests for catalog_review.py -- semantic diff, risk predicates, review
session (acknowledge-gating + TOCTOU blob pinning), key encryption, and
registry-lookup logic for the Catalog Console (issue #62)."""

from __future__ import annotations

import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import bcc_core as c
import catalog_review as r


def _entry(**overrides):
    base = {
        "id": "filesystem",
        "display": "Filesystem",
        "description": "desc",
        "category": "files",
        "homepage": "https://github.com/modelcontextprotocol/servers",
        "stars": 100,
        "official": True,
        "setup": "basic",
        "config": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem@1.0.0"],
        },
        "env_required": {},
        "docs_url": "https://github.com/modelcontextprotocol/servers",
        "notes": "",
        "last_release": "2026-01-01",
    }
    base.update(overrides)
    return base


def _catalog(*entries):
    return {"schema": 1, "version": 1, "servers": list(entries)}


# --------------------------------------------------------------------------- #
# diff_catalogs
# --------------------------------------------------------------------------- #
def test_diff_detects_added_entry():
    old = _catalog()
    new = _catalog(_entry())
    changes = r.diff_catalogs(old, new)
    assert len(changes) == 1
    assert changes[0].status == "added"
    assert changes[0].entry_id == "filesystem"
    assert changes[0].old is None


def test_diff_detects_removed_entry():
    old = _catalog(_entry())
    new = _catalog()
    changes = r.diff_catalogs(old, new)
    assert len(changes) == 1
    assert changes[0].status == "removed"
    assert changes[0].new is None


def test_diff_detects_no_change():
    e = _entry()
    old = _catalog(e)
    new = _catalog(dict(e))
    assert r.diff_catalogs(old, new) == []


def test_diff_detects_changed_command_and_args():
    old = _catalog(_entry())
    new = _catalog(_entry(config={"command": "uvx", "args": ["other-pkg@2.0.0"]}))
    changes = r.diff_catalogs(old, new)
    assert len(changes) == 1
    ch = changes[0]
    assert ch.status == "changed"
    fields = {fc.field for fc in ch.field_changes}
    assert "config.command" in fields
    assert "config.args" in fields


def test_diff_detects_description_change():
    old = _catalog(_entry())
    new = _catalog(_entry(description="new description"))
    changes = r.diff_catalogs(old, new)
    assert changes[0].field_changes == (r.FieldChange("description", "desc", "new description"),)


def test_diff_ignores_entries_without_id():
    old = _catalog()
    new = _catalog({"display": "no id"})
    assert r.diff_catalogs(old, new) == []


def test_diff_multiple_entries_sorted_by_id():
    old = _catalog(_entry(id="zeta"), _entry(id="alpha"))
    new = _catalog(
        _entry(id="zeta", description="changed"), _entry(id="alpha", description="changed")
    )
    changes = r.diff_catalogs(old, new)
    assert [c_.entry_id for c_ in changes] == ["alpha", "zeta"]


# --------------------------------------------------------------------------- #
# risk_env_required
# --------------------------------------------------------------------------- #
def test_risk_env_required_blocking_on_nonempty_value():
    change = r.EntryChange("x", "changed", None, _entry(env_required={"API_KEY": "sk-real-value"}))
    findings = r.risk_env_required(change)
    assert len(findings) == 1
    assert findings[0].severity == "blocking"
    assert findings[0].code == "env_required_value"


def test_risk_env_required_clean_on_empty_value():
    change = r.EntryChange("x", "changed", None, _entry(env_required={"API_KEY": ""}))
    assert r.risk_env_required(change) == []


# --------------------------------------------------------------------------- #
# risk_command_allowlist
# --------------------------------------------------------------------------- #
def test_risk_command_allowlist_blocks_disallowed_command():
    change = r.EntryChange(
        "x", "changed", None, _entry(config={"command": "bash", "args": ["-c", "evil"]})
    )
    findings = r.risk_command_allowlist(change)
    assert len(findings) == 1
    assert findings[0].severity == "blocking"


def test_risk_command_allowlist_allows_listed_command():
    for cmd in sorted(c.CATALOG_ALLOWED_COMMANDS):
        change = r.EntryChange("x", "changed", None, _entry(config={"command": cmd, "args": []}))
        assert r.risk_command_allowlist(change) == []


# --------------------------------------------------------------------------- #
# risk_non_ascii
# --------------------------------------------------------------------------- #
def test_risk_non_ascii_flags_homoglyph_id():
    # Cyrillic 'а' (U+0430) instead of Latin 'a' -- classic homoglyph swap.
    evil_id = "filаsystem"
    change = r.EntryChange(evil_id, "changed", None, _entry(id=evil_id))
    findings = r.risk_non_ascii(change)
    assert any(f.code == "non_ascii_id" for f in findings)
    assert findings[0].severity == "blocking"
    # the offending string must be rendered with escapes, not raw
    assert "\\u0430" in findings[0].message


def test_risk_non_ascii_flags_arg():
    change = r.EntryChange(
        "x", "changed", None, _entry(config={"command": "npx", "args": ["pаckage@1.0.0"]})
    )
    findings = r.risk_non_ascii(change)
    assert any(f.code == "non_ascii_arg" for f in findings)


def test_risk_non_ascii_clean_for_ascii_entry():
    change = r.EntryChange("x", "changed", None, _entry())
    assert r.risk_non_ascii(change) == []


# --------------------------------------------------------------------------- #
# risk_unpinned_package
# --------------------------------------------------------------------------- #
def test_risk_unpinned_npm_package_no_version():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(config={"command": "npx", "args": ["-y", "@scope/pkg"]}),
    )
    findings = r.risk_unpinned_package(change)
    assert len(findings) == 1
    assert findings[0].code == "unpinned_npm_package"
    assert findings[0].severity == "blocking"


def test_risk_pinned_npm_package_is_clean():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(config={"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]}),
    )
    assert r.risk_unpinned_package(change) == []


def test_risk_unpinned_unscoped_npm_package():
    change = r.EntryChange(
        "x", "changed", None, _entry(config={"command": "npx", "args": ["-y", "somepkg"]})
    )
    findings = r.risk_unpinned_package(change)
    assert len(findings) == 1


def test_risk_unpinned_docker_latest_tag():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(config={"command": "docker", "args": ["run", "-i", "--rm", "myimage:latest"]}),
    )
    findings = r.risk_unpinned_package(change)
    assert len(findings) == 1
    assert findings[0].code == "unpinned_docker_image"


def test_risk_unpinned_docker_no_tag():
    change = r.EntryChange(
        "x", "changed", None, _entry(config={"command": "docker", "args": ["run", "myimage"]})
    )
    findings = r.risk_unpinned_package(change)
    assert len(findings) == 1


def test_risk_pinned_docker_image_is_clean():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(config={"command": "docker", "args": ["run", "-i", "--rm", "myimage:1.2.3"]}),
    )
    assert r.risk_unpinned_package(change) == []


def test_risk_docker_digest_pin_is_clean():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(
            config={
                "command": "docker",
                "args": ["run", "myimage@sha256:" + "a" * 64],
            }
        ),
    )
    assert r.risk_unpinned_package(change) == []


# --------------------------------------------------------------------------- #
# risk_url_domain_change
# --------------------------------------------------------------------------- #
def test_risk_url_domain_change_warns_on_lookalike_swap():
    old_entry = _entry(homepage="https://github.com/foo/bar")
    new_entry = _entry(homepage="https://githu6.com/foo/bar")
    change = r.EntryChange("x", "changed", old_entry, new_entry)
    findings = r.risk_url_domain_change(change)
    assert any(f.code == "domain_changed" for f in findings)
    domain_finding = next(f for f in findings if f.code == "domain_changed")
    assert "github.com" in domain_finding.message
    assert "githu6.com" in domain_finding.message
    assert domain_finding.severity == "warning"


def test_risk_url_domain_change_clean_when_domain_unchanged():
    old_entry = _entry(homepage="https://github.com/foo/bar")
    new_entry = _entry(homepage="https://github.com/foo/bar-renamed")
    change = r.EntryChange("x", "changed", old_entry, new_entry)
    assert r.risk_url_domain_change(change) == []


def test_risk_url_non_https_warns():
    new_entry = _entry(homepage="http://example.com")
    change = r.EntryChange("x", "changed", None, new_entry)
    findings = r.risk_url_domain_change(change)
    assert any(f.code == "non_https_url" for f in findings)


# --------------------------------------------------------------------------- #
# risk_new_entry / entry_risk_findings / has_blocking_risk
# --------------------------------------------------------------------------- #
def test_risk_new_entry_flags_added():
    change = r.EntryChange("x", "added", None, _entry())
    findings = r.risk_new_entry(change)
    assert len(findings) == 1
    assert findings[0].severity == "info"


def test_risk_new_entry_silent_for_changed():
    change = r.EntryChange("x", "changed", _entry(), _entry(description="x"))
    assert r.risk_new_entry(change) == []


def test_has_blocking_risk_true_for_disallowed_command():
    change = r.EntryChange("x", "changed", None, _entry(config={"command": "bash", "args": []}))
    assert r.has_blocking_risk(change) is True


def test_has_blocking_risk_false_for_clean_entry():
    change = r.EntryChange("x", "changed", _entry(), _entry(description="new"))
    assert r.has_blocking_risk(change) is False


# --------------------------------------------------------------------------- #
# ReviewSession: acknowledge gating
# --------------------------------------------------------------------------- #
def test_start_review_computes_diff():
    old = _catalog()
    new = _catalog(_entry())
    session = r.start_review("sha1", old, new)
    assert len(session.changes) == 1


def test_all_entries_acknowledged_false_initially():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    assert r.all_entries_acknowledged(session) is False


def test_acknowledge_entry_marks_acknowledged():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    r.acknowledge_entry(session, "filesystem")
    assert r.all_entries_acknowledged(session) is True


def test_acknowledge_unknown_entry_raises():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    with pytest.raises(ValueError):
        r.acknowledge_entry(session, "not-in-diff")


def test_acknowledge_gating_requires_every_entry():
    session = r.start_review("sha1", _catalog(), _catalog(_entry(id="a"), _entry(id="b")))
    r.acknowledge_entry(session, "a")
    assert r.all_entries_acknowledged(session) is False
    r.acknowledge_entry(session, "b")
    assert r.all_entries_acknowledged(session) is True


def test_no_acknowledge_all_function_exists():
    """Deliberate: there must be no shortcut to acknowledge every entry at
    once. See the comment in catalog_review.py above SignDecision."""
    names = [n for n in dir(r) if "acknowledge" in n.lower()]
    assert "acknowledge_all" not in names
    assert "acknowledge_all_entries" not in names


# --------------------------------------------------------------------------- #
# can_sign: TOCTOU blob pinning + acknowledge gating combined
# --------------------------------------------------------------------------- #
def test_can_sign_false_when_not_all_acknowledged():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    decision = r.can_sign(session, "sha1")
    assert decision.ok is False
    assert "acknowledged" in decision.reason


def test_can_sign_true_when_acknowledged_and_blob_matches():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    r.acknowledge_entry(session, "filesystem")
    decision = r.can_sign(session, "sha1")
    assert decision.ok is True
    assert decision.reason is None


def test_can_sign_refuses_on_blob_mismatch_even_if_acknowledged():
    """The core TOCTOU fix: acknowledging everything is not enough if the
    bytes on the remote changed underneath the review."""
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    r.acknowledge_entry(session, "filesystem")
    decision = r.can_sign(session, "sha2-a-new-commit-landed")
    assert decision.ok is False
    assert "changed" in decision.reason.lower() or "mismatch" in decision.reason.lower()


def test_can_sign_blob_mismatch_takes_priority_message():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    decision = r.can_sign(session, "sha2")
    assert decision.ok is False
    assert "blob" in decision.reason.lower() or "changed" in decision.reason.lower()


# --------------------------------------------------------------------------- #
# catalog_signing_message: domain separation must match bcc_core exactly
# --------------------------------------------------------------------------- #
def test_signing_message_uses_bcc_core_domain_prefix():
    raw = b'{"schema":1}'
    msg = r.catalog_signing_message(raw)
    assert msg == c._CATALOG_SIG_DOMAIN + raw
    assert msg.startswith(b"bcc-catalog-v1|")


def test_sign_then_verify_round_trips_with_bcc_core():
    """End-to-end: a signature produced by the Console's sign_catalog_bytes
    must verify with bcc_core.verify_catalog_signature -- proves the two
    modules can never drift on the domain-separation prefix."""
    seed, pubkey = r.generate_keypair()
    raw = b'{"schema":1,"version":2,"servers":[]}'
    sig = r.sign_catalog_bytes(raw, seed)
    assert c.verify_catalog_signature(raw, sig, [pubkey]) is True


def test_sign_tampered_bytes_fails_verify():
    seed, pubkey = r.generate_keypair()
    raw = b'{"schema":1,"version":2,"servers":[]}'
    sig = r.sign_catalog_bytes(raw, seed)
    tampered = raw[:-1] + b"0"
    assert c.verify_catalog_signature(tampered, sig, [pubkey]) is False


# --------------------------------------------------------------------------- #
# Key encryption at rest
# --------------------------------------------------------------------------- #
def test_encrypt_decrypt_round_trip():
    seed, _pub = r.generate_keypair()
    blob = r.encrypt_private_key(seed, "correct horse battery staple")
    decrypted = r.decrypt_private_key(blob, "correct horse battery staple")
    assert decrypted == seed


def test_decrypt_wrong_passphrase_raises():
    seed, _pub = r.generate_keypair()
    blob = r.encrypt_private_key(seed, "right passphrase")
    with pytest.raises(ValueError):
        r.decrypt_private_key(blob, "wrong passphrase")


def test_decrypt_corrupted_blob_raises():
    seed, _pub = r.generate_keypair()
    blob = r.encrypt_private_key(seed, "pass")
    corrupted = blob[:-1] + bytes([blob[-1] ^ 0xFF])
    with pytest.raises(ValueError):
        r.decrypt_private_key(corrupted, "pass")


def test_encrypt_private_key_rejects_wrong_length_seed():
    with pytest.raises(ValueError):
        r.encrypt_private_key(b"too-short", "pass")


def test_encrypt_private_key_rejects_empty_passphrase():
    seed, _pub = r.generate_keypair()
    with pytest.raises(ValueError):
        r.encrypt_private_key(seed, "")


def test_encrypted_blob_never_contains_seed_plaintext():
    seed, _pub = r.generate_keypair()
    blob = r.encrypt_private_key(seed, "some passphrase")
    assert seed not in blob


def test_generate_keypair_produces_valid_ed25519_pair():
    seed, pubkey = r.generate_keypair()
    assert len(seed) == 32
    assert len(pubkey) == 32
    raw = b"test payload"
    sig = r.sign_catalog_bytes(raw, seed)
    assert c.verify_catalog_signature(raw, sig, [pubkey]) is True


# --------------------------------------------------------------------------- #
# Registry lookup / near-neighbour edit distance
# --------------------------------------------------------------------------- #
def test_edit_distance_identical():
    assert r.edit_distance("abc", "abc") == 0


def test_edit_distance_one_substitution():
    assert r.edit_distance("firecrawl-mcp", "f1recrawl-mcp") == 1


def test_near_neighbor_ids_finds_close_match():
    others = ["firecrawl-mcp", "unrelated-server", "totally-different"]
    neighbors = r.near_neighbor_ids("firecrawl-mcp2", others, max_distance=2)
    assert "firecrawl-mcp" in neighbors


def test_near_neighbor_ids_excludes_self():
    others = ["filesystem", "other"]
    assert "filesystem" not in r.near_neighbor_ids("filesystem", others)


def test_near_neighbor_ids_excludes_far_matches():
    others = ["completely-unrelated-name"]
    assert r.near_neighbor_ids("filesystem", others, max_distance=2) == []


def test_extract_package_refs_npm():
    entry = _entry(config={"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]})
    refs = r.extract_package_refs(entry)
    assert len(refs) == 1
    assert refs[0].ecosystem == "npm"
    assert refs[0].name == "@scope/pkg"
    assert refs[0].version == "1.2.3"


def test_extract_package_refs_uvx():
    entry = _entry(config={"command": "uvx", "args": ["some-pypi-pkg==1.0.0"]})
    refs = r.extract_package_refs(entry)
    assert len(refs) == 1
    assert refs[0].ecosystem == "pypi"
    assert refs[0].name == "some-pypi-pkg"
    assert refs[0].version == "1.0.0"


def test_extract_package_refs_link_only_entry_returns_empty():
    entry = {"id": "slack", "setup": "link-only", "docs_url": "https://example.com"}
    assert r.extract_package_refs(entry) == []


def test_lookup_registry_info_fails_soft_on_none():
    ref = r.PackageRef("x", "npm", "somepkg", "1.0.0")
    info = r.lookup_registry_info(ref, lambda _ref: None, [])
    assert info.available is False


def test_lookup_registry_info_fails_soft_on_exception():
    def boom(_ref):
        raise RuntimeError("network down")

    ref = r.PackageRef("x", "npm", "somepkg", "1.0.0")
    info = r.lookup_registry_info(ref, boom, [])
    assert info.available is False


def test_lookup_registry_info_populates_fields_when_available():
    ref = r.PackageRef("x", "npm", "somepkg", "1.0.0")

    def fetcher(_ref):
        return {
            "publisher": "hello_sideguide",
            "age_days": 30,
            "last_release": "2026-01-01",
            "downloads": 500,
        }

    info = r.lookup_registry_info(ref, fetcher, [])
    assert info.available is True
    assert info.publisher == "hello_sideguide"
    assert info.downloads == 500


def test_lookup_registry_info_includes_near_neighbors():
    ref = r.PackageRef("x", "npm", "firecrawl-mcp2", "1.0.0")
    info = r.lookup_registry_info(ref, lambda _ref: None, ["firecrawl-mcp"])
    assert "firecrawl-mcp" in info.near_neighbor_ids


# --------------------------------------------------------------------------- #
# contains_non_ascii
# --------------------------------------------------------------------------- #
def test_contains_non_ascii_true():
    assert r.contains_non_ascii("pаckage") is True


def test_contains_non_ascii_false():
    assert r.contains_non_ascii("package") is False