feat(security): validate offline MCP catalog approval pins

.github/workflows/ci.yml

@@ -29,6 +29,18 @@ concurrency:
   cancel-in-progress: ${{ github.event_name == 'pull_request' || github.event_name == 'pull_request_target' }}
 
 jobs:
+  mcp-catalog-offline:
+    name: MCP catalog offline regressions
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: "3.12"
+      - name: Fixture-only approval and drift regressions
+        run: python -m unittest discover -s tests/security -p test_mcp_catalog_approval.py -v
+
   guards:
     # Fail fast on the class of breakage that shipped to main un-caught:
     # committed merge-conflict markers and import-time Python SyntaxErrors.
@@ -201,3 +213,5 @@ jobs:
           PYTHONPATH=src python -m pytest tests/unit/ -v --timeout=120
           --cov=src/youtube_extension --cov-fail-under=0
           --ignore=tests/unit/test_transcript_action_workflow.py -k "not integration"
+      - name: Offline MCP inventory and approval regressions
+        run: python -m pytest tests/security/test_local_mcp_config.py tests/security/test_mcp_catalog_approval.py -q -o addopts=''

artifacts/security/local-mcp-audit-stage-one.md

@@ -22,7 +22,9 @@ server names, configuration filenames or environment variable names.
 
 The catalog helper hashes the complete recorded tool list plus server identity;
 ordering of tools and object keys is normalized. Its output is evidence only,
-not approval. Approval binding and invalidation integration remain unimplemented.
+not approval. The separate offline recorded-catalog comparison is documented in
+`local-mcp-catalog-approval.md`; launcher/runtime approval integration remains
+unimplemented.
 
 Remaining #2349 work: official publisher/version resolution, credential argv
 migration, sandbox/filesystem/egress policy, complete container inventory, explicit

artifacts/security/local-mcp-catalog-approval.md

@@ -0,0 +1,70 @@
+# Offline recorded MCP catalog approval
+
+This is a bounded follow-up to #2349 / GRV-780. The launcher inventory remains
+fail-closed and its catalog field remains null without trustworthy capture and
+configuration binding. This comparator validates recorded evidence only. It
+does not authorize a launcher, prove publisher identity, capture a live server,
+or enforce active runtime configuration. The parent issue must remain open.
+
+## Inputs and review boundary
+
+A capture is strict JSON with exactly `identity` (a nonempty string) and
+`response` (a complete recorded `tools/list` result containing `tools`). A
+non-null `nextCursor` is rejected: callers must supply a complete, separately
+assembled capture. Duplicate tool names, malformed fields, duplicate JSON keys
+and nonfinite numbers fail closed. No MCP connection, subprocess, credential
+lookup or environment expansion exists in the validator.
+
+```sh
+python scripts/security/validate_mcp_catalog.py --capture /tmp/capture.json
+```
+
+Exit 1 and `REVIEW_REQUIRED` are expected without an approval. The `observed`
+object is a candidate pin with `approved: false`. A reviewer must inspect the
+original identity and complete tools (including descriptions, schemas,
+annotations and other metadata), separately store that object, and explicitly
+set `approved: true` only after approval. There is no auto-approve command.
+Pins are trusted reviewed files, not signatures or attestations. Protect them
+with repository review/ownership policy; a party able to rewrite both the pin
+and capture can bypass this comparison. Supplied identity is bound by hashing,
+but independently establishing that identity and its active configuration is
+still outstanding parent work. Do not commit actual active-server approvals
+based on synthetic fixtures.
+
+```sh
+python scripts/security/validate_mcp_catalog.py --capture /tmp/capture.json --approval /tmp/reviewed-pin.json
+python -m unittest discover -s tests/security -p test_mcp_catalog_approval.py -v
+python -m pytest tests/security/test_local_mcp_config.py tests/security/test_mcp_catalog_approval.py -q -o addopts=''
+```
+
+Only an exact approved match exits 0 with `MATCH`. This means recorded-contract
+equality, not permission to execute. Unapproved pins, changed identity, changed
+catalogs, added/removed tools and inconsistent per-tool hashes block. Every tool
+field participates in the original identity-bound full catalog digest. Tool
+ordering and JSON object ordering are canonical; schema array ordering remains
+significant. Hash algorithm/version is the existing Python SHA-256 canonical
+JSON format, not a claim of cross-language RugSnare hash compatibility.
+
+Separate schema (input/output) and description hashes explain drift. They never
+relax the full-contract gate: description changes, hidden metadata and annotation
+changes still invalidate approval. Output contains hashes and fixed status codes
+only, never original names, identity, schemas, descriptions or exception text.
+Treat source captures as sensitive; hashing is not encrypted storage.
+
+## Reference evaluation
+
+Inspected RugSnare commit `1551e5b28c5608a95fb1ad854c21191d8deabf81`:
+`product/src/hash.js` splits structural/prose hashes, and `product/src/pins.js`
+stores reviewed tool pins and detects NEW/REMOVED/DRIFT. The useful design is
+diagnostic separation plus explicit pins. EventRelay retains its stronger
+complete-tool/identity digest and rejects all drift. No RugSnare source is copied,
+package installed, proxy adopted or runtime dependency added. Its live execution
+paths are outside this offline implementation.
+
+CI runs the dependency-free negative fixtures on every PR and main push, plus
+the existing pytest inventory regressions in the Python matrix. This is a
+regression gate for the comparator, not active-catalog enforcement: there are no
+approved real captures in this change. Remaining parent work includes provenance,
+credential migration, containment, trustworthy capture/configuration binding and
+active-configuration CI/runtime enforcement. Roll back by reverting this change;
+no production MCP configuration is modified.

scripts/security/audit_local_mcp_configs.py

@@ -14,7 +14,7 @@
 
 
 def canonical(value):
-    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False)
 
 
 def catalog_digest(identity, response):
@@ -28,6 +28,80 @@ def catalog_digest(identity, response):
     return hashlib.sha256(canonical({"identity": identity, "tools": sorted(tools, key=lambda t: t["name"])}).encode()).hexdigest()
 
 
+def digest(value):
+    return hashlib.sha256(canonical(value).encode()).hexdigest()
+
+
+def catalog_record(identity, response):
+    """Hash an offline, complete capture; evidence never grants approval."""
+    full_digest = catalog_digest(identity, response)
+    if response.get("nextCursor") is not None:
+        raise ValueError("incomplete catalog")
+    records = []
+    for tool in response["tools"]:
+        if not tool["name"] or ("description" in tool and not isinstance(tool["description"], str)):
+            raise ValueError("invalid tool")
+        for key in ("annotations", "outputSchema"):
+            if key in tool and not isinstance(tool[key], dict):
+                raise ValueError("invalid tool")
+        # Hash every field for enforcement. Split digests are diagnostics only.
+        # Omitted description differs from an explicitly empty description.
+        records.append({
+            "name_digest": digest(tool["name"]),
+            "schema_digest": digest({key: tool[key] for key in ("inputSchema", "outputSchema") if key in tool}),
+            "description_digest": digest({"description": tool["description"]} if "description" in tool else {}),
+            "tool_digest": digest(tool),
+        })
+    return {"schema_version": 1, "approved": False,
+            "identity_digest": digest(identity), "catalog_digest": full_digest,
+            "tools": sorted(records, key=lambda item: item["name_digest"])}
+
+
+def compare_catalog(identity, response, approval):
+    """Compare against a separately reviewed pin; emit hashes/codes only."""
+    observed = catalog_record(identity, response)
+    fields = {"schema_version", "approved", "identity_digest", "catalog_digest", "tools"}
+    tool_fields = {"name_digest", "schema_digest", "description_digest", "tool_digest"}
+    def valid_hash(value):
+        return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value) is not None
+    if (not isinstance(approval, dict) or set(approval) != fields
+            or type(approval["schema_version"]) is not int or approval["schema_version"] != 1
+            or type(approval["approved"]) is not bool
+            or not all(valid_hash(approval[key]) for key in ("identity_digest", "catalog_digest"))
+            or not isinstance(approval["tools"], list)
+            or any(not isinstance(item, dict) or set(item) != tool_fields
+                   or not all(valid_hash(value) for value in item.values()) for item in approval["tools"])):
+        raise ValueError("invalid approval")
+    old = {item["name_digest"]: item for item in approval["tools"]}
+    if len(old) != len(approval["tools"]):
+        raise ValueError("duplicate approval tool")
+    new = {item["name_digest"]: item for item in observed["tools"]}
+    findings = []
+    if approval["approved"] is not True:
+        findings.append("UNAPPROVED_CATALOG")
+    if approval["identity_digest"] != observed["identity_digest"]:
+        findings.append("SERVER_IDENTITY_CHANGED")
+    if approval["catalog_digest"] != observed["catalog_digest"]:
+        findings.append("CATALOG_CHANGED")
+    changes = []
+    for name in sorted(old.keys() | new.keys()):
+        codes = []
+        if name not in old:
+            codes.append("TOOL_ADDED")
+        elif name not in new:
+            codes.append("TOOL_REMOVED")
+        else:
+            for field, code in (("schema_digest", "SCHEMA_CHANGED"),
+                                ("description_digest", "DESCRIPTION_CHANGED"),
+                                ("tool_digest", "TOOL_CONTRACT_CHANGED")):
+                if old[name][field] != new[name][field]:
+                    codes.append(code)
+        if codes:
+            changes.append({"name_digest": name, "codes": codes})
+    return {"schema_version": 1, "status": "BLOCKED" if findings or changes else "MATCH",
+            "findings": findings, "changes": changes, "observed": observed}
+
+
 def audit(config, label):
     if not isinstance(config, dict) or not isinstance(config.get("mcpServers"), dict):
         raise ValueError("invalid config")

scripts/security/validate_mcp_catalog.py

@@ -0,0 +1,28 @@
+"""Compare recorded tools/list evidence with a reviewed pin, entirely offline."""
+import argparse
+
+from audit_local_mcp_configs import canonical, catalog_record, compare_catalog, load
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--capture", required=True, help="JSON: identity and response")
+    parser.add_argument("--approval", help="Separately reviewed JSON pin")
+    options = parser.parse_args()
+    try:
+        capture = load(options.capture)
+        if not isinstance(capture, dict) or set(capture) != {"identity", "response"}:
+            raise ValueError("invalid capture")
+        if options.approval:
+            result = compare_catalog(capture["identity"], capture["response"], load(options.approval))
+        else:
+            result = {"status": "REVIEW_REQUIRED", "observed": catalog_record(capture["identity"], capture["response"])}
+    except (OSError, ValueError, TypeError, RecursionError):
+        # Do not echo exception text, paths, identity, schemas or descriptions.
+        result = {"status": "BLOCKED", "findings": ["INVALID_CATALOG_OR_APPROVAL"]}
+    print(canonical(result))
+    return 0 if result["status"] == "MATCH" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())