#!/usr/bin/env python3 """Deterministic unit and cross-runtime conformance tests for the kernel.""" from __future__ import annotations import os import stat import sys import tempfile import unittest from copy import deepcopy from pathlib import Path from unittest.mock import patch HARNESS_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(HARNESS_ROOT)) from kernel.adapters import AgenticBridgeAdapter, NativeHarnessAdapter from kernel.contracts import validate_bundle from kernel.policy import ( H2GateContext, classify_action, evaluate_failure_policy, evaluate_h2_gate, evaluate_registry_configuration, evaluate_risk, evaluate_trust_capabilities, ) from kernel.taxonomy import CertificationClaim, ReadinessCheck, ReportDimension, RuntimeControl, same_legacy_label from kernel.supervision import RuntimeCapabilities, negotiate_intervention class RiskFloorTests(unittest.TestCase): def test_benign_deploy_has_high_floor_and_requires_approval(self) -> None: decision = evaluate_risk(action="deploy", content="publish a harmless documentation update", actor="alice") self.assertEqual(decision["action_class"], "deployment") self.assertEqual(decision["risk_factors"]["content_risk"], "low") self.assertEqual(decision["effective_risk"], "high") self.assertEqual(decision["decision"], "require_approval") def test_effective_risk_is_maximum_of_all_factors(self) -> None: decision = evaluate_risk(action="read", resource="/.ssh/id_rsa", actor="alice") self.assertEqual(decision["risk_factors"]["action_risk"], "high") self.assertEqual(decision["risk_factors"]["resource_risk"], "high") self.assertEqual(decision["effective_risk"], "high") def test_missing_actor_denies_high_impact_action(self) -> None: decision = evaluate_risk(action="migration", content="apply schema", actor="") self.assertEqual(decision["decision"], "deny") self.assertIn("actor_identity_required", decision["reason_codes"]) def test_read_only_shell_command_is_not_side_effecting(self) -> None: decision = classify_action(tool="Bash", command="git status --short") self.assertEqual(decision["action_class"], "read_only") self.assertFalse(decision["side_effecting"]) class RegistryConfigurationTests(unittest.TestCase): def test_enforce_defaults_registry_on(self) -> None: result = evaluate_registry_configuration("enforce", "development", None) self.assertTrue(result["enabled"]) self.assertEqual(result["reason_code"], "h2_registry_default_enabled") def test_development_bypass_is_visible_and_non_silent(self) -> None: result = evaluate_registry_configuration("enforce", "test", "0") self.assertFalse(result["enabled"]) self.assertTrue(result["unsafe_bypass"]) self.assertEqual(result["severity"], "high") def test_production_bypass_is_rejected(self) -> None: result = evaluate_registry_configuration("enforce", "production", "0") self.assertFalse(result["valid"]) self.assertEqual(result["reason_code"], "h2_registry_bypass_forbidden") class FailurePolicyTests(unittest.TestCase): def test_all_supported_policies_are_explicit(self) -> None: for configured in ("halt", "quarantine", "require_approval", "record_only"): result = evaluate_failure_policy( side_effecting=False, mode="observe", profile="test", configured=configured, ) self.assertEqual(result["failure_policy"], configured) self.assertEqual(result["execution_result"], "failed") self.assertEqual(result["telemetry_result"], "recorded") def test_production_side_effect_cannot_be_record_only(self) -> None: result = evaluate_failure_policy( side_effecting=True, mode="enforce", profile="production", configured="record_only", ) self.assertEqual(result["failure_policy"], "halt") self.assertFalse(result["valid"]) self.assertEqual(result["source"], "production_record_only_forbidden") def test_invalid_policy_fails_closed(self) -> None: result = evaluate_failure_policy( side_effecting=True, mode="enforce", profile="test", configured="continue_anyway", ) self.assertEqual(result["failure_policy"], "halt") self.assertFalse(result["valid"]) class TaxonomyTests(unittest.TestCase): def test_same_h_label_cannot_be_interpreted_as_same_category(self) -> None: self.assertTrue(same_legacy_label(RuntimeControl.H4, ReadinessCheck.H4)) self.assertNotEqual(RuntimeControl.H4.value, ReadinessCheck.H4.value) self.assertNotEqual(ReportDimension.H4.value, CertificationClaim.H4.value) def test_unsupported_runtime_intervention_is_explicit(self) -> None: result = negotiate_intervention("i-1", "roll_back", "operator", RuntimeCapabilities()) self.assertEqual(result["status"], "unsupported") self.assertEqual(result["reason_code"], "runtime_intervention_unsupported") class H2DependencyTests(unittest.TestCase): def context(self, mode: str) -> H2GateContext: return H2GateContext(mode, "alice", "write_file", "Edit", "run-1", "test.h2", "idem", 0.5) def script(self, directory: str, body: str) -> str: path = Path(directory) / "gate.sh" path.write_text("#!/usr/bin/env bash\n" + body + "\n", encoding="utf-8") path.chmod(0o700) return str(path) def test_missing_gate_denies_enforce_and_degrades_observe(self) -> None: with tempfile.TemporaryDirectory() as temp: missing = str(Path(temp) / "missing.sh") enforced = evaluate_h2_gate(missing, self.context("enforce")) observed = evaluate_h2_gate(missing, self.context("observe")) self.assertFalse(enforced["execution_allowed"]) self.assertEqual(enforced["reason_code"], "h2_gate_unavailable") self.assertTrue(observed["execution_allowed"]) self.assertEqual(observed["decision"], "observe_only") self.assertFalse(observed["certifiable"]) def test_permission_denied_fails_closed(self) -> None: with tempfile.TemporaryDirectory() as temp: path = self.script(temp, "echo 'TOOL_APPROVED tool=x reason=test'") os.chmod(path, 0) result = evaluate_h2_gate(path, self.context("enforce")) os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) self.assertFalse(result["execution_allowed"]) self.assertEqual(result["reason_code"], "h2_gate_permission_denied") def test_timeout_malformed_and_internal_exception_fail_closed(self) -> None: with tempfile.TemporaryDirectory() as temp: timeout = self.script(temp, "sleep 1") timed = evaluate_h2_gate(timeout, self.context("enforce")) malformed = self.script(temp, "echo nonsense") malformed_result = evaluate_h2_gate(malformed, self.context("enforce")) with patch("kernel.policy.subprocess.run", side_effect=OSError("boom")): internal = evaluate_h2_gate(malformed, self.context("enforce")) self.assertEqual(timed["reason_code"], "h2_gate_timeout") self.assertEqual(malformed_result["reason_code"], "h2_gate_malformed_response") self.assertEqual(internal["reason_code"], "h2_gate_internal_error") self.assertTrue(all(not item["execution_allowed"] for item in (timed, malformed_result, internal))) def test_explicit_approval_and_denial_protocols(self) -> None: with tempfile.TemporaryDirectory() as temp: allowed = self.script(temp, "echo 'TOOL_APPROVED tool=write_file reason=registered'") allow_result = evaluate_h2_gate(allowed, self.context("enforce")) denied = self.script(temp, "echo 'TOOL_DENIED tool=write_file reason=missing_agent_identity' >&2; exit 2") deny_result = evaluate_h2_gate(denied, self.context("enforce")) self.assertTrue(allow_result["execution_allowed"]) self.assertFalse(deny_result["execution_allowed"]) self.assertEqual(deny_result["reason_code"], "h2_denied:missing_agent_identity") class TrustRootTests(unittest.TestCase): def test_production_refuses_local_fallback(self) -> None: result = evaluate_trust_capabilities({"CASAN_PROFILE": "production"}) self.assertFalse(result["ready"]) self.assertFalse(result["certifiable"]) self.assertIn("external_signing_trust_root_required", result["reason_codes"]) def test_external_configuration_is_capable_but_not_claimed_provisioned(self) -> None: result = evaluate_trust_capabilities({ "CASAN_PROFILE": "production", "CASAN_SIGNING_PROVIDER": "vault_kms", "VAULT_ADDR": "https://vault.example", "VAULT_TOKEN": "redacted-runtime-token", "CASAN_IMMUTABLE_ANCHOR_PROVIDER": "s3_object_lock", "CASAN_S3_BUCKET": "audit-lock", "CASAN_S3_KMS_KEY_ID": "kms-key", }) self.assertTrue(result["ready"]) self.assertTrue(result["certifiable"]) def test_emergency_override_never_certifies(self) -> None: result = evaluate_trust_capabilities({ "CASAN_PROFILE": "production", "CASAN_TRUST_EMERGENCY_OVERRIDE": "1", }) self.assertFalse(result["certifiable"]) self.assertEqual(result["severity"], "critical") class CrossRuntimeConformanceTests(unittest.TestCase): def events(self, **overrides: object) -> tuple[dict[str, object], dict[str, object]]: common: dict[str, object] = { "run_id": "run-1", "trace_id": "run-1", "action": "deploy", "tool": "Bash", "command": "deploy harmless docs", "actor": "alice", "mode": "enforce", "execution_status": "failed", "assurance_status": "failed", "certification_status": "non_certified", "certified": False, "correlation_id": "corr-1", "parent_run_id": "parent-1", "parent_step_id": "parent-step-1", "evidence": [ {"evidence_id": "e-1", "decision": "deny", "category": "runtime_control"}, ], } common.update(overrides) return dict(common), dict(common, adapter_version="20.2.0") def map_both(self, **overrides: object) -> tuple[dict[str, object], dict[str, object]]: native_event, agentic_event = self.events(**overrides) native = NativeHarnessAdapter().map_execution(native_event) agentic = AgenticBridgeAdapter().map_execution(agentic_event) return native, agentic def test_01_same_dangerous_action_classification(self) -> None: native, agentic = self.map_both() self.assertEqual(validate_bundle(native), []) self.assertEqual(validate_bundle(agentic), []) self.assertEqual(native["run"]["risk_summary"]["action_class"], "deployment") self.assertEqual(agentic["run"]["risk_summary"]["action_class"], "deployment") self.assertEqual(native["run"]["risk_summary"]["effective_risk"], "high") def test_02_same_missing_actor_denial(self) -> None: native, agentic = self.map_both(actor="") for bundle in (native, agentic): self.assertEqual(bundle["run"]["risk_summary"]["decision"], "deny") self.assertIn("actor_identity_required", bundle["run"]["risk_summary"]["reason_codes"]) def test_03_same_missing_h2_gate_denial(self) -> None: with tempfile.TemporaryDirectory() as temp: missing = str(Path(temp) / "missing.sh") decisions = [ evaluate_h2_gate(missing, H2GateContext("enforce", "alice", "write_file", "Edit", "run-1", point, "idem")) for point in ("native-harness.h2", "agentic-bridge.h2") ] native, agentic = self.map_both( action="write_file", tool="Edit", command="safe edit", policy_decisions=[decisions[0]], evidence=[{"evidence_id": "e-1", "decision": "deny"}], ) # The second adapter receives the same kernel decision with only its # enforcement-point extension changed. agentic_event = self.events( action="write_file", tool="Edit", command="safe edit", policy_decisions=[decisions[1]], evidence=[{"evidence_id": "e-1", "decision": "deny"}], )[1] agentic = AgenticBridgeAdapter().map_execution(agentic_event) for bundle in (native, agentic): decision = bundle["steps"][0]["policy_decisions"][0] self.assertEqual(decision["decision"], "deny") self.assertIn("h2_gate_unavailable", decision["reason_codes"]) def test_04_same_approval_requirement(self) -> None: native, agentic = self.map_both(actor="alice") for bundle in (native, agentic): self.assertEqual(bundle["run"]["risk_summary"]["decision"], "require_approval") self.assertEqual(bundle["run"]["risk_summary"]["evidence_requirement"], "required") def test_05_same_claim_evidence_relationship(self) -> None: native, agentic = self.map_both() for bundle in (native, agentic): self.assertEqual(bundle["claims"][0]["evidence_refs"], ["e-1"]) self.assertEqual(bundle["evidence"][0]["claim_refs"], [bundle["claims"][0]["claim_id"]]) def test_06_observed_only_execution_is_never_certified(self) -> None: native, agentic = self.map_both( mode="observe", certification_status="certified", certified=True, ) for bundle in (native, agentic): self.assertIn("observed_only_cannot_be_certified", validate_bundle(bundle)) def test_07_execution_and_assurance_outcomes_are_distinct(self) -> None: native, agentic = self.map_both( execution_status="failed", assurance_status="passed", certification_status="non_certified", ) for bundle in (native, agentic): outcome = bundle["run"]["outcome"] self.assertEqual(outcome["execution_result"], "failed") self.assertEqual(outcome["assurance_result"], "passed") self.assertEqual(outcome["certification_result"], "non_certified") def test_08_correlation_and_causation_are_preserved(self) -> None: native, agentic = self.map_both() for bundle in (native, agentic): self.assertEqual(bundle["run"]["correlation_id"], "corr-1") self.assertEqual(bundle["run"]["parent_run_id"], "parent-1") self.assertEqual(bundle["steps"][0]["parent_step_id"], "parent-step-1") self.assertEqual(bundle["trace_links"][0]["from"], "run-1") def test_09_same_verifier_detects_evidence_tampering(self) -> None: native, agentic = self.map_both() for original in (native, agentic): self.assertEqual(validate_bundle(original), []) tampered = deepcopy(original) tampered["evidence"][0]["artifact_ref"] = "inline:tampered" self.assertIn("evidence.0.integrity_invalid", validate_bundle(tampered)) def test_10_runtime_details_are_namespaced_extensions(self) -> None: native, agentic = self.map_both() self.assertNotEqual(native["run"]["runtime"]["type"], agentic["run"]["runtime"]["type"]) self.assertIn("casan.runtime.casan-native-harness", native["extensions"]) self.assertIn("casan.runtime.agentic-bridge", agentic["extensions"]) def test_delegation_depth_is_enforced(self) -> None: native, _ = self.map_both() native["actors"].append({ "actor_id": "child", "actor_type": "child_agent", "issuer": "alice", "trust_level": "verified", "authentication_evidence_ref": None, }) native["delegations"] = [ {"delegation_id": "d1", "delegator_ref": "alice", "delegate_ref": "child", "allowed_authority": ["read"], "resource_scope": ["project"], "expiry": "2030-01-01T00:00:00Z", "maximum_depth": 3, "parent_delegation_ref": None, "revoked": False, "approval_ref": None}, {"delegation_id": "d2", "delegator_ref": "alice", "delegate_ref": "child", "allowed_authority": ["read"], "resource_scope": ["project"], "expiry": "2030-01-01T00:00:00Z", "maximum_depth": 3, "parent_delegation_ref": "d1", "revoked": False, "approval_ref": None}, {"delegation_id": "d3", "delegator_ref": "alice", "delegate_ref": "child", "allowed_authority": ["read"], "resource_scope": ["project"], "expiry": "2030-01-01T00:00:00Z", "maximum_depth": 1, "parent_delegation_ref": "d2", "revoked": False, "approval_ref": None}, ] self.assertIn("delegations.2.maximum_depth_exceeded", validate_bundle(native)) if __name__ == "__main__": unittest.main(verbosity=2)