tegwick
cdf0d9bf25
Window-scoped write role (CCR-2026-0058 approved as amended), bind self-rotation (pilot step is OPERATE on the one bind; svc-executor-pw dropped), pilot without the PDP gate (not-evaluated), started event plus one receipt per run, rapp Job in sso, owner workplans approved. New tasks: T11 core runner, T12 pilot operations and packaging; T06 depends on T12. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 994874@bnt-lap001 Assistant-Session: 6017c416-d619-47eb-bd1a-f670a02d7d87
198 lines
10 KiB
Python
198 lines
10 KiB
Python
"""Tests for the Bounded Execution Admission validator (NK-WP-0056-T01). Synthetic data only."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
import unittest
|
|
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
TOOL = HERE.parent
|
|
ROOT = TOOL.parents[1]
|
|
sys.path.insert(0, str(TOOL))
|
|
import validate as V # noqa: E402
|
|
|
|
SCHEMA = json.loads(V.SCHEMA.read_text())
|
|
EXAMPLES = ROOT / "examples/bounded-execution-admission"
|
|
ROTATION = V.load(EXAMPLES / "bind-rotation-svc-authelia.yaml")
|
|
SUPERADMIN = V.load(EXAMPLES / "bind-split-switch-super-admin.yaml")
|
|
VERIFY = V.load(EXAMPLES / "bind-split-read-only-verify.yaml")
|
|
|
|
|
|
def mutate(document, fn, redigest=True):
|
|
d = copy.deepcopy(document)
|
|
fn(d["spec"])
|
|
if redigest:
|
|
d["spec"]["plan_digest"] = V.plan_digest(d)
|
|
return d
|
|
|
|
|
|
def rules(document):
|
|
return {f["rule"] for f in V.evaluate(document, SCHEMA)["findings"]}
|
|
|
|
|
|
def record(document, **over):
|
|
r = {"apiVersion": V.CARRIER_RECORD, "carrier": "approval-engine", "approval_ref": "apr-synthetic-0001",
|
|
"admission_id": document["spec"]["id"], "admission_digest": V.admission_digest(document),
|
|
"approver": {"name": "admin", "principal_type": "HUMAN", "assurance": "aal2"},
|
|
"approved_at": "2026-11-03T08:30:00Z",
|
|
"consumptions": [{"at": "2026-11-03T09:05:00Z", "by": document["spec"]["executor"]["identity"]}]}
|
|
r.update(over)
|
|
return r
|
|
|
|
|
|
class ExampleTests(unittest.TestCase):
|
|
def test_all_examples_valid(self):
|
|
for path in sorted(EXAMPLES.glob("*.yaml")):
|
|
with self.subTest(path=path.name):
|
|
report = V.evaluate(V.load(path), SCHEMA)
|
|
self.assertTrue(report["valid"], report["findings"])
|
|
|
|
def test_invalid_fixtures_refused(self):
|
|
for path in sorted((TOOL / "fixtures/invalid").glob("*.yaml")):
|
|
with self.subTest(path=path.name):
|
|
self.assertFalse(V.evaluate(V.load(path), SCHEMA)["valid"])
|
|
|
|
def test_examples_pin_current_source(self):
|
|
"""Pinned hashes are real files of this repository (they drift when the files change)."""
|
|
for path in sorted(EXAMPLES.glob("*.yaml")):
|
|
for f in V.load(path)["spec"]["plan"]["source"]["files"]:
|
|
self.assertTrue((ROOT / f["path"]).is_file(), f["path"])
|
|
|
|
|
|
class RuleTests(unittest.TestCase):
|
|
def test_bea1_schema(self):
|
|
self.assertEqual(rules(mutate(ROTATION, lambda s: s.pop("budget"), redigest=False)), {"BEA-1"})
|
|
self.assertEqual(rules(mutate(ROTATION, lambda s: s["approval"].update(assurance="aal1"))), {"BEA-1"})
|
|
self.assertEqual(rules(mutate(ROTATION, lambda s: s["consumption"].update(single_use=False))), {"BEA-1"})
|
|
|
|
def test_bea2_plan_digest_binds_the_exact_plan(self):
|
|
changed = mutate(ROTATION, lambda s: s["plan"]["steps"][0].update(operation="something-else"), redigest=False)
|
|
self.assertEqual(rules(changed), {"BEA-2"})
|
|
|
|
def test_bea3_window(self):
|
|
self.assertIn("BEA-3", rules(mutate(ROTATION, lambda s: s["window"].update(not_after="2026-11-04T10:00:00Z"))))
|
|
self.assertIn("BEA-3", rules(mutate(ROTATION, lambda s: s["window"].update(not_after="2026-11-03T08:00:00Z"))))
|
|
self.assertIn("BEA-3", rules(mutate(SUPERADMIN, lambda s: s["window"].update(not_after="2026-11-03T14:00:00Z"))))
|
|
|
|
def test_window_state(self):
|
|
at = V.parse_time
|
|
self.assertEqual(V.evaluate(ROTATION, SCHEMA, at("2026-11-03T10:00:00Z"))["window_state"], "open")
|
|
self.assertEqual(V.evaluate(ROTATION, SCHEMA, at("2026-11-03T11:00:00Z"))["window_state"], "expired")
|
|
self.assertEqual(V.evaluate(ROTATION, SCHEMA, at("2026-11-02T10:00:00Z"))["window_state"], "not-yet")
|
|
|
|
def test_bea4_executor(self):
|
|
self.assertIn("BEA-4", rules(mutate(ROTATION, lambda s: s["executor"].update(principal_type="HUMAN"))))
|
|
self.assertIn("BEA-4", rules(mutate(ROTATION, lambda s: s["executor"].update(identity="admin"))))
|
|
self.assertIn("BEA-4", rules(mutate(ROTATION, lambda s: s["executor"].update(principal_type="AGENT"))))
|
|
harnessed = mutate(ROTATION, lambda s: s["executor"].update(principal_type="AGENT", harness_launched=True))
|
|
self.assertEqual(rules(harnessed), set())
|
|
|
|
def test_bea5_ceiling(self):
|
|
self.assertIn("BEA-5", rules(mutate(ROTATION, lambda s: s["scope_ceiling"].append(
|
|
{"mode": "GOVERN", "scope": "lldap", "environments": ["PRODUCTION"]}))))
|
|
# whole-directory ADMINISTER for a delegable step at medium risk
|
|
def widen(s):
|
|
s["scope_ceiling"][1].update(mode="ADMINISTER", scope="lldap")
|
|
for step in s["plan"]["steps"] + s["plan"]["revert_steps"]:
|
|
if step["requires"]["scope"].startswith("lldap/"):
|
|
step["requires"]["mode"] = "ADMINISTER"
|
|
self.assertIn("BEA-5", rules(mutate(ROTATION, widen)))
|
|
|
|
def test_bea6_least_privilege(self):
|
|
self.assertIn("BEA-6", rules(mutate(ROTATION, lambda s: s["scope_ceiling"].pop(3)))) # restart uncovered
|
|
self.assertIn("BEA-6", rules(mutate(ROTATION, lambda s: s["scope_ceiling"].append(
|
|
{"mode": "OPERATE", "scope": "k8s/sso/deployment/keycape", "environments": ["PRODUCTION"]})))) # unused
|
|
|
|
def test_bea7_root_class_needs_high_risk(self):
|
|
self.assertIn("BEA-7", rules(mutate(ROTATION, lambda s: s["plan"]["steps"][0].update({"class": "root-class"}))))
|
|
|
|
def test_bea8_blast_radius(self):
|
|
self.assertIn("BEA-8", rules(mutate(ROTATION, lambda s: s["plan"]["steps"][3].update(consumer="keycape"))))
|
|
def two_at_once(s):
|
|
s["blast_radius"].update(consumers=["authelia", "keycape"], max_concurrent=2)
|
|
self.assertIn("BEA-8", rules(mutate(ROTATION, two_at_once)))
|
|
|
|
def test_bea9_checks(self):
|
|
self.assertIn("BEA-9", rules(mutate(ROTATION, lambda s: s["plan"]["steps"][3].pop("checks_after"))))
|
|
self.assertIn("BEA-9", rules(mutate(ROTATION, lambda s: s["checks"][3].update(sha256="0" * 64))))
|
|
self.assertIn("BEA-9", rules(mutate(ROTATION, lambda s: s["checks"].append(dict(s["checks"][0], id="spare")))))
|
|
|
|
def test_bea10_revert(self):
|
|
self.assertIn("BEA-10", rules(mutate(ROTATION, lambda s: s["plan"]["steps"][1].pop("revert"))))
|
|
def irreversible(s):
|
|
s["plan"]["steps"][1].pop("revert")
|
|
s["plan"]["steps"][1]["irreversible"] = True
|
|
self.assertIn("BEA-10", rules(mutate(ROTATION, irreversible))) # medium risk
|
|
|
|
def test_bea11_stop_policy(self):
|
|
self.assertIn("BEA-11", rules(mutate(ROTATION, lambda s: s["stop_and_revert"].update(on_check_failure="stop"))))
|
|
self.assertIn("BEA-11", rules(mutate(VERIFY, lambda s: s["stop_and_revert"].update(on_check_failure="revert"))))
|
|
|
|
def test_bea12_budget(self):
|
|
self.assertIn("BEA-12", rules(mutate(ROTATION, lambda s: s["budget"].update(wall_time_seconds=8000))))
|
|
self.assertIn("BEA-12", rules(mutate(SUPERADMIN, lambda s: s["budget"].update(attempts=2))))
|
|
self.assertIn("BEA-12", rules(mutate(ROTATION, lambda s: s["budget"].update(wall_time_seconds=100))))
|
|
|
|
def test_bea14_secret_shaped_values(self):
|
|
self.assertIn("BEA-14", rules(mutate(ROTATION, lambda s: s.update(purpose="token hvs.abcdefghijklmnop"))))
|
|
|
|
def test_findings_never_echo_values(self):
|
|
marker = "hvs.SYNTHETICMARKER12345"
|
|
report = V.evaluate(mutate(ROTATION, lambda s: s.update(purpose=marker)), SCHEMA)
|
|
self.assertNotIn(marker, json.dumps(report))
|
|
|
|
|
|
class ApprovalTests(unittest.TestCase):
|
|
def check(self, document, rec):
|
|
return V.check_approval(document, rec)
|
|
|
|
def test_valid_record_admits_and_is_consumed(self):
|
|
report = self.check(ROTATION, record(ROTATION))
|
|
self.assertTrue(report["admitted"], report)
|
|
self.assertTrue(report["consumed"])
|
|
self.assertTrue(self.check(ROTATION, record(ROTATION, consumptions=[]))["admitted"])
|
|
|
|
def test_refusals(self):
|
|
other = mutate(ROTATION, lambda s: s.update(title="a different request"))
|
|
cases = {
|
|
"BEA-C1": record(ROTATION, carrier="email"),
|
|
"BEA-C2": record(ROTATION, admission_digest=V.admission_digest(other)),
|
|
"BEA-C3": record(ROTATION, approver={"name": "tegwick", "principal_type": "HUMAN", "assurance": "aal2"}),
|
|
"BEA-C4": record(ROTATION, approver={"name": "admin", "principal_type": "HUMAN", "assurance": "aal1"}),
|
|
"BEA-C5": record(ROTATION, approved_at="2026-11-03T12:00:00Z", consumptions=[]),
|
|
"BEA-C6": record(ROTATION, consumptions=[{"at": "2026-11-03T09:05:00Z", "by": "service:nk-bind-executor"},
|
|
{"at": "2026-11-03T09:40:00Z", "by": "service:nk-bind-executor"}]),
|
|
"BEA-C7": record(ROTATION, consumptions=[{"at": "2026-11-03T11:30:00Z", "by": "service:nk-bind-executor"}]),
|
|
}
|
|
for rule, rec in cases.items():
|
|
with self.subTest(rule=rule):
|
|
report = self.check(ROTATION, rec)
|
|
self.assertFalse(report["admitted"])
|
|
self.assertIn(rule, {f["rule"] for f in report["findings"]})
|
|
|
|
def test_consumed_by_someone_else(self):
|
|
rec = record(ROTATION, consumptions=[{"at": "2026-11-03T09:05:00Z", "by": "service:other"}])
|
|
self.assertIn("BEA-C7", {f["rule"] for f in self.check(ROTATION, rec)["findings"]})
|
|
|
|
|
|
class CliTests(unittest.TestCase):
|
|
def test_cli(self):
|
|
import contextlib
|
|
import io
|
|
import tempfile
|
|
out = io.StringIO()
|
|
with contextlib.redirect_stdout(out):
|
|
self.assertEqual(V.main(["validate", str(EXAMPLES / "bind-rotation-svc-authelia.yaml")]), 0)
|
|
self.assertEqual(V.main(["validate", str(TOOL / "fixtures/invalid/human-intrinsic-step.yaml")]), 1)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = pathlib.Path(tmp) / "record.json"
|
|
path.write_text(json.dumps(record(ROTATION)))
|
|
self.assertEqual(V.main(["check-approval", str(EXAMPLES / "bind-rotation-svc-authelia.yaml"), str(path)]), 0)
|
|
self.assertEqual(V.main(["digest", str(EXAMPLES / "bind-rotation-svc-authelia.yaml")]), 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|