net-kingdom/tools/bounded-execution-admission/tests/test_validate.py

"""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]["scope"] = "lldap"
        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()