feat(mcp): add isolated protocol replay verifier
This commit is contained in:
ogt
372
scripts/security/tests/test_external_mcp_replay_controller.py
Normal file
372
scripts/security/tests/test_external_mcp_replay_controller.py
Normal file
@@ -0,0 +1,372 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for the bounded external MCP replay controller without mocks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import sys
|
||||
import tarfile
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
CONTROLLER_PATH = ROOT / "scripts/security/external_mcp_replay_controller.py"
|
||||
POLICY_PATH = ROOT / "config/mcp/playwright-mcp-replay-policy.json"
|
||||
ARTIFACT_POLICY_PATH = ROOT / "config/mcp/playwright-mcp-artifact-policy.json"
|
||||
WORKFLOW_PATH = ROOT / ".gitea/workflows/mcp-external-replay.yaml"
|
||||
|
||||
SPEC = importlib.util.spec_from_file_location("external_mcp_replay_controller", CONTROLLER_PATH)
|
||||
assert SPEC and SPEC.loader
|
||||
controller = importlib.util.module_from_spec(SPEC)
|
||||
sys.modules[SPEC.name] = controller
|
||||
SPEC.loader.exec_module(controller)
|
||||
|
||||
|
||||
def _canonical(payload: object) -> bytes:
|
||||
return json.dumps(
|
||||
payload,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def _write_json(path: Path, payload: object) -> bytes:
|
||||
raw = _canonical(payload) + b"\n"
|
||||
path.write_bytes(raw)
|
||||
return raw
|
||||
|
||||
|
||||
def _policy_payload() -> dict:
|
||||
return json.loads(POLICY_PATH.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def _artifact_receipts(policy_payload: dict) -> tuple[dict, dict]:
|
||||
artifact = policy_payload["artifact_source"]
|
||||
artifact_run_id = "11111111-2222-4333-8444-555555555555"
|
||||
artifact_trace_id = f"mcp-artifact-{artifact_run_id}"
|
||||
packages = []
|
||||
for key, digest in artifact["package_integrities"].items():
|
||||
name, version = key.rsplit("@", 1)
|
||||
packages.append(
|
||||
{
|
||||
"name": name,
|
||||
"version": version,
|
||||
"tarball_sha512_hex": digest,
|
||||
}
|
||||
)
|
||||
artifact_controller = {
|
||||
"schema_version": "awoooi_external_mcp_artifact_receipt_v1",
|
||||
"status": "completed_internal_mirror_pending_independent_verifier",
|
||||
"run_id": artifact_run_id,
|
||||
"trace_id": artifact_trace_id,
|
||||
"promotion_allowed": False,
|
||||
"replay_allowed": False,
|
||||
"shadow_allowed": False,
|
||||
"canary_allowed": False,
|
||||
"internal_mirror_ref": artifact["artifact_ref"],
|
||||
"runtime_mirror_ref": artifact["artifact_runtime_ref"],
|
||||
"policy_checksum": artifact["policy_checksum"],
|
||||
"bundle_manifest_sha256": artifact["bundle_manifest_sha256"],
|
||||
"cyclonedx_sbom_sha256": artifact["cyclonedx_sbom_sha256"],
|
||||
"packages": packages,
|
||||
}
|
||||
artifact_verifier = {
|
||||
"schema_version": "awoooi_external_mcp_artifact_verifier_receipt_v1",
|
||||
"status": "completed_internal_mirror_verified",
|
||||
"run_id": artifact_run_id,
|
||||
"trace_id": artifact_trace_id,
|
||||
"promotion_allowed": False,
|
||||
"replay_allowed": False,
|
||||
"shadow_allowed": False,
|
||||
"canary_allowed": False,
|
||||
"internal_mirror_ref": artifact["artifact_ref"],
|
||||
"runtime_mirror_ref": artifact["artifact_runtime_ref"],
|
||||
"policy_checksum": artifact["policy_checksum"],
|
||||
"independent_post_verifier": {"status": "passed"},
|
||||
}
|
||||
return artifact_controller, artifact_verifier
|
||||
|
||||
|
||||
def _source_fixture(directory: Path) -> tuple[Path, Path, Path]:
|
||||
payload = _policy_payload()
|
||||
artifact_controller, artifact_verifier = _artifact_receipts(payload)
|
||||
controller_path = directory / "artifact-controller.json"
|
||||
verifier_path = directory / "artifact-verifier.json"
|
||||
controller_raw = _write_json(controller_path, artifact_controller)
|
||||
artifact_verifier["controller_receipt_sha256"] = hashlib.sha256(
|
||||
controller_raw
|
||||
).hexdigest()
|
||||
verifier_raw = _write_json(verifier_path, artifact_verifier)
|
||||
payload["artifact_source"]["controller_receipt_sha256"] = hashlib.sha256(
|
||||
controller_raw
|
||||
).hexdigest()
|
||||
payload["artifact_source"]["verifier_receipt_sha256"] = hashlib.sha256(
|
||||
verifier_raw
|
||||
).hexdigest()
|
||||
policy_path = directory / "replay-policy.json"
|
||||
_write_json(policy_path, payload)
|
||||
return policy_path, controller_path, verifier_path
|
||||
|
||||
|
||||
class ReplayPolicyTests(unittest.TestCase):
|
||||
def _mutated_policy(self, mutate) -> Path:
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
path = Path(self.temp.name) / "policy.json"
|
||||
payload = _policy_payload()
|
||||
mutate(payload)
|
||||
_write_json(path, payload)
|
||||
return path
|
||||
|
||||
def tearDown(self) -> None:
|
||||
if hasattr(self, "temp"):
|
||||
self.temp.cleanup()
|
||||
|
||||
def test_committed_policy_is_exact_and_minimal(self) -> None:
|
||||
policy = controller.load_policy(POLICY_PATH)
|
||||
self.assertEqual(policy.work_item_id, "MCP-REPLAY-PLAYWRIGHT-001")
|
||||
self.assertEqual(policy.risk_class, "high")
|
||||
self.assertEqual(policy.tool_count, 24)
|
||||
self.assertEqual(len(policy.adapter_allowed_tools), 4)
|
||||
self.assertEqual(len(policy.adapter_denied_tools), 20)
|
||||
self.assertRegex(policy.artifact_ref, r"@sha256:[0-9a-f]{64}$")
|
||||
self.assertRegex(policy.runtime_image_ref, r"@sha256:[0-9a-f]{64}$")
|
||||
|
||||
def test_expired_policy_fails_closed(self) -> None:
|
||||
path = self._mutated_policy(
|
||||
lambda payload: payload.__setitem__(
|
||||
"expires_at", "2026-07-01T00:00:00+08:00"
|
||||
)
|
||||
)
|
||||
with self.assertRaisesRegex(controller.ReplayError, "replay_policy_expired"):
|
||||
controller.load_policy(path)
|
||||
|
||||
def test_network_must_remain_none(self) -> None:
|
||||
path = self._mutated_policy(
|
||||
lambda payload: payload["runtime"].__setitem__("network_mode", "bridge")
|
||||
)
|
||||
with self.assertRaisesRegex(controller.ReplayError, "runtime_isolation_invalid"):
|
||||
controller.load_policy(path)
|
||||
|
||||
def test_runtime_must_remain_non_root(self) -> None:
|
||||
path = self._mutated_policy(
|
||||
lambda payload: payload["runtime"].__setitem__("user", "0:0")
|
||||
)
|
||||
with self.assertRaisesRegex(controller.ReplayError, "runtime_user_invalid"):
|
||||
controller.load_policy(path)
|
||||
|
||||
def test_floating_runtime_reference_is_rejected(self) -> None:
|
||||
path = self._mutated_policy(
|
||||
lambda payload: payload["runtime"].__setitem__(
|
||||
"image_ref", "registry.wooo.work/awoooi/ci-runner:current"
|
||||
)
|
||||
)
|
||||
with self.assertRaisesRegex(controller.ReplayError, "runtime_image_ref_invalid"):
|
||||
controller.load_policy(path)
|
||||
|
||||
def test_direct_registration_cannot_be_enabled(self) -> None:
|
||||
path = self._mutated_policy(
|
||||
lambda payload: payload["execution_boundaries"].__setitem__(
|
||||
"direct_gateway_registration_allowed", True
|
||||
)
|
||||
)
|
||||
with self.assertRaisesRegex(controller.ReplayError, "execution_boundary_invalid"):
|
||||
controller.load_policy(path)
|
||||
|
||||
def test_dangerous_tool_cannot_leave_denied_partition(self) -> None:
|
||||
def mutate(payload: dict) -> None:
|
||||
payload["protocol_contract"]["adapter_denied_tools"].remove(
|
||||
"browser_run_code_unsafe"
|
||||
)
|
||||
|
||||
path = self._mutated_policy(mutate)
|
||||
with self.assertRaises(controller.ReplayError):
|
||||
controller.load_policy(path)
|
||||
|
||||
|
||||
class SourceEvidenceTests(unittest.TestCase):
|
||||
def test_verified_source_receipt_chain_passes(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
policy_path, controller_path, verifier_path = _source_fixture(Path(temp))
|
||||
policy = controller.load_policy(policy_path)
|
||||
evidence = controller.validate_source_evidence(
|
||||
policy,
|
||||
artifact_policy_path=ARTIFACT_POLICY_PATH,
|
||||
controller_receipt_path=controller_path,
|
||||
verifier_receipt_path=verifier_path,
|
||||
)
|
||||
self.assertEqual(
|
||||
evidence.artifact_run_id, "11111111-2222-4333-8444-555555555555"
|
||||
)
|
||||
self.assertRegex(evidence.artifact_verifier_receipt_sha256, r"^[0-9a-f]{64}$")
|
||||
|
||||
def test_tampered_controller_receipt_is_rejected(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
policy_path, controller_path, verifier_path = _source_fixture(Path(temp))
|
||||
policy = controller.load_policy(policy_path)
|
||||
controller_path.write_bytes(controller_path.read_bytes() + b" ")
|
||||
with self.assertRaisesRegex(
|
||||
controller.ReplayError,
|
||||
"artifact_controller_receipt_checksum_mismatch",
|
||||
):
|
||||
controller.validate_source_evidence(
|
||||
policy,
|
||||
artifact_policy_path=ARTIFACT_POLICY_PATH,
|
||||
controller_receipt_path=controller_path,
|
||||
verifier_receipt_path=verifier_path,
|
||||
)
|
||||
|
||||
def test_artifact_policy_checksum_drift_is_rejected(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
policy_path, controller_path, verifier_path = _source_fixture(root)
|
||||
artifact_policy = json.loads(ARTIFACT_POLICY_PATH.read_text())
|
||||
artifact_policy["risk_level"] = "medium"
|
||||
drift_path = root / "artifact-policy.json"
|
||||
_write_json(drift_path, artifact_policy)
|
||||
with self.assertRaisesRegex(
|
||||
controller.ReplayError, "artifact_policy_checksum_mismatch"
|
||||
):
|
||||
controller.validate_source_evidence(
|
||||
controller.load_policy(policy_path),
|
||||
artifact_policy_path=drift_path,
|
||||
controller_receipt_path=controller_path,
|
||||
verifier_receipt_path=verifier_path,
|
||||
)
|
||||
|
||||
|
||||
class ArchiveAndProtocolTests(unittest.TestCase):
|
||||
def test_archive_traversal_is_rejected(self) -> None:
|
||||
with self.assertRaisesRegex(controller.ReplayError, "artifact_archive_path_invalid"):
|
||||
controller._safe_relative_member("package/../../escape")
|
||||
|
||||
def test_archive_symlink_is_rejected(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
tarball = root / "bad.tgz"
|
||||
with tarfile.open(tarball, "w:gz") as archive:
|
||||
member = tarfile.TarInfo("package/link")
|
||||
member.type = tarfile.SYMTYPE
|
||||
member.linkname = "/etc/passwd"
|
||||
archive.addfile(member)
|
||||
with self.assertRaisesRegex(
|
||||
controller.ReplayError, "artifact_archive_member_type_invalid"
|
||||
):
|
||||
controller._extract_package(tarball, root / "out")
|
||||
|
||||
def test_regular_archive_extracts_with_exact_size(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
tarball = root / "good.tgz"
|
||||
content = b'{"name":"fixture"}'
|
||||
with tarfile.open(tarball, "w:gz") as archive:
|
||||
member = tarfile.TarInfo("package/package.json")
|
||||
member.size = len(content)
|
||||
archive.addfile(member, io.BytesIO(content))
|
||||
files, size = controller._extract_package(tarball, root / "out")
|
||||
self.assertEqual((files, size), (1, len(content)))
|
||||
self.assertEqual((root / "out/package.json").read_bytes(), content)
|
||||
|
||||
def test_tool_hashes_are_order_independent(self) -> None:
|
||||
tools = [
|
||||
{
|
||||
"name": "beta",
|
||||
"inputSchema": {"type": "object"},
|
||||
"annotations": {"readOnlyHint": True},
|
||||
},
|
||||
{"name": "alpha", "inputSchema": {"type": "object"}},
|
||||
]
|
||||
first = controller._tool_hashes(tools)
|
||||
second = controller._tool_hashes(list(reversed(tools)))
|
||||
self.assertEqual(first, second)
|
||||
self.assertEqual(first[0], 2)
|
||||
self.assertEqual(first[3], ("alpha", "beta"))
|
||||
|
||||
def test_duplicate_tool_names_are_rejected(self) -> None:
|
||||
tools = [
|
||||
{"name": "same", "inputSchema": {}},
|
||||
{"name": "same", "inputSchema": {}},
|
||||
]
|
||||
with self.assertRaisesRegex(controller.ReplayError, "mcp_tool_names_not_unique"):
|
||||
controller._tool_hashes(tools)
|
||||
|
||||
|
||||
class ReceiptAndWorkflowTests(unittest.TestCase):
|
||||
def test_success_receipt_preserves_no_browser_no_write_boundary(self) -> None:
|
||||
policy = controller.load_policy(POLICY_PATH)
|
||||
source = controller.SourceEvidence("a", "b", "c", "d")
|
||||
observation = controller.ReplayObservation(
|
||||
policy.server_name,
|
||||
policy.server_version,
|
||||
policy.protocol_version,
|
||||
policy.tool_count,
|
||||
policy.tool_name_set_sha256,
|
||||
policy.tool_schema_manifest_sha256,
|
||||
0,
|
||||
"stdin_eof_clean_exit",
|
||||
True,
|
||||
)
|
||||
receipt = controller.build_receipt(
|
||||
policy=policy,
|
||||
source=source,
|
||||
run_id="11111111-2222-4333-8444-555555555555",
|
||||
trace_id="mcp-replay-11111111-2222-4333-8444-555555555555",
|
||||
mode="apply",
|
||||
observation=observation,
|
||||
package_file_count=10,
|
||||
package_unpacked_bytes=20,
|
||||
)
|
||||
self.assertEqual(
|
||||
receipt["status"],
|
||||
"completed_protocol_schema_replay_pending_independent_verifier",
|
||||
)
|
||||
for field in (
|
||||
"browser_started",
|
||||
"tool_call_performed",
|
||||
"external_rag_write_performed",
|
||||
"production_write_performed",
|
||||
"promotion_allowed",
|
||||
):
|
||||
self.assertIs(receipt[field], False)
|
||||
|
||||
def test_failure_receipt_keeps_candidate_disabled(self) -> None:
|
||||
policy = controller.load_policy(POLICY_PATH)
|
||||
receipt = controller._failure_receipt(
|
||||
policy=policy,
|
||||
run_id="11111111-2222-4333-8444-555555555555",
|
||||
trace_id="mcp-replay-11111111-2222-4333-8444-555555555555",
|
||||
error_class="fixture_failure",
|
||||
)
|
||||
self.assertEqual(receipt["status"], "blocked_with_safe_next_action")
|
||||
terminal = receipt["stage_receipts"]["rollback_or_no_write_terminal"]
|
||||
self.assertEqual(terminal["candidate_registration_state"], "disabled")
|
||||
self.assertIs(receipt["promotion_allowed"], False)
|
||||
|
||||
def test_gitea_workflow_has_no_floating_or_external_source_controls(self) -> None:
|
||||
raw = WORKFLOW_PATH.read_text(encoding="utf-8").lower()
|
||||
forbidden = (
|
||||
"uses" + ":",
|
||||
"actions" + "/checkout",
|
||||
"github" + ".com",
|
||||
"api" + ".github" + ".com",
|
||||
"raw" + ".githubusercontent" + ".com",
|
||||
"codeload" + ".github" + ".com",
|
||||
"ghcr" + ".io",
|
||||
"npx" + " -y",
|
||||
"@" + "latest",
|
||||
":" + "latest",
|
||||
)
|
||||
self.assertFalse([value for value in forbidden if value in raw])
|
||||
self.assertIn("runs-on: awoooi-non110-host", raw)
|
||||
self.assertIn("wait-host-web-build-pressure.sh", raw)
|
||||
self.assertIn("mcp-replay-receipts", raw)
|
||||
self.assertIn("verify_external_mcp_replay_receipt.py", raw)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,325 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for the standalone external MCP replay verifier without mocks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
VERIFIER_PATH = ROOT / "scripts/security/verify_external_mcp_replay_receipt.py"
|
||||
CONTROLLER_PATH = ROOT / "scripts/security/external_mcp_replay_controller.py"
|
||||
POLICY_PATH = ROOT / "config/mcp/playwright-mcp-replay-policy.json"
|
||||
|
||||
|
||||
def _load(name: str, path: Path):
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
assert spec and spec.loader
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[name] = module
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
verifier = _load("verify_external_mcp_replay_receipt", VERIFIER_PATH)
|
||||
controller = _load("external_mcp_replay_controller_for_verifier_test", CONTROLLER_PATH)
|
||||
|
||||
|
||||
def _canonical(payload: object) -> bytes:
|
||||
return json.dumps(
|
||||
payload,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def _write(path: Path, payload: object) -> bytes:
|
||||
raw = _canonical(payload) + b"\n"
|
||||
path.write_bytes(raw)
|
||||
return raw
|
||||
|
||||
|
||||
def _policy_payload() -> dict:
|
||||
return json.loads(POLICY_PATH.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def _valid_controller_receipt() -> tuple[dict, dict, str]:
|
||||
policy = controller.load_policy(POLICY_PATH)
|
||||
source = controller.SourceEvidence(
|
||||
artifact_run_id="aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee",
|
||||
artifact_trace_id="mcp-artifact-aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee",
|
||||
artifact_controller_receipt_sha256="a" * 64,
|
||||
artifact_verifier_receipt_sha256="b" * 64,
|
||||
)
|
||||
observation = controller.ReplayObservation(
|
||||
server_name=policy.server_name,
|
||||
server_version=policy.server_version,
|
||||
protocol_version=policy.protocol_version,
|
||||
tool_count=policy.tool_count,
|
||||
tool_name_set_sha256=policy.tool_name_set_sha256,
|
||||
tool_schema_manifest_sha256=policy.tool_schema_manifest_sha256,
|
||||
process_exit_code=0,
|
||||
shutdown_mode="stdin_eof_clean_exit",
|
||||
ephemeral_container_removed=True,
|
||||
)
|
||||
receipt = controller.build_receipt(
|
||||
policy=policy,
|
||||
source=source,
|
||||
run_id="11111111-2222-4333-8444-555555555555",
|
||||
trace_id="mcp-replay-11111111-2222-4333-8444-555555555555",
|
||||
mode="apply",
|
||||
observation=observation,
|
||||
package_file_count=173,
|
||||
package_unpacked_bytes=1_000,
|
||||
)
|
||||
payload, checksum = verifier.load_and_validate_policy(POLICY_PATH)
|
||||
return payload, receipt, checksum
|
||||
|
||||
|
||||
def _artifact_chain(policy: dict) -> tuple[dict, dict]:
|
||||
artifact = policy["artifact_source"]
|
||||
run_id = "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee"
|
||||
controller_receipt = {
|
||||
"schema_version": "awoooi_external_mcp_artifact_receipt_v1",
|
||||
"status": "completed_internal_mirror_pending_independent_verifier",
|
||||
"run_id": run_id,
|
||||
"trace_id": f"mcp-artifact-{run_id}",
|
||||
"internal_mirror_ref": artifact["artifact_ref"],
|
||||
"policy_checksum": artifact["policy_checksum"],
|
||||
"promotion_allowed": False,
|
||||
"replay_allowed": False,
|
||||
"shadow_allowed": False,
|
||||
"canary_allowed": False,
|
||||
}
|
||||
verifier_receipt = {
|
||||
"schema_version": "awoooi_external_mcp_artifact_verifier_receipt_v1",
|
||||
"status": "completed_internal_mirror_verified",
|
||||
"run_id": run_id,
|
||||
"trace_id": f"mcp-artifact-{run_id}",
|
||||
"internal_mirror_ref": artifact["artifact_ref"],
|
||||
"policy_checksum": artifact["policy_checksum"],
|
||||
"promotion_allowed": False,
|
||||
"replay_allowed": False,
|
||||
"shadow_allowed": False,
|
||||
"canary_allowed": False,
|
||||
"independent_post_verifier": {
|
||||
"status": "passed",
|
||||
"container_or_mcp_started": False,
|
||||
},
|
||||
}
|
||||
return controller_receipt, verifier_receipt
|
||||
|
||||
|
||||
class StandalonePolicyTests(unittest.TestCase):
|
||||
def test_committed_policy_passes_independent_validation(self) -> None:
|
||||
policy, checksum = verifier.load_and_validate_policy(POLICY_PATH)
|
||||
self.assertEqual(policy["work_item_id"], "MCP-REPLAY-PLAYWRIGHT-001")
|
||||
self.assertRegex(checksum, r"^sha256:[0-9a-f]{64}$")
|
||||
|
||||
def test_any_enabled_execution_boundary_is_rejected(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
payload = _policy_payload()
|
||||
payload["execution_boundaries"]["browser_start_allowed"] = True
|
||||
path = Path(temp) / "policy.json"
|
||||
_write(path, payload)
|
||||
with self.assertRaisesRegex(verifier.VerifierError, "boundary_must_remain_false"):
|
||||
verifier.load_and_validate_policy(path)
|
||||
|
||||
def test_adapter_tool_partition_overlap_is_rejected(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
payload = _policy_payload()
|
||||
payload["protocol_contract"]["adapter_denied_tools"].append(
|
||||
"browser_snapshot"
|
||||
)
|
||||
path = Path(temp) / "policy.json"
|
||||
_write(path, payload)
|
||||
with self.assertRaisesRegex(verifier.VerifierError, "adapter_tool_partition_invalid"):
|
||||
verifier.load_and_validate_policy(path)
|
||||
|
||||
def test_tagged_artifact_reference_is_rejected(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
payload = _policy_payload()
|
||||
payload["artifact_source"]["artifact_ref"] = (
|
||||
"registry.wooo.work/awoooi/mcp-artifacts/playwright-mcp:0.0.78"
|
||||
)
|
||||
path = Path(temp) / "policy.json"
|
||||
_write(path, payload)
|
||||
with self.assertRaisesRegex(verifier.VerifierError, "artifact_ref_invalid"):
|
||||
verifier.load_and_validate_policy(path)
|
||||
|
||||
|
||||
class ControllerReceiptTests(unittest.TestCase):
|
||||
def test_valid_controller_receipt_passes_standalone_validation(self) -> None:
|
||||
policy, receipt, checksum = _valid_controller_receipt()
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
path = Path(temp) / "controller.json"
|
||||
raw = _write(path, receipt)
|
||||
observed, observed_sha = verifier.validate_controller_receipt(
|
||||
policy, checksum, path
|
||||
)
|
||||
self.assertEqual(observed["run_id"], receipt["run_id"])
|
||||
self.assertEqual(observed_sha, hashlib.sha256(raw).hexdigest())
|
||||
|
||||
def test_controller_browser_start_is_rejected(self) -> None:
|
||||
policy, receipt, checksum = _valid_controller_receipt()
|
||||
receipt["browser_started"] = True
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
path = Path(temp) / "controller.json"
|
||||
_write(path, receipt)
|
||||
with self.assertRaisesRegex(verifier.VerifierError, "controller_boundary_invalid"):
|
||||
verifier.validate_controller_receipt(policy, checksum, path)
|
||||
|
||||
def test_controller_missing_cleanup_terminal_is_rejected(self) -> None:
|
||||
policy, receipt, checksum = _valid_controller_receipt()
|
||||
receipt["stage_receipts"]["rollback_or_no_write_terminal"][
|
||||
"ephemeral_container_removed"
|
||||
] = False
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
path = Path(temp) / "controller.json"
|
||||
_write(path, receipt)
|
||||
with self.assertRaisesRegex(verifier.VerifierError, "controller_rollback_invalid"):
|
||||
verifier.validate_controller_receipt(policy, checksum, path)
|
||||
|
||||
def test_controller_protocol_hash_drift_is_rejected(self) -> None:
|
||||
policy, receipt, checksum = _valid_controller_receipt()
|
||||
receipt["protocol_observation"]["tool_name_set_sha256"] = "0" * 64
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
path = Path(temp) / "controller.json"
|
||||
_write(path, receipt)
|
||||
with self.assertRaisesRegex(
|
||||
verifier.VerifierError, "controller_protocol_observation_mismatch"
|
||||
):
|
||||
verifier.validate_controller_receipt(policy, checksum, path)
|
||||
|
||||
|
||||
class ArtifactChainAndToolTests(unittest.TestCase):
|
||||
def test_valid_artifact_chain_passes(self) -> None:
|
||||
policy = _policy_payload()
|
||||
artifact = policy["artifact_source"]
|
||||
artifact_controller, artifact_verifier = _artifact_chain(policy)
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
controller_path = root / "artifact-controller.json"
|
||||
verifier_path = root / "artifact-verifier.json"
|
||||
controller_raw = _write(controller_path, artifact_controller)
|
||||
artifact_verifier["controller_receipt_sha256"] = hashlib.sha256(
|
||||
controller_raw
|
||||
).hexdigest()
|
||||
verifier_raw = _write(verifier_path, artifact_verifier)
|
||||
artifact["controller_receipt_sha256"] = hashlib.sha256(
|
||||
controller_raw
|
||||
).hexdigest()
|
||||
artifact["verifier_receipt_sha256"] = hashlib.sha256(
|
||||
verifier_raw
|
||||
).hexdigest()
|
||||
observed = verifier.validate_artifact_receipt_chain(
|
||||
policy, controller_path, verifier_path
|
||||
)
|
||||
self.assertEqual(observed[0], artifact["controller_receipt_sha256"])
|
||||
self.assertEqual(observed[1], artifact["verifier_receipt_sha256"])
|
||||
|
||||
def test_artifact_chain_tamper_is_rejected(self) -> None:
|
||||
policy = _policy_payload()
|
||||
artifact = policy["artifact_source"]
|
||||
artifact_controller, artifact_verifier = _artifact_chain(policy)
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
controller_path = root / "artifact-controller.json"
|
||||
verifier_path = root / "artifact-verifier.json"
|
||||
controller_raw = _write(controller_path, artifact_controller)
|
||||
artifact_verifier["controller_receipt_sha256"] = hashlib.sha256(
|
||||
controller_raw
|
||||
).hexdigest()
|
||||
verifier_raw = _write(verifier_path, artifact_verifier)
|
||||
artifact["controller_receipt_sha256"] = hashlib.sha256(
|
||||
controller_raw
|
||||
).hexdigest()
|
||||
artifact["verifier_receipt_sha256"] = hashlib.sha256(
|
||||
verifier_raw
|
||||
).hexdigest()
|
||||
verifier_path.write_bytes(verifier_raw + b" ")
|
||||
with self.assertRaisesRegex(verifier.VerifierError, "artifact_receipt_chain_invalid"):
|
||||
verifier.validate_artifact_receipt_chain(
|
||||
policy, controller_path, verifier_path
|
||||
)
|
||||
|
||||
def test_independent_tool_hashes_are_order_independent(self) -> None:
|
||||
tools = [
|
||||
{"name": "zeta", "inputSchema": {"type": "object"}},
|
||||
{
|
||||
"name": "alpha",
|
||||
"inputSchema": {"type": "object"},
|
||||
"annotations": {"readOnlyHint": True},
|
||||
},
|
||||
]
|
||||
first = verifier._independent_tool_hashes(tools)
|
||||
second = verifier._independent_tool_hashes(list(reversed(tools)))
|
||||
self.assertEqual(first, second)
|
||||
self.assertEqual(first[3], {"alpha", "zeta"})
|
||||
|
||||
def test_duplicate_tool_names_are_rejected(self) -> None:
|
||||
tools = [
|
||||
{"name": "duplicate", "inputSchema": {}},
|
||||
{"name": "duplicate", "inputSchema": {}},
|
||||
]
|
||||
with self.assertRaisesRegex(verifier.VerifierError, "tool_names_not_unique"):
|
||||
verifier._independent_tool_hashes(tools)
|
||||
|
||||
|
||||
class VerifierReceiptTests(unittest.TestCase):
|
||||
def test_completed_verifier_receipt_keeps_promotion_disabled(self) -> None:
|
||||
policy, controller_receipt, policy_checksum = _valid_controller_receipt()
|
||||
receipt = verifier.build_verifier_receipt(
|
||||
policy=policy,
|
||||
policy_checksum=policy_checksum,
|
||||
controller=controller_receipt,
|
||||
controller_sha256="c" * 64,
|
||||
artifact_controller_sha256="a" * 64,
|
||||
artifact_verifier_sha256="b" * 64,
|
||||
observation={
|
||||
"protocol_version": "2025-06-18",
|
||||
"server_name": "Playwright",
|
||||
"server_version": "1.62.0-alpha-1783623505000",
|
||||
"tool_count": 24,
|
||||
"tool_name_set_sha256": "d" * 64,
|
||||
"tool_schema_manifest_sha256": "e" * 64,
|
||||
"process_exit_code": 0,
|
||||
"shutdown_mode": "stdin_eof_clean_exit",
|
||||
"ephemeral_container_removed": True,
|
||||
},
|
||||
package_file_count=173,
|
||||
package_unpacked_bytes=17_000_000,
|
||||
)
|
||||
self.assertEqual(receipt["status"], "completed_protocol_schema_replay_verified")
|
||||
self.assertEqual(
|
||||
receipt["independent_post_verifier"]["implementation"],
|
||||
"standalone_no_controller_import",
|
||||
)
|
||||
for field in (
|
||||
"browser_started",
|
||||
"tool_call_performed",
|
||||
"external_rag_write_performed",
|
||||
"production_write_performed",
|
||||
"promotion_allowed",
|
||||
):
|
||||
self.assertIs(receipt[field], False)
|
||||
|
||||
def test_failure_receipt_has_rollback_terminal(self) -> None:
|
||||
policy = _policy_payload()
|
||||
receipt = verifier._failure_receipt(policy, {}, "fixture_failure")
|
||||
self.assertEqual(receipt["status"], "blocked_with_safe_next_action")
|
||||
self.assertEqual(
|
||||
receipt["rollback_terminal"]["candidate_registration_state"], "disabled"
|
||||
)
|
||||
self.assertIs(receipt["promotion_allowed"], False)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user