Add permission_gate smoke tests: per-tier capability sets + monotonicity + severity + sink + permissive + unknown-cap diagnostic

smoke_test_permission_gate.py

@@ -0,0 +1,222 @@
+"""
+Smoke test for permission_gate.
+
+Coverage:
+  1. T0 (inert) has only compute.cpu — denied everything privileged.
+  2. T1 (read-only observer) has vault.read but NOT vault.append.
+  3. T2 (default I/O worker) has vault.append but NOT vault.schema.
+  4. T3 has net.egress.allowlisted but NOT net.egress.arbitrary.
+  5. T4 has compute.gpu.train but NOT vault.schema.
+  6. T5 has vault.amend but NOT vault.schema.
+  7. T6 has every capability defined.
+  8. Tier upgrade is monotonic (T_{n+1} grants ⊇ T_n grants).
+  9. High-severity capabilities flagged correctly in audit records.
+ 10. Audit sink callable receives every check.
+ 11. permissive() grants everything regardless of tier.
+ 12. Unknown capability string raises PermissionDenied with diagnostic.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+import permission_gate as pg
+from permission_gate import Capability, PermissionGate, PermissionDenied
+
+
+def hr(title: str) -> None:
+    print(f"\n{'=' * 6} {title} {'=' * 6}")
+
+
+def expect_denied(gate, capability, tier, *, label):
+    try:
+        gate.check(capability, f"agent-tier{tier}", tier)
+        print(f"  FAIL: T{tier} {capability.value} should have been denied — {label}")
+        sys.exit(1)
+    except PermissionDenied:
+        return  # expected
+
+
+def expect_granted(gate, capability, tier, *, label):
+    try:
+        gate.check(capability, f"agent-tier{tier}", tier)
+    except PermissionDenied as e:
+        print(f"  FAIL: T{tier} {capability.value} should have been granted — {label}: {e}")
+        sys.exit(1)
+
+
+# ===========================================================================
+# 1. T0 — inert. Only compute.cpu.
+# ===========================================================================
+hr("1. T0 (inert)")
+gate = PermissionGate()
+expect_granted(gate, Capability.COMPUTE_CPU, 0, label="T0 compute.cpu")
+expect_denied(gate, Capability.VAULT_READ, 0, label="T0 cannot vault.read")
+expect_denied(gate, Capability.VAULT_APPEND, 0, label="T0 cannot vault.append")
+expect_denied(gate, Capability.FS_READ_CACHE, 0, label="T0 cannot fs.read.cache")
+print("  ✓ T0 capability set behaves as inert")
+
+
+# ===========================================================================
+# 2. T1 — read-only observer
+# ===========================================================================
+hr("2. T1 (read-only observer)")
+gate = PermissionGate()
+expect_granted(gate, Capability.VAULT_READ, 1, label="T1 vault.read")
+expect_granted(gate, Capability.FS_READ_CACHE, 1, label="T1 fs.read.cache")
+expect_granted(gate, Capability.FS_READ_CODE, 1, label="T1 fs.read.code")
+expect_denied(gate, Capability.VAULT_APPEND, 1, label="T1 cannot vault.append")
+expect_denied(gate, Capability.FS_WRITE_SCRATCH, 1, label="T1 cannot fs.write.scratch")
+expect_denied(gate, Capability.COMPUTE_GPU_INFERENCE, 1, label="T1 cannot compute.gpu.inference")
+print("  ✓ T1 read-only enforced")
+
+
+# ===========================================================================
+# 3. T2 — default I/O worker
+# ===========================================================================
+hr("3. T2 (default I/O worker)")
+gate = PermissionGate()
+expect_granted(gate, Capability.VAULT_APPEND, 2, label="T2 vault.append")
+expect_granted(gate, Capability.COMPUTE_GPU_INFERENCE, 2, label="T2 compute.gpu.inference")
+expect_granted(gate, Capability.FS_WRITE_SCRATCH, 2, label="T2 fs.write.scratch")
+expect_denied(gate, Capability.COMPUTE_GPU_PROFILE, 2, label="T2 cannot compute.gpu.profile")
+expect_denied(gate, Capability.FS_WRITE_OUTPUTS, 2, label="T2 cannot fs.write.outputs")
+expect_denied(gate, Capability.NET_EGRESS_ALLOWLISTED, 2, label="T2 cannot net.egress")
+expect_denied(gate, Capability.VAULT_SCHEMA, 2, label="T2 cannot vault.schema")
+print("  ✓ T2 default-tier behavior correct")
+
+
+# ===========================================================================
+# 4. T3 — net-fetcher / profiler
+# ===========================================================================
+hr("4. T3 (network fetcher / profiler)")
+gate = PermissionGate()
+expect_granted(gate, Capability.NET_EGRESS_ALLOWLISTED, 3, label="T3 net.egress.allowlisted")
+expect_granted(gate, Capability.COMPUTE_GPU_PROFILE, 3, label="T3 compute.gpu.profile")
+expect_granted(gate, Capability.FS_WRITE_OUTPUTS, 3, label="T3 fs.write.outputs")
+expect_denied(gate, Capability.COMPUTE_GPU_TRAIN, 3, label="T3 cannot compute.gpu.train")
+expect_denied(gate, Capability.NET_EGRESS_ARBITRARY, 3, label="T3 cannot net.egress.arbitrary")
+expect_denied(gate, Capability.FS_WRITE_CONFIG, 3, label="T3 cannot fs.write.config")
+print("  ✓ T3 profiler tier correct")
+
+
+# ===========================================================================
+# 5. T4 — trainer / quantizer
+# ===========================================================================
+hr("5. T4 (trainer / quantizer)")
+gate = PermissionGate()
+expect_granted(gate, Capability.COMPUTE_GPU_TRAIN, 4, label="T4 compute.gpu.train")
+expect_granted(gate, Capability.COMPUTE_SUBPROCESS_SANDBOXED, 4,
+               label="T4 compute.subprocess.sandboxed")
+expect_granted(gate, Capability.AGENT_SPAWN_ESCALATED, 4, label="T4 agent.spawn.escalated")
+expect_denied(gate, Capability.FS_WRITE_CONFIG, 4, label="T4 cannot fs.write.config")
+expect_denied(gate, Capability.VAULT_AMEND, 4, label="T4 cannot vault.amend")
+expect_denied(gate, Capability.COMPUTE_SUBPROCESS_UNSANDBOXED, 4,
+              label="T4 cannot compute.subprocess.unsandboxed")
+print("  ✓ T4 trainer tier correct")
+
+
+# ===========================================================================
+# 6. T5 — orchestrator
+# ===========================================================================
+hr("6. T5 (orchestrator)")
+gate = PermissionGate()
+expect_granted(gate, Capability.FS_WRITE_CONFIG, 5, label="T5 fs.write.config")
+expect_granted(gate, Capability.VAULT_AMEND, 5, label="T5 vault.amend")
+expect_granted(gate, Capability.AGENT_TERMINATE_PEER, 5, label="T5 agent.terminate.peer")
+expect_denied(gate, Capability.VAULT_SCHEMA, 5, label="T5 cannot vault.schema")
+expect_denied(gate, Capability.FS_WRITE_CODE, 5, label="T5 cannot fs.write.code")
+expect_denied(gate, Capability.NET_EGRESS_ARBITRARY, 5, label="T5 cannot net.egress.arbitrary")
+print("  ✓ T5 orchestrator tier correct")
+
+
+# ===========================================================================
+# 7. T6 — system operator. Ceiling.
+# ===========================================================================
+hr("7. T6 (system operator)")
+gate = PermissionGate()
+for cap in Capability:
+    expect_granted(gate, cap, 6, label=f"T6 {cap.value}")
+print(f"  ✓ T6 grants all {len(list(Capability))} capabilities")
+
+
+# ===========================================================================
+# 8. Monotonic tier grants
+# ===========================================================================
+hr("8. Tier upgrades are monotonic")
+gate = PermissionGate()
+for tier in range(6):
+    lower = gate.tier_grants(tier)
+    upper = gate.tier_grants(tier + 1)
+    missing = lower - upper
+    assert not missing, (
+        f"T{tier+1} missing capabilities T{tier} had: {sorted(c.value for c in missing)}"
+    )
+    print(f"  T{tier} ({len(lower)} caps) ⊆ T{tier+1} ({len(upper)} caps) ✓")
+
+
+# ===========================================================================
+# 9. High-severity flagging
+# ===========================================================================
+hr("9. High-severity capabilities marked in audit")
+gate = PermissionGate()
+gate.check(Capability.VAULT_APPEND, "agent-A", 2)   # NORMAL
+gate.check(Capability.VAULT_SCHEMA, "agent-A", 6)   # HIGH
+gate.check(Capability.FS_WRITE_CODE, "agent-A", 6)  # HIGH
+gate.check(Capability.NET_EGRESS_ARBITRARY, "agent-A", 6)  # HIGH
+
+by_severity = {}
+for r in gate.audit_log:
+    by_severity.setdefault(r.severity, []).append(r.capability)
+print(f"  NORMAL entries: {by_severity.get('NORMAL', [])}")
+print(f"  HIGH entries:   {by_severity.get('HIGH', [])}")
+assert len(by_severity["NORMAL"]) == 1
+assert len(by_severity["HIGH"]) == 3
+
+
+# ===========================================================================
+# 10. Audit sink callable
+# ===========================================================================
+hr("10. Audit sink receives every check")
+sunk = []
+gate = PermissionGate(audit_sink=lambda r: sunk.append(r))
+gate.check(Capability.VAULT_READ, "agent-S", 1)
+gate.check(Capability.VAULT_APPEND, "agent-S", 2)
+try:
+    gate.check(Capability.VAULT_SCHEMA, "agent-S", 2)
+except PermissionDenied:
+    pass
+assert len(sunk) == 3, f"expected 3 sink calls, got {len(sunk)}"
+assert sunk[2].granted is False
+print(f"  sink received {len(sunk)} records (incl. 1 denial) ✓")
+
+
+# ===========================================================================
+# 11. permissive() grants everything
+# ===========================================================================
+hr("11. PermissionGate.permissive() bypasses tier checks")
+gate = PermissionGate.permissive()
+gate.check(Capability.FS_WRITE_CODE, "test-agent", 0)
+gate.check(Capability.VAULT_SCHEMA, "test-agent", 0)
+gate.check(Capability.NET_EGRESS_ARBITRARY, "test-agent", 0)
+assert len(gate.audit_log) == 3
+assert all(r.granted for r in gate.audit_log)
+print(f"  permissive granted {len(gate.audit_log)} caps to T0 (test-only)")
+
+
+# ===========================================================================
+# 12. Unknown capability raises with diagnostic
+# ===========================================================================
+hr("12. Unknown capability string fails fast")
+gate = PermissionGate()
+try:
+    gate.check("foo.bar.baz", "agent-X", 6)
+    print(f"  FAIL: unknown capability should have raised")
+    sys.exit(1)
+except PermissionDenied as e:
+    assert "unknown capability" in str(e).lower()
+    print(f"  caught: {e} ✓")
+
+
+print("\nAll assertions passed.")