From df48ee96e012fd460e234aede63b1db41ae25070 Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 9 Sep 2026 14:40:05 +0200 Subject: [PATCH] feat: assert flex-auth ceiling keys are declared, not assumed flex-auth fixed enrichment so registry facts beat caller-supplied ones (FLEX-DEC-2026-012) and asked each consumer to confirm the ceiling and allowlist keys are actually declared -- the fix wins only where the registry HAS a value, and a manifest omitting max_ttl_hours hands that ceiling back to the caller. Confirmed, and made durable rather than read once. scripts/check_flex_auth_manifest_coverage.py audits both ways a ceiling gets handed back: an actor with no manifest resource at all (warden sign names ssh-cert:actor/ whether or not the snapshot was rebuilt -- an honour-system step in SCOPE.md), and a resource missing one of the seven keys. A null is treated as absent, because for enrichment it is. Also asserts the property their exploitability assessment rested on and nothing here held: ops-warden sends no resource.attributes. It was true when they read it, secrets-engine sends them on every request, and it was one refactor from silently stopping being true. Current state: no gap. 4 actors, 4 resources, all seven keys declared. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013EPuTc18FjU5WFqoSEKH3C Assistant: claude-code Assistant-Model: opus Assistant-Process: 1276224@bnt-lap001 Assistant-Session: 426ec497-e1c4-4dd3-b417-dfce1ca1dbc3 --- scripts/check_flex_auth_manifest_coverage.py | 161 +++++++++++++++++++ tests/test_flex_auth_registry.py | 83 ++++++++++ tests/test_policy.py | 44 +++++ 3 files changed, 288 insertions(+) create mode 100644 scripts/check_flex_auth_manifest_coverage.py diff --git a/scripts/check_flex_auth_manifest_coverage.py b/scripts/check_flex_auth_manifest_coverage.py new file mode 100644 index 0000000..755ba02 --- /dev/null +++ b/scripts/check_flex_auth_manifest_coverage.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Assert every ssh-certificate resource declares its ceilings in the manifest. + +Read-only. Reads `inventory.yaml` and the flex-auth registry snapshot; touches +no network, no OpenBao, and no secret material. + +Why this exists (FLEX-DEC-2026-012, 2026-09-07). flex-auth's enrichment used to +overlay registry facts additive-if-absent, so a caller-supplied value for a key +won and the registry's ceiling never applied. That is fixed on their side: +registry facts now win. But they win only where the registry HAS a value — + + "A resource whose manifest omits max_ttl_hours hands that ceiling back to + the caller, and registering the resource is not sufficient; the specific + key must be present." + +So there are two ways to hand a ceiling back, and this check covers both: + + 1. An actor with no manifest resource at all. `warden sign` names + `ssh-cert:actor/` for any actor in inventory, whether or not the + snapshot was regenerated. Adding an actor and forgetting to rebuild is an + honour-system step in `SCOPE.md`, and honour-system steps are what + `ADR-0004`'s WARDEN_AGENT_ID marker taught us to stop relying on. + 2. A manifest resource missing one of the ceiling or allowlist keys. The + builder emits all of them today; this asserts it stays true, including for + resources added by hand or by a future code path. + +ops-warden sends no `resource.attributes` on a CheckRequest (`src/warden/policy.py`), +which is why FLEX-DEC-2026-012 was defence-in-depth rather than a live hole for +this repo. That property is asserted separately in `tests/test_policy.py`; this +check covers the half that survives it — if anything ever does reach a request +field, the ceiling it would have to beat must actually exist. + +Usage: + python scripts/check_flex_auth_manifest_coverage.py \ + --inventory examples/inventory.seed.yaml \ + --registry registry/flex-auth/production_registry_snapshot.json [--json] + +Exit: 0 covered, 2 coverage gap, 1 usage/IO error. +""" +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +import yaml + +#: Every key ops-warden's shipped policy package branches on as a ceiling or an +#: allowlist. Named by flex-auth in FLEX-DEC-2026-012; kept here rather than +#: derived from the package so a package edit that drops a branch does not +#: silently shrink what this check requires. +REQUIRED_CEILING_KEYS = ( + "actor_id", + "actor_type", + "allowed_principals", + "allowed_subjects", + "max_ttl_hours", + "security_zone", + "security_zone_admission", +) + +RESOURCE_TYPE = "ssh-certificate" + + +def _resources(registry: dict[str, Any]) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + for manifest in registry.get("resource_manifests") or []: + out.extend(manifest.get("resources") or []) + return out + + +def audit(inventory: dict[str, Any], registry: dict[str, Any]) -> dict[str, Any]: + actors = sorted((inventory.get("actors") or {}).keys()) + resources = _resources(registry) + by_id = {str(r.get("id")): r for r in resources} + + unregistered: list[str] = [] + missing_keys: list[dict[str, Any]] = [] + + for name in actors: + if f"ssh-cert:actor/{name}" not in by_id: + unregistered.append(name) + + # Check every ssh-certificate resource in the manifest, not only the ones an + # inventory actor maps to: a resource flex-auth can be asked about is one it + # holds, regardless of where it came from. + for resource in resources: + if str(resource.get("type")) != RESOURCE_TYPE: + continue + attributes = resource.get("attributes") or {} + absent = [ + key for key in REQUIRED_CEILING_KEYS + # A null is not a declaration: flex-auth overlays a registry value + # only where one exists, so `key: null` hands the ceiling back + # exactly as an absent key does. + if attributes.get(key) is None + ] + if absent: + missing_keys.append({"resource": str(resource.get("id")), "missing": absent}) + + # Not a security defect — a stale resource has ceilings, it just has no + # actor. Reported so drift is visible rather than accumulating silently. + orphaned = sorted( + rid for rid, r in by_id.items() + if str(r.get("type")) == RESOURCE_TYPE + and rid.removeprefix("ssh-cert:actor/") not in actors + ) + + return { + "actors": len(actors), + "resources": sum(1 for r in resources if str(r.get("type")) == RESOURCE_TYPE), + "required_keys": list(REQUIRED_CEILING_KEYS), + "unregistered_actors": unregistered, + "resources_missing_keys": missing_keys, + "orphaned_resources": orphaned, + "covered": not unregistered and not missing_keys, + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--inventory", default="examples/inventory.seed.yaml", type=Path) + parser.add_argument( + "--registry", + default="registry/flex-auth/production_registry_snapshot.json", + type=Path, + ) + parser.add_argument("--json", action="store_true", dest="as_json") + args = parser.parse_args(argv) + + try: + inventory = yaml.safe_load(args.inventory.read_text()) or {} + registry = json.loads(args.registry.read_text()) + except (OSError, ValueError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + + report = audit(inventory, registry) + report["inventory"] = str(args.inventory) + report["registry"] = str(args.registry) + + if args.as_json: + print(json.dumps(report, indent=2)) + else: + print(f"inventory : {args.inventory} ({report['actors']} actors)") + print(f"registry : {args.registry} ({report['resources']} ssh-certificate resources)") + for name in report["unregistered_actors"]: + print(f" MISSING {name} — no manifest resource; every ceiling is caller-supplied") + for row in report["resources_missing_keys"]: + print(f" UNDECLARED {row['resource']} — {', '.join(row['missing'])}") + for rid in report["orphaned_resources"]: + print(f" orphaned {rid} — manifest resource with no inventory actor") + print("covered" if report["covered"] else "NOT COVERED") + + return 0 if report["covered"] else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_flex_auth_registry.py b/tests/test_flex_auth_registry.py index c329d27..70dd26f 100644 --- a/tests/test_flex_auth_registry.py +++ b/tests/test_flex_auth_registry.py @@ -81,3 +81,86 @@ def test_compiler_joins_explicit_workload_reference_to_resolved_zone(tmp_path): assert attrs["security_zone"] == "z2-continuity" assert attrs["security_zone_admission"] == "satisfied" assert attrs["security_zone_revision"] == "sha256:zone-revision" + + +# --- manifest ceiling coverage (FLEX-DEC-2026-012) ---------------------------- +# +# flex-auth fixed enrichment so registry facts beat caller-supplied ones, then +# asked every consumer to confirm the ceiling keys are actually declared: the +# fix wins only where the registry HAS a value. These assert the two ways a +# ceiling gets handed back to the caller. + +import importlib.util # noqa: E402 + +import pytest # noqa: E402 + +_REPO = ROOT +_spec = importlib.util.spec_from_file_location( + "check_flex_auth_manifest_coverage", + _REPO / "scripts" / "check_flex_auth_manifest_coverage.py", +) +_coverage = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_coverage) +REQUIRED_CEILING_KEYS = _coverage.REQUIRED_CEILING_KEYS +audit = _coverage.audit + +_build_spec = importlib.util.spec_from_file_location("build_flex_auth_registry", SCRIPT) +_builder = importlib.util.module_from_spec(_build_spec) +_build_spec.loader.exec_module(_builder) +build_registry = _builder.build_registry + + +def _repo_inventory_and_registry(): + inventory = yaml.safe_load((_REPO / "examples" / "inventory.seed.yaml").read_text()) + registry = json.loads( + (_REPO / "registry" / "flex-auth" / "production_registry_snapshot.json").read_text() + ) + return inventory, registry + + +def test_shipped_snapshot_declares_every_ceiling_key(): + """The confirmation flex-auth asked for, as a test rather than a reading.""" + report = audit(*_repo_inventory_and_registry()) + assert report["unregistered_actors"] == [] + assert report["resources_missing_keys"] == [] + assert report["covered"] is True + + +def test_actor_with_no_manifest_resource_is_a_coverage_gap(): + """`warden sign` names ssh-cert:actor/ whether or not the snapshot was rebuilt.""" + inventory, registry = _repo_inventory_and_registry() + inventory["actors"]["adm-added-after-the-last-build"] = { + "type": "adm", "principals": ["adm-full"], "ttl_hours": 48, + } + report = audit(inventory, registry) + assert report["unregistered_actors"] == ["adm-added-after-the-last-build"] + assert report["covered"] is False + + +@pytest.mark.parametrize("key", REQUIRED_CEILING_KEYS) +def test_dropping_any_single_ceiling_key_is_a_coverage_gap(key): + inventory, registry = _repo_inventory_and_registry() + resource = registry["resource_manifests"][0]["resources"][0] + resource["attributes"].pop(key) + report = audit(inventory, registry) + assert report["resources_missing_keys"] == [ + {"resource": resource["id"], "missing": [key]} + ] + assert report["covered"] is False + + +@pytest.mark.parametrize("key", REQUIRED_CEILING_KEYS) +def test_a_null_ceiling_is_not_a_declaration(key): + """flex-auth overlays a registry value only where one exists.""" + inventory, registry = _repo_inventory_and_registry() + resource = registry["resource_manifests"][0]["resources"][0] + resource["attributes"][key] = None + report = audit(inventory, registry) + assert report["covered"] is False + + +def test_builder_output_is_covered_by_construction(tmp_path): + """Every resource the builder emits carries the full ceiling set.""" + inventory = yaml.safe_load((_REPO / "examples" / "inventory.seed.yaml").read_text()) + report = audit(inventory, build_registry(inventory)) + assert report["covered"] is True diff --git a/tests/test_policy.py b/tests/test_policy.py index 0e2f948..8257e23 100644 --- a/tests/test_policy.py +++ b/tests/test_policy.py @@ -307,3 +307,47 @@ def test_advisory_decision_is_recorded_and_does_not_block(tmp_path): assert check_sign_policy(cfg, spec) == "decision:advisory" assert spec.policy_zone == "unknown" assert spec.policy_outcome == "audit_only" + + +def test_check_request_asserts_no_resource_attributes(tmp_path, monkeypatch): + """The property FLEX-DEC-2026-012 turned on, asserted rather than assumed. + + flex-auth's enrichment used to overlay registry facts additive-if-absent, so + a caller-supplied `resource.attributes` value won and the registry ceiling + never applied. Their exploitability assessment rested on ops-warden sending + no `resource.attributes` at all -- true when they read it, and nothing here + held it true. secrets-engine sends them on every request, so this is a + property of this code rather than of the protocol. + + Ceilings must arrive from the registry. If a future change ever needs to send + an attribute, this test is the place that argument gets made. + """ + from warden import policy as policy_mod + + pubkey = tmp_path / "id.pub" + pubkey.write_text("ssh-ed25519 AAAA test\n") + cfg = PolicyConfig(flex_auth_url="http://flex-auth.test") + seen = {} + + class _Response: + status_code = 200 + + def raise_for_status(self): + return None + + def json(self): + return {"effect": "allow", "id": "decision:49350f1064f674d7"} + + def fake_post(url, json=None, headers=None, timeout=None): + seen["body"] = json + return _Response() + + monkeypatch.setattr(policy_mod.httpx, "post", fake_post) + policy_mod.check_sign_policy(cfg, _spec(pubkey)) + + assert "attributes" not in seen["body"]["resource"] + # The requested TTL is a policy input (ttl_out_of_bounds is denied against + # the registry ceiling), so it must travel as context, never as a resource + # attribute that would be compared against itself. + assert seen["body"]["context"]["ttl_hours"] == 24 + assert set(seen["body"]["resource"]) == {"id", "type", "system", "tenant"}