Add breakpoints (B1-B6), policy engine, and framework mappings

Operationalises the six control breakpoints from Table 6 of the paper
plus the YAML policy engine and the framework-clause mappings that turn
an attestation into an auditable artefact.

- breakpoints: b1_change, b2_thirdparty, b3_dlp, b4_privilege, b5_audit,
  b6_capability_state. Each returns a CheckResult with severity
  pass/info/warn/block, structured evidence, and paper-aligned titles
- policy.engine: Policy (YAML) + PolicyReport; Policy.load() reads the
  document; evaluate() runs all six checks with the policy context
- policies/: default.yaml (balanced), developer.yaml (TCS <= 20),
  restricted.yaml (analyst, TCS <= 10 and execute requires approval)
- mappings/: ISO 42001 Annex A, NIST AI 600-1, EU AI Act YAMLs mapping
  each B-check to the concrete framework clauses it addresses
- CLI: `mcp-gov check`, `mcp-gov mappings list|show`

51/51 tests green, mypy strict clean, ruff clean, coverage 82%.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: linus131313

policies/default.yaml

@@ -0,0 +1,18 @@
+# Balanced default policy. Warns on deviations but only blocks the
+# highest-privilege mis-configurations. Intended as a starting point
+# for teams adopting the kit.
+version: 1
+name: default
+max_tcs: 30
+warn_tcs: 15
+max_third_party: 6
+third_party_allowlist:
+  - "@modelcontextprotocol/server-github"
+  - "@modelcontextprotocol/server-filesystem"
+  - "@modelcontextprotocol/server-sqlite"
+  - "@modelcontextprotocol/server-fetch"
+  - "shell-mcp"
+  - "docker-mcp"
+require_approval_for_execute: false
+require_approval_for_changes: false
+allow_exfil_paths: false

policies/developer.yaml

@@ -0,0 +1,14 @@
+# Developer role: full git/shell workflow permitted, strict cap.
+version: 1
+name: developer
+max_tcs: 20
+warn_tcs: 14
+max_third_party: 5
+third_party_allowlist:
+  - "@modelcontextprotocol/server-github"
+  - "@modelcontextprotocol/server-filesystem"
+  - "shell-mcp"
+  - "docker-mcp"
+require_approval_for_execute: false
+require_approval_for_changes: false
+allow_exfil_paths: false

policies/restricted.yaml

@@ -0,0 +1,13 @@
+# Analyst / data-science role: read and bounded write, no execute.
+version: 1
+name: restricted
+max_tcs: 10
+warn_tcs: 5
+max_third_party: 3
+third_party_allowlist:
+  - "@modelcontextprotocol/server-filesystem"
+  - "@modelcontextprotocol/server-sqlite"
+  - "@modelcontextprotocol/server-fetch"
+require_approval_for_execute: true
+require_approval_for_changes: true
+allow_exfil_paths: false

src/mcp_governance_kit/breakpoints/__init__.py

@@ -0,0 +1,36 @@
+"""Control-breakpoint checks (B1–B6 from Table 6 of the paper).
+
+Each check operationalises one control family that the paper identifies
+as silently broken under an MCP-native agent architecture:
+
+* B1 — change management (ISO 27001 A.8.32, COBIT BAI06)
+* B2 — third-party risk (ISO 27001 A.5.19, NIST SP 800-161)
+* B3 — data loss prevention (ISO 27001 A.8.12)
+* B4 — privileged access (ISO 27001 A.8.2, NIST AC-6)
+* B5 — audit and monitoring (ISO 27001 A.8.15, NIST AU family)
+* B6 — AI-specific capability state (ISO 42001 A.6, NIST AI 600-1)
+"""
+
+from __future__ import annotations
+
+from mcp_governance_kit.breakpoints.b1_change import b1_change
+from mcp_governance_kit.breakpoints.b2_thirdparty import b2_thirdparty
+from mcp_governance_kit.breakpoints.b3_dlp import b3_dlp
+from mcp_governance_kit.breakpoints.b4_privilege import b4_privilege
+from mcp_governance_kit.breakpoints.b5_audit import b5_audit
+from mcp_governance_kit.breakpoints.b6_capability_state import b6_capability_state
+from mcp_governance_kit.breakpoints.base import CheckResult, Severity
+
+ALL_CHECKS = ("B1", "B2", "B3", "B4", "B5", "B6")
+
+__all__ = [
+    "ALL_CHECKS",
+    "CheckResult",
+    "Severity",
+    "b1_change",
+    "b2_thirdparty",
+    "b3_dlp",
+    "b4_privilege",
+    "b5_audit",
+    "b6_capability_state",
+]

src/mcp_governance_kit/breakpoints/b1_change.py

@@ -0,0 +1,91 @@
+"""B1 — change management.
+
+Compares two attestations of the same host. Any added, removed, or
+reclassified tool is a change that would, under traditional change
+management, generate a change record. The check is WARN if any change
+is present and the ``changes_approved`` flag is not set; BLOCK if
+``require_approval`` is enabled in the policy context and the flag is
+not set. PASS otherwise.
+"""
+
+from __future__ import annotations
+
+from mcp_governance_kit.attest.schema import Attestation, ToolRecord
+from mcp_governance_kit.breakpoints.base import CheckResult, Severity
+
+
+def _tool_key(t: ToolRecord) -> str:
+    return f"{t.name}@{t.server.identity}"
+
+
+def b1_change(
+    previous: Attestation | None,
+    current: Attestation,
+    *,
+    changes_approved: bool = False,
+    require_approval: bool = False,
+) -> CheckResult:
+    """Compare two attestations and surface the change set.
+
+    If ``previous`` is ``None`` the check is considered INFO (first-time
+    attestation, no change to compare against).
+    """
+    if previous is None:
+        return CheckResult(
+            check_id="B1",
+            title="Change management",
+            severity=Severity.INFO,
+            summary="No prior attestation supplied; change set cannot be computed.",
+        )
+
+    prev_by_key = {_tool_key(t): t for t in previous.tools}
+    curr_by_key = {_tool_key(t): t for t in current.tools}
+
+    added = sorted(set(curr_by_key) - set(prev_by_key))
+    removed = sorted(set(prev_by_key) - set(curr_by_key))
+    reclassified: list[str] = []
+    for key in set(prev_by_key) & set(curr_by_key):
+        p, c = prev_by_key[key], curr_by_key[key]
+        if (p.reach, p.action, p.server.third_party) != (c.reach, c.action, c.server.third_party):
+            reclassified.append(key)
+    reclassified.sort()
+
+    total_changes = len(added) + len(removed) + len(reclassified)
+    tcs_delta = current.tcs.value - previous.tcs.value
+
+    if total_changes == 0:
+        return CheckResult(
+            check_id="B1",
+            title="Change management",
+            severity=Severity.PASS,
+            summary="Tool graph unchanged since previous attestation.",
+            evidence={"tcs_delta": tcs_delta},
+        )
+
+    details = (
+        [f"+ {k}" for k in added] + [f"- {k}" for k in removed] + [f"~ {k}" for k in reclassified]
+    )
+
+    if changes_approved:
+        sev = Severity.INFO
+        msg = f"{total_changes} approved change(s) since previous attestation."
+    elif require_approval:
+        sev = Severity.BLOCK
+        msg = f"{total_changes} unapproved change(s) detected. Policy requires approval for B1."
+    else:
+        sev = Severity.WARN
+        msg = f"{total_changes} change(s) detected without change record."
+
+    return CheckResult(
+        check_id="B1",
+        title="Change management",
+        severity=sev,
+        summary=msg,
+        details=details,
+        evidence={
+            "added": len(added),
+            "removed": len(removed),
+            "reclassified": len(reclassified),
+            "tcs_delta": tcs_delta,
+        },
+    )

src/mcp_governance_kit/breakpoints/b2_thirdparty.py

@@ -0,0 +1,60 @@
+"""B2 — third-party risk.
+
+Enumerates the third-party servers bound to the host and cross-checks
+them against an allow-list. Servers that are not on the list are WARN
+(INFO if an explicit exception tag is supplied); exceeding a configured
+max count is BLOCK.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from mcp_governance_kit.attest.schema import Attestation
+from mcp_governance_kit.breakpoints.base import CheckResult, Severity
+
+
+def b2_thirdparty(
+    attestation: Attestation,
+    *,
+    allowlist: Iterable[str] = (),
+    max_third_party: int | None = None,
+) -> CheckResult:
+    """Check third-party server exposure."""
+    allowed = set(allowlist)
+    third_party = [t for t in attestation.tools if t.server.third_party]
+    unapproved = sorted({t.server.identity for t in third_party} - allowed)
+
+    if max_third_party is not None and len(third_party) > max_third_party:
+        return CheckResult(
+            check_id="B2",
+            title="Third-party risk",
+            severity=Severity.BLOCK,
+            summary=(
+                f"{len(third_party)} third-party server(s), "
+                f"exceeds policy maximum of {max_third_party}."
+            ),
+            details=[t.server.identity for t in third_party],
+            evidence={"third_party_count": len(third_party)},
+        )
+
+    if unapproved:
+        return CheckResult(
+            check_id="B2",
+            title="Third-party risk",
+            severity=Severity.WARN,
+            summary=f"{len(unapproved)} third-party server(s) not on allow-list.",
+            details=unapproved,
+            evidence={
+                "unapproved_count": len(unapproved),
+                "third_party_count": len(third_party),
+            },
+        )
+
+    return CheckResult(
+        check_id="B2",
+        title="Third-party risk",
+        severity=Severity.PASS,
+        summary=f"All {len(third_party)} third-party server(s) are on the allow-list.",
+        evidence={"third_party_count": len(third_party)},
+    )

src/mcp_governance_kit/breakpoints/b3_dlp.py

@@ -0,0 +1,55 @@
+"""B3 — data loss prevention.
+
+Flags tool-graph compositions where data can move from local disk to a
+network-reaching write tool without crossing a DLP egress point. The
+simplest proxy: any tool graph that contains both a local-read/write
+tool and a network-write tool is a potential exfil path.
+"""
+
+from __future__ import annotations
+
+from mcp_governance_kit.attest.schema import Attestation
+from mcp_governance_kit.breakpoints.base import CheckResult, Severity
+from mcp_governance_kit.tcs.models import Action, Reach
+
+
+def b3_dlp(attestation: Attestation, *, allow_exfil_paths: bool = False) -> CheckResult:
+    """Detect potential data-exfiltration paths in the tool graph."""
+    local_access = [
+        t
+        for t in attestation.tools
+        if t.reach is Reach.LOCAL and t.action in (Action.READ, Action.WRITE, Action.EXECUTE)
+    ]
+    net_write = [
+        t
+        for t in attestation.tools
+        if t.reach is Reach.NETWORK and t.action in (Action.WRITE, Action.EXECUTE)
+    ]
+
+    if not (local_access and net_write):
+        return CheckResult(
+            check_id="B3",
+            title="Data loss prevention",
+            severity=Severity.PASS,
+            summary="No local-to-network exfil path in tool graph.",
+        )
+
+    paths = [
+        f"{lo.server.identity} -> {no.server.identity}" for lo in local_access for no in net_write
+    ]
+    sev = Severity.INFO if allow_exfil_paths else Severity.WARN
+    return CheckResult(
+        check_id="B3",
+        title="Data loss prevention",
+        severity=sev,
+        summary=(
+            f"{len(local_access)} local-access + {len(net_write)} network-write tool(s) "
+            f"= {len(paths)} potential exfil path(s)."
+        ),
+        details=paths[:20],
+        evidence={
+            "local_access_count": len(local_access),
+            "net_write_count": len(net_write),
+            "path_count": len(paths),
+        },
+    )

src/mcp_governance_kit/breakpoints/b4_privilege.py

@@ -0,0 +1,87 @@
+"""B4 — privileged access.
+
+Two sub-checks:
+
+1. TCS threshold. If the attestation's TCS exceeds the configured
+   threshold, the check escalates to WARN or BLOCK.
+2. Execute capability. Any tool with ``action=execute`` is flagged;
+   network+execute combinations are the strictest and go to BLOCK when
+   ``require_approval_for_execute`` is enabled.
+"""
+
+from __future__ import annotations
+
+from mcp_governance_kit.attest.schema import Attestation
+from mcp_governance_kit.breakpoints.base import CheckResult, Severity
+from mcp_governance_kit.tcs.models import Action, Reach
+
+
+def b4_privilege(
+    attestation: Attestation,
+    *,
+    max_tcs: float | None = None,
+    warn_tcs: float | None = None,
+    require_approval_for_execute: bool = False,
+    execute_approved: bool = False,
+) -> CheckResult:
+    """Evaluate privileged-access exposure."""
+    tcs_value = attestation.tcs.value
+    execute_tools = [t for t in attestation.tools if t.action is Action.EXECUTE]
+    net_execute_tools = [t for t in execute_tools if t.reach is Reach.NETWORK]
+
+    details: list[str] = []
+    evidence: dict[str, str | int | float | bool] = {
+        "tcs": tcs_value,
+        "execute_tools": len(execute_tools),
+        "network_execute_tools": len(net_execute_tools),
+    }
+
+    if max_tcs is not None and tcs_value > max_tcs:
+        details.extend(f"execute: {t.server.identity}" for t in execute_tools)
+        return CheckResult(
+            check_id="B4",
+            title="Privileged access",
+            severity=Severity.BLOCK,
+            summary=f"TCS {tcs_value} exceeds policy maximum {max_tcs}.",
+            details=details,
+            evidence=evidence,
+        )
+
+    if require_approval_for_execute and net_execute_tools and not execute_approved:
+        return CheckResult(
+            check_id="B4",
+            title="Privileged access",
+            severity=Severity.BLOCK,
+            summary=(
+                f"{len(net_execute_tools)} network-reaching execute tool(s) require approval."
+            ),
+            details=[t.server.identity for t in net_execute_tools],
+            evidence=evidence,
+        )
+
+    if warn_tcs is not None and tcs_value > warn_tcs:
+        return CheckResult(
+            check_id="B4",
+            title="Privileged access",
+            severity=Severity.WARN,
+            summary=f"TCS {tcs_value} exceeds advisory threshold {warn_tcs}.",
+            evidence=evidence,
+        )
+
+    if execute_tools:
+        return CheckResult(
+            check_id="B4",
+            title="Privileged access",
+            severity=Severity.INFO,
+            summary=f"{len(execute_tools)} execute-capable tool(s) bound; within thresholds.",
+            details=[t.server.identity for t in execute_tools],
+            evidence=evidence,
+        )
+
+    return CheckResult(
+        check_id="B4",
+        title="Privileged access",
+        severity=Severity.PASS,
+        summary=f"No execute-capable tools; TCS {tcs_value} within thresholds.",
+        evidence=evidence,
+    )