tegwick
a3bab9210b
Founder clarifications: the SI-14 token ceiling is 60 minutes (3600 s); ld-admin is for LLDAP only and the name admin remains the human platform administrator's wherever it appears. The LLDAP rehearsal is now introducing ld-admin by configuration and demoting admin, not renaming. The validator rejects a service-admin named admin. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 328925@bnt-lap001 Assistant-Session: 14b795fd-c632-4c01-8d28-e7e490c7b0ed
302 lines
16 KiB
Python
302 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import copy
|
|
import datetime as dt
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
SPEC = importlib.util.spec_from_file_location("service_identity_validate", HERE.parent / "validate.py")
|
|
assert SPEC and SPEC.loader
|
|
V = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(V)
|
|
FIXTURES = HERE.parent / "fixtures"
|
|
DECLARATIONS = HERE.parent / "declarations"
|
|
SCHEMA = V.load_schema()
|
|
NOW = dt.date(2026, 10, 24)
|
|
|
|
|
|
def identity(**over):
|
|
base = {"name": "svc-x", "classes": ["bind"], "purpose": "Read-only lookup", "privilege_level": "read",
|
|
"interactive_login": False, "second_factor": "not_applicable", "custody": ["delivered-secret"],
|
|
"consumers": [{"name": "x", "kind": "service"}], "owner": "Owners",
|
|
"rotation": {"path": "Rotate with the consumer", "coupling": ["x"]}, "status": "active"}
|
|
base.update(over)
|
|
return base
|
|
|
|
|
|
def declaration(*identities, exceptions=None, complete=True, confirmed=True):
|
|
doc = {"schema_version": "0.1", "service": "example", "owner": "Owners", "declared_at": "2026-10-24",
|
|
"attestation": {"authored_by": "test", "basis": ["test"], "owner_confirmed": confirmed},
|
|
"complete": complete, "identities": list(identities or [identity()])}
|
|
if exceptions:
|
|
doc["exceptions"] = exceptions
|
|
return doc
|
|
|
|
|
|
def exception(**over):
|
|
base = {"id": "SI-X9", "rules": ["SI-3"], "identity": "svc-x", "reason": "Reason for the test",
|
|
"owner": "Owners", "expires": "2027-01-31", "compensating_control": "A compensating control"}
|
|
base.update(over)
|
|
return base
|
|
|
|
|
|
def result_of(doc, as_of=NOW, name=None):
|
|
assert V.structural_errors(doc, SCHEMA) == [], V.structural_errors(doc, SCHEMA)
|
|
report = V.evaluate(doc, as_of)
|
|
found = [i for i in report["identities"] if name in (None, i["identity"])]
|
|
return found[0]
|
|
|
|
|
|
class StructureTests(unittest.TestCase):
|
|
def test_conformant_fixture_is_valid_and_conformant(self):
|
|
doc = json.loads((FIXTURES / "valid/conformant.json").read_text())
|
|
self.assertEqual(V.structural_errors(doc, SCHEMA), [])
|
|
self.assertEqual({i["result"] for i in V.evaluate(doc, NOW)["identities"]}, {"conformant"})
|
|
|
|
def test_secret_shaped_value_is_rejected_without_echo(self):
|
|
doc = json.loads((FIXTURES / "invalid/secret-value.json").read_text())
|
|
errors = V.structural_errors(doc, SCHEMA)
|
|
self.assertEqual(errors, ["value looks like secret material"])
|
|
self.assertNotIn("hvs.", " ".join(errors))
|
|
|
|
def test_unknown_field_and_class_are_rejected(self):
|
|
doc = declaration(identity(password="x"))
|
|
self.assertTrue(V.structural_errors(doc, SCHEMA))
|
|
doc = declaration(identity(classes=["wizard"]))
|
|
self.assertTrue(V.structural_errors(doc, SCHEMA))
|
|
|
|
def test_duplicate_identity_and_undeclared_exception_target(self):
|
|
self.assertIn("duplicate identity name", V.structural_errors(declaration(identity(), identity()), SCHEMA))
|
|
doc = declaration(identity(), exceptions=[exception(identity="ghost")])
|
|
self.assertIn("exception names an undeclared identity", V.structural_errors(doc, SCHEMA))
|
|
|
|
def test_legacy_identity_needs_a_disposition(self):
|
|
self.assertTrue(V.structural_errors(declaration(identity(status="legacy")), SCHEMA))
|
|
ok = identity(status="legacy", disposition={"action": "retire", "owner": "Owners", "date": "2026-12-31"})
|
|
self.assertEqual(V.structural_errors(declaration(ok), SCHEMA), [])
|
|
|
|
|
|
class RuleTests(unittest.TestCase):
|
|
def rules(self, **over):
|
|
return result_of(declaration(identity(**over)))["rules"]
|
|
|
|
def test_si1_more_than_one_class(self):
|
|
self.assertIn("SI-1", self.rules(name="admin", classes=["human", "bind"]))
|
|
|
|
def test_si2_person_and_service_share_a_credential(self):
|
|
consumers = [{"name": "p", "kind": "person"}, {"name": "s", "kind": "service"}]
|
|
self.assertIn("SI-2", self.rules(consumers=consumers))
|
|
|
|
def test_si3_bind_must_be_read_only_non_interactive_and_unshared(self):
|
|
self.assertIn("SI-3", self.rules(privilege_level="admin"))
|
|
self.assertIn("SI-3", self.rules(interactive_login=True))
|
|
shared = [{"name": "a", "kind": "service"}, {"name": "b", "kind": "service"}]
|
|
self.assertIn("SI-3", self.rules(consumers=shared))
|
|
self.assertEqual(self.rules(), [])
|
|
|
|
def test_si8_admin_without_enforced_second_factor_or_assurance(self):
|
|
admin = dict(name="ex-admin", classes=["service-admin"], privilege_level="admin", interactive_login=True,
|
|
consumers=[{"name": "p", "kind": "person"}], custody=["personal-safe"])
|
|
self.assertIn("SI-8", self.rules(second_factor="not_enforced", **admin))
|
|
self.assertEqual(self.rules(second_factor="required", **admin), [])
|
|
self.assertIn("SI-8", self.rules(second_factor="required", grants_admin_policy=True, **admin))
|
|
self.assertEqual(self.rules(second_factor="required", grants_admin_policy=True, assurance_claim_bound=True, **admin), [])
|
|
|
|
def test_si9_a_person_safe_must_not_hold_a_service_credential(self):
|
|
self.assertIn("SI-9", self.rules(custody=["delivered-secret", "personal-safe"]))
|
|
|
|
def test_si6_break_glass_is_offline_and_has_no_service_consumer(self):
|
|
bg = dict(name="platform-root", classes=["break-glass"], privilege_level="admin", interactive_login=True,
|
|
second_factor="required", consumers=[{"name": "p", "kind": "person"}])
|
|
self.assertEqual(self.rules(custody=["offline-key"], **bg), [])
|
|
self.assertIn("SI-6", self.rules(custody=["personal-safe"], **bg))
|
|
self.assertIn("SI-6", self.rules(custody=["offline-key"], **{**bg, "consumers": [{"name": "s", "kind": "service"}]}))
|
|
|
|
def test_si12_recovery_identity_must_be_in_the_crisis_list(self):
|
|
self.assertIn("SI-12", self.rules(recovery={"needed": True, "works_with_sso_down": True, "in_crisis_list": False}))
|
|
|
|
def test_naming_is_reserved_by_class(self):
|
|
self.assertIn("NAMING", self.rules(name="lookup"))
|
|
self.assertIn("NAMING", self.rules(name="x-admin", classes=["human"], second_factor="required", interactive_login=True,
|
|
consumers=[{"name": "p", "kind": "person"}]))
|
|
# A federated role holds no credential and cannot be mistaken for an account.
|
|
self.assertNotIn("NAMING", self.rules(name="group-admin", classes=["human"], second_factor="required", interactive_login=True,
|
|
custody=["none"], consumers=[{"name": "p", "kind": "person"}]))
|
|
admin = dict(classes=["service-admin"], privilege_level="admin", second_factor="required", interactive_login=True,
|
|
consumers=[{"name": "p", "kind": "person"}], custody=["personal-safe"])
|
|
self.assertIn("NAMING", self.rules(name="administrator", **admin))
|
|
self.assertIn("NAMING", self.rules(name="admin", **admin)) # `admin` is the human's name everywhere
|
|
self.assertEqual(self.rules(name="ld-admin", **admin), [])
|
|
|
|
def test_overdue_legacy_disposition_is_a_violation(self):
|
|
legacy = identity(status="legacy", name="old-admin", classes=["service-admin"], privilege_level="admin",
|
|
second_factor="required", interactive_login=True, consumers=[],
|
|
disposition={"action": "defer", "owner": "Owners", "date": "2026-10-01"})
|
|
self.assertIn("SI-11", result_of(declaration(legacy))["rules"])
|
|
legacy["disposition"]["date"] = "2026-12-31"
|
|
self.assertNotIn("SI-11", result_of(declaration(legacy))["rules"])
|
|
|
|
|
|
class AgentTests(unittest.TestCase):
|
|
def agent(self, **agent_over):
|
|
agent = {"mode": "autonomous", "presents_principal_type": "agent", "max_token_ttl_seconds": 900,
|
|
"boundary_policy": "agent-high-risk-boundary"}
|
|
agent.update(agent_over)
|
|
return identity(name="coding-agent", classes=["agent"], agent=agent, privilege_level="read",
|
|
custody=["none"], consumers=[{"name": "agent-host", "kind": "service"}])
|
|
|
|
def test_agent_class_needs_the_agent_object(self):
|
|
doc = declaration(identity(name="a", classes=["agent"], custody=["none"]))
|
|
self.assertTrue(V.structural_errors(doc, SCHEMA))
|
|
|
|
def test_conformant_agent(self):
|
|
self.assertEqual(result_of(declaration(self.agent()))["result"], "conformant")
|
|
|
|
def test_agent_presenting_as_a_service_breaks_si14(self):
|
|
r = result_of(declaration(self.agent(presents_principal_type="service")))
|
|
self.assertEqual((r["result"], r["rules"]), ("non-conformant", ["SI-14"]))
|
|
|
|
def test_agent_token_lifetime_above_the_ceiling_breaks_si14(self):
|
|
self.assertIn("SI-14", result_of(declaration(self.agent(max_token_ttl_seconds=7200)))["rules"])
|
|
|
|
def test_unknown_agent_mode_is_unknown_not_a_pass(self):
|
|
r = result_of(declaration(self.agent(mode="unknown")))
|
|
self.assertEqual((r["result"], r["unknown"]), ("unknown", ["agent.mode"]))
|
|
|
|
def test_agent_must_not_live_in_a_persons_safe_or_serve_a_person(self):
|
|
doc = declaration({**self.agent(), "custody": ["personal-safe"],
|
|
"consumers": [{"name": "p", "kind": "person"}]})
|
|
self.assertLessEqual({"SI-9", "SI-2"}, set(result_of(doc)["rules"]))
|
|
|
|
|
|
class ResultTests(unittest.TestCase):
|
|
def test_unknown_is_a_finding_for_every_class(self):
|
|
for fact, value in (("privilege_level", "unknown"), ("custody", ["unknown"])):
|
|
r = result_of(declaration(identity(classes=["workload"], name="client", **{fact: value})))
|
|
self.assertEqual(r["result"], "unknown")
|
|
self.assertEqual(r["unknown"], [fact])
|
|
|
|
def test_violation_outranks_unknown(self):
|
|
r = result_of(declaration(identity(privilege_level="unknown", interactive_login=True)))
|
|
self.assertEqual(r["result"], "non-conformant")
|
|
|
|
def test_planned_identity_is_not_counted_as_conformant(self):
|
|
self.assertEqual(result_of(declaration(identity(status="planned")))["result"], "planned")
|
|
|
|
def test_live_exception_excepts_and_expired_one_does_not(self):
|
|
doc = declaration(identity(privilege_level="admin"), exceptions=[exception()])
|
|
r = result_of(doc)
|
|
self.assertEqual((r["result"], r["excepted"]), ("excepted", {"SI-3": "SI-X9"}))
|
|
r = result_of(doc, as_of=dt.date(2027, 2, 1))
|
|
self.assertEqual(r["result"], "non-conformant")
|
|
self.assertIn("SI-EXC-EXPIRED", r["rules"])
|
|
|
|
def test_exception_for_another_rule_does_not_cover_the_violation(self):
|
|
doc = declaration(identity(privilege_level="admin"), exceptions=[exception(rules=["SI-9"])])
|
|
self.assertEqual(result_of(doc)["result"], "non-conformant")
|
|
|
|
def test_stale_exception_and_unconfirmed_declaration_are_advisories(self):
|
|
doc = declaration(identity(), exceptions=[exception()], complete=False, confirmed=False)
|
|
advisories = V.evaluate(doc, NOW)["advisories"]
|
|
self.assertTrue(any("stale exception" in a for a in advisories))
|
|
self.assertTrue(any("not asserted complete" in a for a in advisories))
|
|
self.assertTrue(any("not confirmed" in a for a in advisories))
|
|
|
|
def test_evaluation_date_is_explicit(self):
|
|
doc = declaration(identity(privilege_level="admin"), exceptions=[exception(expires="2026-10-30")])
|
|
self.assertEqual(result_of(doc, as_of=dt.date(2026, 10, 30))["result"], "excepted")
|
|
self.assertEqual(result_of(doc, as_of=dt.date(2026, 10, 31))["result"], "non-conformant")
|
|
|
|
|
|
class CheckedInDeclarations(unittest.TestCase):
|
|
def paths(self):
|
|
found = sorted(DECLARATIONS.glob("*.yaml"))
|
|
self.assertGreaterEqual(len(found), 7)
|
|
return found
|
|
|
|
def test_all_declarations_are_valid(self):
|
|
reports, rejected = V.run(self.paths(), NOW)
|
|
self.assertEqual(rejected, [])
|
|
self.assertEqual({r["service"] for r in reports} >= {"lldap", "keycape", "privacyidea", "openbao", "platform"}, True)
|
|
|
|
def test_the_coding_agent_role_is_a_finding_against_the_iam_profile(self):
|
|
reports, _ = V.run(self.paths(), NOW)
|
|
agent = [i for r in reports for i in r["identities"] if i["identity"] == "coding-agent"][0]
|
|
self.assertEqual((agent["result"], agent["rules"]), ("non-conformant", ["SI-14"]))
|
|
|
|
def test_the_shared_admin_credential_is_excepted_today_and_fails_after_expiry(self):
|
|
def admin(as_of):
|
|
reports, _ = V.run(self.paths(), as_of)
|
|
return [i for r in reports for i in r["identities"] if r["service"] == "lldap" and i["identity"] == "admin"][0]
|
|
today = admin(NOW)
|
|
self.assertEqual(today["result"], "excepted")
|
|
self.assertEqual(set(today["excepted"]), {"SI-1", "SI-2", "SI-3", "SI-9", "NAMING"})
|
|
after = admin(dt.date(2027, 2, 1))
|
|
self.assertEqual(after["result"], "non-conformant")
|
|
|
|
def test_no_declaration_claims_owner_confirmation_yet(self):
|
|
for path in self.paths():
|
|
doc = V.load_declaration(path)
|
|
self.assertFalse(doc["attestation"]["owner_confirmed"], path.name)
|
|
|
|
|
|
class CommandLineTests(unittest.TestCase):
|
|
def run_cli(self, *argv):
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
|
code = V.main(list(argv))
|
|
return code, out.getvalue(), err.getvalue()
|
|
|
|
def test_invalid_input_reports_only_a_count_and_exits_1(self):
|
|
code, out, err = self.run_cli(str(FIXTURES / "invalid/secret-value.json"))
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("rejected (1 errors)", err)
|
|
self.assertNotIn("hvs.", out + err)
|
|
|
|
def test_valid_fixture_exits_0_and_prints_summary(self):
|
|
code, out, _ = self.run_cli(str(FIXTURES / "valid/conformant.json"), "--as-of", "2026-10-24")
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("summary: 0 non-conformant, 0 unknown, 0 excepted, 0 planned, 2 conformant", out)
|
|
|
|
def test_fail_on_findings_exits_2_for_unknowns(self):
|
|
paths = [str(p) for p in sorted(DECLARATIONS.glob("*.yaml"))]
|
|
code, _, _ = self.run_cli(*paths, "--as-of", "2026-10-24", "--fail-on-findings")
|
|
self.assertEqual(code, 2)
|
|
code, _, _ = self.run_cli(*paths, "--as-of", "2026-10-24")
|
|
self.assertEqual(code, 0)
|
|
|
|
def test_json_output_has_counts_and_no_values(self):
|
|
code, out, _ = self.run_cli(str(FIXTURES / "valid/conformant.json"), "--json", "--as-of", "2026-10-24")
|
|
self.assertEqual(code, 0)
|
|
data = json.loads(out)
|
|
self.assertEqual(data["counts"]["conformant"], 2)
|
|
|
|
def test_bad_as_of_is_refused(self):
|
|
code, _, err = self.run_cli(str(FIXTURES / "valid/conformant.json"), "--as-of", "yesterday")
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("invalid --as-of", err)
|
|
|
|
def test_more_than_one_break_glass_identity_is_an_advisory(self):
|
|
def bg(name):
|
|
return declaration(identity(name=name, classes=["break-glass"], privilege_level="admin", interactive_login=True,
|
|
second_factor="required", custody=["offline-key"],
|
|
consumers=[{"name": "p", "kind": "person"}]))
|
|
with tempfile.TemporaryDirectory() as d:
|
|
paths = []
|
|
for n in ("platform-root", "king"):
|
|
p = pathlib.Path(d) / f"{n}.json"
|
|
p.write_text(json.dumps(bg(n)))
|
|
paths.append(p)
|
|
reports, rejected = V.run(paths, NOW)
|
|
self.assertEqual(rejected, [])
|
|
self.assertTrue(any("more than one break-glass" in a for r in reports for a in r["advisories"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|