feat(catalog): AGT MCPSecurityScanner for catalog drift and tool poisoning

src/cmcp_gateway/catalog/scanner.py

@@ -0,0 +1,180 @@
+"""
+Tool catalog security scanning via AGT MCPSecurityScanner — implements issue #58.
+
+AGT's MCPSecurityScanner provides:
+  - SHA-256 tool fingerprinting (detects definition mutation / rug-pull P4.2)
+  - Typosquatting detection (P4.1)
+  - Hidden instruction scanning in tool descriptions (P2.1)
+  - Tool drift detection via check_rug_pull()
+
+This module wires the scanner into cMCP's catalog load and drift detection flow.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+from cmcp_gateway.catalog.loader import CatalogEntry, ToolCatalog
+
+logger = logging.getLogger(__name__)
+
+try:
+    from agent_os.mcp_security import MCPSecurityScanner, MCPThreatType, MCPSeverity
+    _AGT_AVAILABLE = True
+except ImportError:
+    _AGT_AVAILABLE = False
+
+
+@dataclass
+class CatalogScanResult:
+    """Result of scanning the full tool catalog at load time."""
+
+    safe: bool
+    tools_scanned: int
+    tools_flagged: int
+    threats: list[dict[str, str]]  # [{tool_name, threat_type, severity, description}]
+
+
+@dataclass
+class DriftResult:
+    """Result of a rug-pull / drift check on a single tool."""
+
+    tool_name: str
+    drifted: bool
+    threats: list[dict[str, str]]
+
+
+class CatalogScanner:
+    """
+    Wraps AGT MCPSecurityScanner for catalog-level security checks.
+
+    Used in two contexts:
+    1. At catalog load time: scan all tools for typosquatting and hidden instructions
+    2. At runtime: check_drift() called when notifications/tools/list_changed received
+       from upstream (P4.2 rug-pull detection)
+
+    Falls back to a no-op scanner if agent-os-kernel is not installed.
+    """
+
+    def __init__(self) -> None:
+        if _AGT_AVAILABLE:
+            try:
+                self._scanner: Any = MCPSecurityScanner()
+                self._available = True
+                logger.info("CatalogScanner: AGT MCPSecurityScanner active")
+            except Exception as exc:
+                logger.warning("CatalogScanner: AGT init failed (%s) — running without security scan", exc)
+                self._scanner = None
+                self._available = False
+        else:
+            self._scanner = None
+            self._available = False
+            logger.info("CatalogScanner: agent-os-kernel not installed — no catalog scanning")
+
+    def scan_catalog(self, catalog: ToolCatalog) -> CatalogScanResult:
+        """
+        Scan all catalog entries at load time.
+
+        Registers every tool with the AGT scanner (for future drift detection)
+        and scans for:
+          - Hidden instructions in tool descriptions (P2.1 tool poisoning)
+          - Typosquatting / look-alike tool names (P4.1)
+        """
+        if not self._available or self._scanner is None:
+            return CatalogScanResult(
+                safe=True,
+                tools_scanned=len(catalog.entries),
+                tools_flagged=0,
+                threats=[],
+            )
+
+        all_threats: list[dict[str, str]] = []
+        tools_flagged = 0
+
+        for tool_name, entry in catalog.entries.items():
+            server_name = entry.server.display_name or entry.server.url
+            description = entry.approved_definition.description
+
+            try:
+                # Register the tool fingerprint (enables future drift detection)
+                self._scanner.register_tool(
+                    tool_name=tool_name,
+                    server_name=server_name,
+                    description=description,
+                    schema=entry.approved_definition.input_schema or {},
+                )
+
+                # Scan for threats
+                threats = self._scanner.scan_tool(
+                    tool_name=tool_name,
+                    description=description,
+                    server_name=server_name,
+                )
+
+                if threats:
+                    tools_flagged += 1
+                    for threat in threats:
+                        threat_type = threat.threat_type.value if hasattr(threat.threat_type, "value") else str(threat.threat_type)
+                        severity = threat.severity.value if hasattr(threat.severity, "value") else str(getattr(threat, "severity", "unknown"))
+                        all_threats.append({
+                            "tool_name": tool_name,
+                            "threat_type": threat_type,
+                            "severity": severity,
+                            "description": str(getattr(threat, "description", "")),
+                        })
+                        logger.warning(
+                            "CATALOG_THREAT: tool=%s type=%s severity=%s",
+                            tool_name, threat_type, severity,
+                        )
+            except Exception as exc:
+                logger.debug("CatalogScanner: scan_tool failed for %s: %s", tool_name, exc)
+
+        return CatalogScanResult(
+            safe=(tools_flagged == 0),
+            tools_scanned=len(catalog.entries),
+            tools_flagged=tools_flagged,
+            threats=all_threats,
+        )
+
+    def check_drift(
+        self,
+        tool_name: str,
+        server_name: str,
+        current_definition: dict[str, Any],
+    ) -> DriftResult:
+        """
+        Check if a tool's definition has drifted from the registered fingerprint.
+
+        Called when notifications/tools/list_changed is received from an upstream
+        server (P4.2 rug-pull detection). Returns DriftResult with drifted=True
+        if the definition has changed since the catalog was sealed.
+        """
+        if not self._available or self._scanner is None:
+            return DriftResult(tool_name=tool_name, drifted=False, threats=[])
+
+        try:
+            threats = self._scanner.check_rug_pull(
+                tool_name=tool_name,
+                server_name=server_name,
+                current_definition=current_definition,
+            )
+            if threats:
+                threat_list = [
+                    {
+                        "tool_name": tool_name,
+                        "threat_type": t.threat_type.value if hasattr(t.threat_type, "value") else str(t.threat_type),
+                        "description": str(getattr(t, "description", "")),
+                    }
+                    for t in threats
+                ]
+                logger.error(
+                    "CATALOG_DRIFT_DETECTED: tool=%s threats=%d",
+                    tool_name, len(threats),
+                )
+                return DriftResult(tool_name=tool_name, drifted=True, threats=threat_list)
+        except Exception as exc:
+            logger.debug("CatalogScanner: check_rug_pull failed for %s: %s", tool_name, exc)
+
+        return DriftResult(tool_name=tool_name, drifted=False, threats=[])

tests/unit/test_catalog_scanner.py

@@ -0,0 +1,152 @@
+"""Tests for CatalogScanner / AGT MCPSecurityScanner integration (issue #58)."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from cmcp_gateway.catalog.loader import (
+    ApprovedDefinition, CatalogEntry, ServerIdentity, ToolCatalog,
+)
+from cmcp_gateway.catalog.scanner import CatalogScanner, CatalogScanResult, DriftResult
+
+
+def _make_entry(tool_name: str, description: str = "test tool") -> CatalogEntry:
+    return CatalogEntry(
+        tool_name=tool_name,
+        server=ServerIdentity(
+            display_name="Test Server",
+            url="https://test.example.com/mcp",
+            tls_fingerprint="SHA256:AAAA==",
+            spiffe_id=None,
+            transport="http-sse",
+            rotation_mode="key-pinned",
+        ),
+        approved_definition=ApprovedDefinition(
+            description=description,
+            input_schema={},
+            output_schema=None,
+        ),
+        definition_hash="sha256:" + "0" * 64,
+        compliance_domain="external",
+        requires_baa=False,
+        sensitivity_level="public",
+        added_at="2026-06-05T00:00:00Z",
+        approved_by="test",
+    )
+
+
+def _make_catalog(*tools: str) -> ToolCatalog:
+    entries = {t: _make_entry(t) for t in (tools or ("test.tool",))}
+    return ToolCatalog(entries=entries, catalog_hash="sha256:" + "1" * 64)
+
+
+# ── When AGT is available ─────────────────────────────────────────────────────
+
+def test_scan_catalog_safe_returns_clean_result():
+    with patch("cmcp_gateway.catalog.scanner._AGT_AVAILABLE", True):
+        with patch("cmcp_gateway.catalog.scanner.MCPSecurityScanner") as MockScanner:
+            mock_instance = MagicMock()
+            mock_instance.scan_tool.return_value = []  # no threats
+            mock_instance.register_tool.return_value = MagicMock()
+            MockScanner.return_value = mock_instance
+
+            scanner = CatalogScanner()
+            result = scanner.scan_catalog(_make_catalog("crm.query", "hr.lookup"))
+
+    assert isinstance(result, CatalogScanResult)
+    assert result.safe is True
+    assert result.tools_scanned == 2
+    assert result.tools_flagged == 0
+    assert result.threats == []
+
+
+def test_scan_catalog_flags_threat():
+    mock_threat = MagicMock()
+    mock_threat.threat_type.value = "tool_poisoning"
+    mock_threat.severity.value = "high"
+    mock_threat.description = "hidden instruction in description"
+
+    with patch("cmcp_gateway.catalog.scanner._AGT_AVAILABLE", True):
+        with patch("cmcp_gateway.catalog.scanner.MCPSecurityScanner") as MockScanner:
+            mock_instance = MagicMock()
+            mock_instance.scan_tool.return_value = [mock_threat]
+            mock_instance.register_tool.return_value = MagicMock()
+            MockScanner.return_value = mock_instance
+
+            scanner = CatalogScanner()
+            result = scanner.scan_catalog(_make_catalog("malicious.tool"))
+
+    assert result.safe is False
+    assert result.tools_flagged == 1
+    assert len(result.threats) == 1
+    assert result.threats[0]["tool_name"] == "malicious.tool"
+    assert result.threats[0]["threat_type"] == "tool_poisoning"
+
+
+def test_scan_catalog_registers_all_tools():
+    with patch("cmcp_gateway.catalog.scanner._AGT_AVAILABLE", True):
+        with patch("cmcp_gateway.catalog.scanner.MCPSecurityScanner") as MockScanner:
+            mock_instance = MagicMock()
+            mock_instance.scan_tool.return_value = []
+            mock_instance.register_tool.return_value = MagicMock()
+            MockScanner.return_value = mock_instance
+
+            scanner = CatalogScanner()
+            scanner.scan_catalog(_make_catalog("tool.a", "tool.b", "tool.c"))
+
+    assert mock_instance.register_tool.call_count == 3
+
+
+def test_check_drift_returns_clean_when_no_changes():
+    with patch("cmcp_gateway.catalog.scanner._AGT_AVAILABLE", True):
+        with patch("cmcp_gateway.catalog.scanner.MCPSecurityScanner") as MockScanner:
+            mock_instance = MagicMock()
+            mock_instance.check_rug_pull.return_value = []
+            MockScanner.return_value = mock_instance
+
+            scanner = CatalogScanner()
+            result = scanner.check_drift("crm.query", "CRM Server", {"description": "same"})
+
+    assert isinstance(result, DriftResult)
+    assert result.drifted is False
+    assert result.threats == []
+
+
+def test_check_drift_detects_rug_pull():
+    mock_threat = MagicMock()
+    mock_threat.threat_type.value = "rug_pull"
+    mock_threat.description = "description changed after approval"
+
+    with patch("cmcp_gateway.catalog.scanner._AGT_AVAILABLE", True):
+        with patch("cmcp_gateway.catalog.scanner.MCPSecurityScanner") as MockScanner:
+            mock_instance = MagicMock()
+            mock_instance.check_rug_pull.return_value = [mock_threat]
+            MockScanner.return_value = mock_instance
+
+            scanner = CatalogScanner()
+            result = scanner.check_drift("crm.query", "CRM Server", {"description": "changed description with injected instruction"})
+
+    assert result.drifted is True
+    assert result.threats[0]["threat_type"] == "rug_pull"
+
+
+# ── When AGT is not available (graceful fallback) ─────────────────────────────
+
+def test_scan_catalog_safe_without_agt():
+    with patch("cmcp_gateway.catalog.scanner._AGT_AVAILABLE", False):
+        scanner = CatalogScanner()
+        result = scanner.scan_catalog(_make_catalog("crm.query"))
+
+    assert result.safe is True
+    assert result.tools_scanned == 1
+    assert result.threats == []
+
+
+def test_check_drift_returns_clean_without_agt():
+    with patch("cmcp_gateway.catalog.scanner._AGT_AVAILABLE", False):
+        scanner = CatalogScanner()
+        result = scanner.check_drift("crm.query", "CRM", {})
+
+    assert result.drifted is False