diff --git a/README.md b/README.md index ee158ac..184058f 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,7 @@ SECRETS_ENGINE_HUB_URL="" bash scripts/demo-e2e.sh - KeyCape service-auth consumer boundary: [docs/service-auth.md](docs/service-auth.md) - Approval consume-before-OpenBao (GH-DEC-2026-003): [docs/approval-consumption.md](docs/approval-consumption.md) - OpenBao JWT login contract (engine consumer): [docs/openbao-jwt-login.md](docs/openbao-jwt-login.md) +- Secret-use evidence surface: [docs/secret-use-evidence-contract.md](docs/secret-use-evidence-contract.md) The implementation is a Python package (`src/secrets_engine/`). OpenBao is reached only through the `bao` CLI adapter (`openbao.py`); the rest of the code diff --git a/docs/cli.md b/docs/cli.md index e626981..e0bfa86 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -64,6 +64,7 @@ secrets-engine lifecycle suspend [--dry-run] secrets-engine lifecycle deactivate [--dry-run] secrets-engine lifecycle destroy [--dry-run] [--confirm-destroy ] secrets-engine audit [--json] +secrets-engine secret-use snapshot [--catalog-id ID] [--json] ``` `policy publication` resolves a lane's effective publication scope and the env diff --git a/docs/secret-use-evidence-contract.md b/docs/secret-use-evidence-contract.md index ae94cf7..0bf45ea 100644 --- a/docs/secret-use-evidence-contract.md +++ b/docs/secret-use-evidence-contract.md @@ -1,8 +1,9 @@ -# Secret-use evidence engine contract (proposed) +# Secret-use evidence engine contract -Status: **proposed**. `layer.yaml` owner_status remains `proposed` until this -surface ships. kings-guard must assent or contest before it is treated as an -observation input. +Status: **shipped** as `secrets-engine secret-use snapshot`. Observation-input +admission remains **proposed** until kings-guard consumes it. `layer.yaml` +`owner_status` stays `proposed` until that admission. Completeness is not +claimed. Standard: NetKingdom Security Layer Model v0.7 §9.6. Companion: `net-kingdom/SECURITY-COMPANION.md` §6. @@ -14,7 +15,9 @@ mount, rotation, and delivery-session metadata, so kings-guard can evaluate secret-abuse posture without a vault client. `secrets-engine route` and `secrets-engine audit` are operator summaries over -local JSONL. They are not this surface. +local JSONL. They are not this surface. The shipped command is +`secrets-engine secret-use snapshot [--catalog-id ID] [--json]`. It reads the +catalog and local evidence only. It never contacts OpenBao. ## Bound (normative) @@ -53,9 +56,18 @@ exception prose. ## Freshness Every row carries `as_of`. There is no cached authorization verdict. A consumer -must not treat a stale snapshot as a standing allow. Recommended maximum age -for posture evaluation is the heartbeat interval declared in -`evidence-classification.yaml` (`1d`) until a tighter contract is assented. +must not treat a stale snapshot as a standing allow. Cadence for load-bearing +classes is the heartbeat declared in `evidence-classification.yaml`: + +```text +form: heartbeat +interval: 1d +command: secrets-engine evidence heartbeat +``` + +The snapshot envelope repeats that cadence. Evidence-derived fields (`ready`, +session, stance, decision id) appear only when a local record exists. Omission +is not non-occurrence. ## Destination diff --git a/layer.yaml b/layer.yaml index e42d8f3..fed1ba5 100644 --- a/layer.yaml +++ b/layer.yaml @@ -90,11 +90,10 @@ proposed_capabilities: owner_status: proposed contract: docs/secret-use-evidence-contract.md blocked_on: >- - Contract published; kings-guard assented 2026-09-01 (message - 75ebd2cc-a166-4676-94aa-deef2791c0c9) and will not treat this as an - observation input until the surface ships and publishes its cadence - declaration. Local JSONL, the load-bearing outbox, and `route`/`audit` - are not this observation API. Completeness is not claimed. + Surface shipped as `secrets-engine secret-use snapshot`; cadence is the + declared 1d heartbeat. kings-guard assented 2026-09-01 (message + 75ebd2cc-a166-4676-94aa-deef2791c0c9) and has not yet admitted the + snapshot as an observation input. Completeness is not claimed. review: "2026-11-28" consequence: >- kings-guard secret-abuse posture stays fixture-driven. diff --git a/src/secrets_engine/cli.py b/src/secrets_engine/cli.py index 0aaedc7..6523960 100644 --- a/src/secrets_engine/cli.py +++ b/src/secrets_engine/cli.py @@ -15,6 +15,7 @@ Command surface (FR7): session revoke --accessor-file F [--stage stage] lifecycle suspend|deactivate|destroy audit [--json] + secret-use snapshot [--catalog-id ID] [--json] evidence heartbeat|drain|classify Every privileged action is decision-gated and writes non-secret evidence. @@ -617,6 +618,35 @@ def cmd_lifecycle(cfg: Config, args) -> int: return 0 +def cmd_secret_use_snapshot(cfg: Config, args) -> int: + """Read-only secret-use evidence snapshot. Never contacts OpenBao.""" + import json + + from secrets_engine.secret_use import snapshot + + payload = snapshot( + cfg.catalog_dir, + cfg.evidence_dir, + catalog_id=getattr(args, "catalog_id", "") or "", + ) + if args.json: + print(json.dumps(payload, indent=2, sort_keys=True)) + else: + print( + f"secret-use {payload['as_of']} lanes={len(payload['lanes'])} " + f"completeness_claimed={payload['completeness_claimed']} " + f"cadence={payload['cadence'].get('form')}/{payload['cadence'].get('interval')}" + ) + for lane in payload["lanes"]: + ready = lane.get("ready") + ready_s = "unknown" if ready is None else str(ready).lower() + print( + f" {lane['catalog_id']:32s} stage={lane['stage']:5s} " + f"ready={ready_s} kind={lane.get('evidence_kind', '-')}" + ) + return 0 + + def cmd_audit(cfg: Config, args) -> int: """Summarize allowlisted non-secret evidence for one cataloged lane.""" import json @@ -828,6 +858,16 @@ def build_parser() -> argparse.ArgumentParser: add_token_arg(lp) lp.set_defaults(func=cmd_lifecycle) + su = sub.add_parser("secret-use", help="read-only secret-use evidence surface") + susub = su.add_subparsers(dest="subcmd", required=True) + snap = susub.add_parser( + "snapshot", + help="non-secret lane metadata for kings-guard; never contacts OpenBao", + ) + snap.add_argument("--catalog-id", default="", help="limit to one catalog id") + snap.add_argument("--json", action="store_true") + snap.set_defaults(func=cmd_secret_use_snapshot) + au = sub.add_parser("audit", help="summarize non-secret local lane evidence") au.add_argument("catalog_id") au.add_argument("--json", action="store_true") diff --git a/src/secrets_engine/secret_use.py b/src/secrets_engine/secret_use.py new file mode 100644 index 0000000..6de8153 --- /dev/null +++ b/src/secrets_engine/secret_use.py @@ -0,0 +1,206 @@ +"""Read-only secret-use evidence surface for kings-guard. + +This is not ``route`` or ``audit``. It never contacts OpenBao and never +returns secret material. Completeness is not claimed; missing local evidence +is not treated as non-occurrence. +""" +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import yaml + +from secrets_engine.catalog import CatalogEntry, load_catalog +from secrets_engine.evidence_class import classification_path, load_classification_rules +from secrets_engine.redact import looks_secret, redact_text + +LANE_FIELDS = ( + "as_of", + "catalog_id", + "stage", + "kind", + "mount", + "path", + "field_names", + "ready", + "session_handle", + "revocation_attempted", + "revocation_succeeded", + "lifecycle_operation", + "decision_id", + "stance_stage", + "stance_failure_mode", + "evidence_kind", +) +LIFECYCLE_ACTIONS = { + "lifecycle-suspend": "suspend", + "lifecycle-deactivate": "deactivate", + "lifecycle-destroy": "destroy", + "revoke": "revoke", + "session-revoke": "revoke", +} + + +def declared_cadence() -> dict[str, str]: + """Publish the load-bearing cadence already declared next to the bound.""" + path = classification_path() + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + except (OSError, yaml.YAMLError): + data = {} + raw = data.get("cadence") if isinstance(data, dict) else {} + if not isinstance(raw, dict): + raw = {} + cadence = { + "form": str(raw.get("load_bearing_form") or "heartbeat"), + "interval": str(raw.get("interval") or "1d"), + "command": str(raw.get("command") or "secrets-engine evidence heartbeat"), + } + # Keep the pin honest: classification rules must still load. + load_classification_rules() + return cadence + + +def _iter_records(evidence_dir: Path, catalog_id: str) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for path in sorted(Path(evidence_dir).glob("evidence-*.jsonl")): + try: + lines = path.read_text(encoding="utf-8").splitlines() + except OSError: + continue + for line in lines: + try: + record = json.loads(line) + except (json.JSONDecodeError, TypeError): + continue + if isinstance(record, dict) and record.get("catalog_id") == catalog_id: + records.append(record) + return records + + +def _last_with(records: list[dict[str, Any]], predicate) -> dict[str, Any] | None: + for record in reversed(records): + if predicate(record): + return record + return None + + +def _safe_text(value: object) -> str: + if not isinstance(value, str) or not value: + return "" + return redact_text(value) + + +def snapshot_lane( + entry: CatalogEntry, + evidence_dir: Path, + *, + now: datetime | None = None, +) -> dict[str, Any]: + """One lane row. Catalog fields always; evidence fields only when present.""" + as_of = (now or datetime.now(timezone.utc)).astimezone(timezone.utc).isoformat() + row: dict[str, Any] = { + "as_of": as_of, + "catalog_id": entry.id, + "stage": entry.stage, + "kind": entry.kind, + "mount": entry.mount, + "path": entry.path, + "field_names": list(entry.fields), + } + records = _iter_records(evidence_dir, entry.id) + verify = _last_with( + records, lambda rec: rec.get("action") == "verify" and rec.get("result") in {"pass", "verification-failed"} + ) + if verify is not None: + row["ready"] = verify.get("result") == "pass" + + def _session_bits(record: dict[str, Any]) -> None: + detail = record.get("detail") if isinstance(record.get("detail"), dict) else {} + session = detail.get("session") if isinstance(detail.get("session"), dict) else {} + handle = session.get("session_handle") or detail.get("session_handle") or detail.get("auth_session_handle") + if isinstance(handle, str) and handle and not looks_secret(handle): + row["session_handle"] = redact_text(handle) + if "revocation_attempted" in session or "auth_revocation_attempted" in detail: + row["revocation_attempted"] = bool( + session.get("revocation_attempted") or detail.get("auth_revocation_attempted") + ) + if "revocation_succeeded" in session or "auth_revocation_succeeded" in detail: + row["revocation_succeeded"] = bool( + session.get("revocation_succeeded") or detail.get("auth_revocation_succeeded") + ) + + session_rec = _last_with( + records, + lambda rec: isinstance(rec.get("detail"), dict) + and ( + isinstance(rec["detail"].get("session"), dict) + or "session_handle" in rec["detail"] + or "auth_session_handle" in rec["detail"] + ), + ) + if session_rec is not None: + _session_bits(session_rec) + + life = _last_with(records, lambda rec: rec.get("action") in LIFECYCLE_ACTIONS) + if life is not None: + row["lifecycle_operation"] = LIFECYCLE_ACTIONS[str(life.get("action"))] + + decided = _last_with(records, lambda rec: bool(rec.get("decision_id"))) + if decided is not None: + decision_id = _safe_text(decided.get("decision_id")) + if decision_id: + row["decision_id"] = decision_id + + stance = _last_with( + records, + lambda rec: isinstance(rec.get("detail"), dict) + and rec["detail"].get("stance_stage"), + ) + if stance is not None: + detail = stance["detail"] + stage = _safe_text(detail.get("stance_stage")) + mode = _safe_text(detail.get("stance_failure_mode")) + if stage: + row["stance_stage"] = stage + if mode: + row["stance_failure_mode"] = mode + + kind_rec = _last_with(records, lambda rec: rec.get("evidence_kind") in {"load-bearing", "attributive", "heartbeat"}) + if kind_rec is not None: + row["evidence_kind"] = kind_rec["evidence_kind"] + + extra = {key: row[key] for key in list(row) if key not in LANE_FIELDS} + for key in extra: + del row[key] + return row + + +def snapshot( + catalog_dir: Path, + evidence_dir: Path, + *, + catalog_id: str = "", + now: datetime | None = None, +) -> dict[str, Any]: + """Catalog snapshot for the secret-use surface. Never talks to OpenBao.""" + current = now or datetime.now(timezone.utc) + entries = load_catalog(catalog_dir) + if catalog_id: + if catalog_id not in entries: + from secrets_engine.errors import CatalogError + + raise CatalogError(f"unknown catalog id '{catalog_id}'") + chosen = [entries[catalog_id]] + else: + chosen = [entries[key] for key in sorted(entries)] + return { + "as_of": current.astimezone(timezone.utc).isoformat(), + "surface": "secret-use-evidence", + "completeness_claimed": False, + "cadence": declared_cadence(), + "lanes": [snapshot_lane(entry, evidence_dir, now=current) for entry in chosen], + } diff --git a/tests/test_secret_use.py b/tests/test_secret_use.py new file mode 100644 index 0000000..0c8f319 --- /dev/null +++ b/tests/test_secret_use.py @@ -0,0 +1,97 @@ +import copy +import json +from datetime import datetime, timezone +from pathlib import Path + +import yaml + +from secrets_engine.catalog import validate_entry +from secrets_engine.secret_use import LANE_FIELDS, snapshot, snapshot_lane +from tests.test_catalog import VALID + +NOW = datetime(2026, 9, 2, 12, 0, tzinfo=timezone.utc) + + +def _write_catalog(tmp_path: Path, data=None): + entry = copy.deepcopy(data or VALID) + path = tmp_path / "catalog" + path.mkdir() + (path / f"{entry['id']}.yaml").write_text( + yaml.safe_dump(entry), encoding="utf-8" + ) + return path + + +def test_snapshot_omits_unknown_ready_and_never_claims_completeness(tmp_path): + catalog = _write_catalog(tmp_path) + payload = snapshot(catalog, tmp_path / "evidence", now=NOW) + assert payload["completeness_claimed"] is False + assert payload["cadence"]["form"] == "heartbeat" + assert payload["cadence"]["interval"] == "1d" + assert payload["surface"] == "secret-use-evidence" + assert len(payload["lanes"]) == 1 + lane = payload["lanes"][0] + assert lane["catalog_id"] == "test-lane" + assert "ready" not in lane + assert set(lane) <= set(LANE_FIELDS) + + +def test_snapshot_reads_last_verify_and_session_without_secrets(tmp_path): + catalog = _write_catalog(tmp_path) + evidence = tmp_path / "evidence" + evidence.mkdir() + records = [ + { + "catalog_id": "test-lane", + "action": "verify", + "result": "pass", + "decision_id": "dec-1", + "evidence_kind": "attributive", + "detail": { + "session": { + "session_handle": "abc123def456", + "revocation_attempted": True, + "revocation_succeeded": True, + }, + "secret": "npm_SHOULDNEVERAPPEAR", + }, + }, + { + "catalog_id": "test-lane", + "action": "revoke", + "result": "native-access-deactivated", + "detail": {"stance_stage": "test", "stance_failure_mode": "fail_open"}, + }, + ] + (evidence / "evidence-2026-09-02.jsonl").write_text( + "".join(json.dumps(row) + "\n" for row in records), encoding="utf-8" + ) + lane = snapshot_lane( + validate_entry(copy.deepcopy(VALID)), evidence, now=NOW + ) + assert lane["ready"] is True + assert lane["session_handle"] == "abc123def456" + assert lane["revocation_attempted"] is True + assert lane["revocation_succeeded"] is True + assert lane["lifecycle_operation"] == "revoke" + assert lane["decision_id"] == "dec-1" + assert lane["stance_stage"] == "test" + dumped = json.dumps(lane) + assert "npm_SHOULDNEVERAPPEAR" not in dumped + assert "secret" not in lane + assert set(lane) <= set(LANE_FIELDS) + + +def test_snapshot_does_not_contact_openbao(tmp_path, monkeypatch): + from secrets_engine import secret_use + + monkeypatch.setattr( + secret_use, + "load_catalog", + lambda *_args, **_kwargs: {"test-lane": validate_entry(copy.deepcopy(VALID))}, + ) + # If OpenBao were imported/called, this would be the failure mode to catch + # in handlers. The snapshot function has no OpenBao client parameter. + payload = snapshot(tmp_path, tmp_path, now=NOW) + assert payload["lanes"][0]["mount"] == "secret" + assert payload["completeness_claimed"] is False diff --git a/workplans/SECRETS-WP-0008-layer-model-lifecycle-conformance.md b/workplans/SECRETS-WP-0008-layer-model-lifecycle-conformance.md index a48c26a..12536d2 100644 --- a/workplans/SECRETS-WP-0008-layer-model-lifecycle-conformance.md +++ b/workplans/SECRETS-WP-0008-layer-model-lifecycle-conformance.md @@ -204,6 +204,11 @@ proposed. `route`/`audit` are still not this surface. an observation input until the surface ships and publishes its cadence declaration. No Tooling contact will be opened to fill that wait. +**2026-09-02 later:** shipped `secrets-engine secret-use snapshot`. Cadence is +the declared 1d heartbeat on the snapshot envelope. `owner_status` stays +`proposed` until kings-guard admits the snapshot as an observation input. +`route`/`audit` remain distinct operator summaries. + kings-guard's secret-abuse posture is fixture-driven because no engine exposes lease, revocation, mount, rotation, and delivery-session metadata. `route` and `audit` are operator summaries over local JSONL and are not that surface.