diff --git a/INTENT.md b/INTENT.md index dbafc28..448c97d 100644 --- a/INTENT.md +++ b/INTENT.md @@ -202,7 +202,7 @@ human-readable view of it. | `capability` | Needed for | `intended_owner` | `blocked_on` | `review` | | --- | --- | --- | --- | --- | | Authentication and assurance evidence (token assurance, attestation outcomes, authentication anomalies) exposed as an engine surface | identity-drift posture | identity layer + `audit-core` — **`access-engine` declined** (v0.6 §13); reproposed, not assented | no engine surface exists; kings-guard consumes fixtures only | 2026-11-28 | -| Secret-use evidence (lease, revocation, mount and rotation metadata) exposed as an engine surface | secret-abuse posture | `secrets-engine` | no engine surface exists; kings-guard consumes fixtures only | 2026-11-28 | +| Secret-use evidence (lease, revocation, mount and rotation metadata) exposed as an engine surface | secret-abuse posture | `secrets-engine` | `secret-use snapshot` shipped with a 1d heartbeat declaration; snapshot parsing admitted under `KG-WP-0006`; event provenance and posture admission remain pending under `KG-IN-0005` | 2026-11-28 | | Actuation surface — reduce authority, require step-up, isolate a workload — as a deterministic engine API carrying a decision record | containment kings-guard **proposes but does not own** | `access-engine` + runtime PEPs; not reviewed (`FLEX-DEC-2026-002`) | ruled an Engine concept held at zero (v0.6 §9.2); recorded here as a dependency, not a kings-guard gap to close | 2026-11-28 | Until a gap closes, the corresponding posture lane stays advisory and diff --git a/SCOPE.md b/SCOPE.md index cb20a91..d848f30 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -6,10 +6,11 @@ ## Layer -**Staff** — interactive, non-deterministic; adaptive defence, observation, -containment. Binding rule: kings-guard never touches Tooling directly; it acts -only through Engine APIs. See `INTENT.md` and -`net-kingdom/canon/standards/security-layer-model_v0.6.md`. +**Staff** — interactive, non-deterministic; adaptive defence, observation, and +judgment. Kings-guard proposes containment but does not own or perform it. +Binding rule: kings-guard never touches Tooling directly; it acts only through +Engine APIs. See `INTENT.md` and +`net-kingdom/canon/standards/security-layer-model_v0.7.md`. --- @@ -103,9 +104,17 @@ does it decide and act?** The first is in scope. The second is another layer's. no-standing-credential claim executable; the companion cites them estate-wide as the reference for a repository with no Tooling contacts at all. - Live observation of `qonto-assistant` uses the source's own AuditLogger emit - path. The hand-built fixture is retained as a regression case. Stream - completeness for the load-bearing deny class stays unknown until the source - publishes a cadence declaration and heartbeats. + path. The hand-built fixture is retained as a regression case. The local + contract proof now consumes the source-owned cadence declaration and checks + heartbeat, sequence, and reconciliation evidence for the load-bearing deny + class. Validation against a deployed stream remains operational work. +- Engine secret-use snapshots have a pure, scoped metadata adapter with explicit + unknown completeness and snapshot freshness findings. Event-based secret-abuse + posture awaits source provenance and operational evidence (`KG-IN-0005`). See + `docs/SecretUseSnapshotAdmission.md`. +- Cadence fixtures consume InfoTechCanon `standard/emission-cadence` 0.1; + the local handover draft is historical and NetKingdom profile adoption remains + with its owner. - `specs/NetKingdomImmuneArchitecture.md` was swept under `KG-WP-0004` onto Staff / Engine / Tooling (v0.7). It is an estate map; this repository implements the Staff slice only. Containment is proposed, not performed. diff --git a/decisions/decisions.md b/decisions/decisions.md index bb17285..c745cc3 100644 --- a/decisions/decisions.md +++ b/decisions/decisions.md @@ -229,3 +229,32 @@ Until §17's schemas exist, two parties can disagree about whether they hold the same evidence and recomputability is a thought experiment rather than a check. §17 is therefore load-bearing for this decision; `kings-guard` owns the emission-cadence half under `KG-WP-0003-T02`. + +## KG-DEC-2026-003 — Admit scoped snapshots without claiming event completeness + +```yaml +id: KG-DEC-2026-003 +kind: decision +title: Admit scoped snapshots without claiming event completeness +status: resolved +owner: codex +repo: kings-guard +origin_ref: KG-WP-0006 +created: '2026-09-05' +updated: '2026-09-05' +decided_by: codex +state_hub_decision_id: "200e63b3-973a-4e3d-a687-3da0949bdcb6" +``` + +Consume secrets-engine snapshots through a pure typed adapter with caller-owned +catalog-to-tenant/subject bindings. Preserve omitted evidence as unknown and +assess only snapshot freshness. The source merges independently selected +historical fields without event timestamps or actors, so do not synthesize an +authorization event or infer current secret-abuse posture. Completeness stays +unknown; decision ids and session handles are not retained. Source provenance +and operational evidence are handed off as KG-IN-0005. + +Consume InfoTechCanon standard/emission-cadence 0.1 in the Qonto fixtures; +NetKingdom security fields and local provenance are namespaced extensions. +The handover draft remains historical and NK-WP-0035 profile adoption stays +with its owner. Deployed Qonto validation remains KG-WP-0005-T03. diff --git a/docs/AdjacentSystemBoundary.md b/docs/AdjacentSystemBoundary.md index 31f08f3..1d19936 100644 --- a/docs/AdjacentSystemBoundary.md +++ b/docs/AdjacentSystemBoundary.md @@ -31,7 +31,7 @@ The following rules apply to every integration below: 4. `kings-guard` must preserve **tenant isolation**: any retained evidence or memory must stay bounded by declared confidentiality rules. 5. `kings-guard` is a **Staff-layer** repository and is bound by §5 of the - NetKingdom Security Layer Model v0.6: it never holds a direct client for a + NetKingdom Security Layer Model v0.7: it never holds a direct client for a Tooling-layer system. Evidence from Tooling (OpenBao, key-cape components) is consumed **through the owning engine**. Where no engine surface exists, the lane stays fixture-driven and the gap is declared in `INTENT.md`. diff --git a/docs/SecretUseSnapshotAdmission.md b/docs/SecretUseSnapshotAdmission.md new file mode 100644 index 0000000..34138f9 --- /dev/null +++ b/docs/SecretUseSnapshotAdmission.md @@ -0,0 +1,40 @@ +# Secret-use snapshot admission + +`adapters.secrets_engine.snapshot_from_engine` consumes the non-secret envelope +returned by `secrets-engine secret-use snapshot --json`. It opens no client and +reads no Engine evidence directory. The caller obtains the Engine response and +supplies an explicit `catalog_id -> (tenant_id, subject_id)` binding; unbound or +duplicate lanes are rejected. Catalog identifiers alone do not establish scope. + +| Source field | Consumer meaning | +| --- | --- | +| `surface`, `completeness_claimed` | Require `secret-use-evidence` and explicit `false`. | +| Envelope and lane `as_of` | Snapshot creation time; lane times must match. Not the times of underlying events. | +| `catalog_id`, `stage`, `kind` | Scoped catalog metadata. | +| `ready`, `revocation_attempted`, `revocation_succeeded` | Optional historical booleans. Absence stays unknown; false stays false. | +| `lifecycle_operation`, `evidence_kind` | Source-reported metadata, not a new event or inferred evidence classification. | +| `cadence.form`, `cadence.interval` | Require heartbeat form and a positive interval; translate `1d` to 86400 seconds. | +| Decision IDs, session handles, paths, mounts, field names, extra fields | Not retained in the normalized observation. | + +The result is a separate `SecretUseSnapshot`, because `ImmuneObservation` +requires an event actor, event time, and outcome this surface cannot provide. +Readiness never becomes allow/deny. A snapshot older than the declared interval +receives `snapshot:stale`; a future snapshot receives `snapshot:future_timestamp`. +Even a fresh snapshot has **unknown completeness**. Its cadence declaration is +not a received heartbeat. It supplies neither heartbeat assertions nor +reconciliation counts, so generic stream evaluation is not invoked. + +Each lane merges independently selected historical evidence. A newly created +snapshot may contain old readiness or revocation information. No content posture +or response proposal is generated from these untimed fields. KG-IN-0005 tracks +source event timestamps/provenance, evidence-class bindings, scoped heartbeat and +reconciliation evidence, and an authorized deployment capture before that lane +can support posture. Snapshot parsing is locally admitted; secret-abuse posture +remains pending. + +Cadence examples for Qonto now follow InfoTechCanon +`standard/emission-cadence` 0.1. `extensions.net-kingdom` holds source evidence +classification and rate-monitoring metadata; `extensions.kings-guard` holds +provenance. The old handover spec is historical. Tests validate both fixtures +against the external owner schema when its checkout is available, without a +copied generic schema. NetKingdom profile adoption remains with NK-WP-0035. diff --git a/docs/pilots/QontoAssistantPosturePilot.md b/docs/pilots/QontoAssistantPosturePilot.md index f23e207..76ae8d2 100644 --- a/docs/pilots/QontoAssistantPosturePilot.md +++ b/docs/pilots/QontoAssistantPosturePilot.md @@ -71,8 +71,8 @@ client request | `protocol` | `protocol` | `rest` or `mcp` | | `decision` | `decision` | `allow` / `deny` | | `deny_reason` | `deny_reason` | Pilot currently exercises `credential_exfil` | -| derived constant | `identity_binding` | `self_asserted` until key-cape integration lands | -| derived constant | `egress_destination` | `qonto-thirdparty-api` for this pilot | +| `identity_binding` | `identity_binding` | Direct mapping; legacy records fall back to the genome normalization hint | +| `egress_destination` | `egress_destination` | Direct mapping; legacy records fall back to the genome normalization hint | | genome `evidence_sources` | `evidence_class` / `event_class` | Copied from the source declaration. `audit.deny` is load-bearing; `audit.allow` is attributive. Not inferred from `decision`. | ## 6. Output Shape diff --git a/intakes/intakes.md b/intakes/intakes.md index a901a77..6a1f7d2 100644 --- a/intakes/intakes.md +++ b/intakes/intakes.md @@ -1,5 +1,35 @@ # Intake records +## KG-IN-0004 — Admit the shipped secrets-engine secret-use snapshot as observation input + +```yaml +id: KG-IN-0004 +kind: intake +title: Admit the shipped secrets-engine secret-use snapshot as observation input +status: closed +outcome: promoted +promoted_to: KG-WP-0006 +closed: '2026-09-05' +origin: cross-repo +origin_ref: SECRETS-WP-0008-T05 +priority: medium +owner: kings-guard +requested_by: secrets-engine +description: > + secrets-engine has shipped `secrets-engine secret-use snapshot [--catalog-id + ID] [--json]` as a read-only Engine / Lifecycle surface over non-secret local + evidence and catalog metadata. The envelope explicitly denies completeness, + omits evidence-derived fields when no record exists, and declares a 1d + heartbeat through `secrets-engine evidence heartbeat`. Review and admit the + surface as an immune-observation input without adding a Tooling client, + treating omission as non-occurrence, or treating readiness or decision ids + as cached authorization. Define the lane-row mapping, cadence translation, + stale-snapshot behavior, and tests before changing the secret-observation + capability from pending. +created: '2026-09-04' +updated: '2026-09-05' +``` + ## KG-IN-0001 — Assent requested: Staff layer placement, control-plane vocabulary, and the posture asymmetry ```yaml @@ -123,3 +153,30 @@ resolution: 'Answered with a proposed revision. The boundary is recomputability, docs/PostureMaturityBoundary.md. Decision: KG-DEC-2026-002.' state_hub_intake_id: "01a04d8d-66f8-70ef-b2b5-1eff03569471" ``` + +## KG-IN-0005 — Source evidence needed for secret-abuse posture + +```yaml +id: KG-IN-0005 +kind: intake +title: Source evidence needed for secret-abuse posture +status: open +origin: residual +origin_ref: KG-WP-0006 +priority: medium +owner: kings-guard +requested_by: kings-guard +related: +- SECRETS-WP-0008-T05 +- KG-IN-0004 +description: > + Snapshot parsing is admitted, but the secrets-engine envelope combines + historical fields without their event timestamps, actors or record provenance. + Obtain an Engine-owned event/provenance contract and explicit evidence-class + bindings, scoped heartbeat assertions and reconciliation evidence, plus an + authorized deployment capture. Review those inputs before enabling secret-abuse + posture. Keep snapshot completeness unknown and never infer allow/deny from + readiness, lifecycle metadata or decision ids. No direct Tooling contact. +created: '2026-09-05' +updated: '2026-09-05' +``` diff --git a/layer.yaml b/layer.yaml index 24c8f4d..357afda 100644 --- a/layer.yaml +++ b/layer.yaml @@ -4,7 +4,7 @@ # Assent: decisions/decisions.md KG-DEC-2026-001 (kings-guard's own voice, §11) # Validate: python3 scripts/check_layer_conformance.py # -# §11 (v0.6) requires a machine-readable declaration: prose cannot distinguish a +# §11 (v0.7) requires a machine-readable declaration: prose cannot distinguish a # declaration from a transcribed review. Form adapted from ops-warden's # reference implementation, offered under §11. # @@ -109,10 +109,12 @@ unowned_capabilities: Lease, revocation, mount and rotation metadata exposed as an engine surface, for secret-abuse posture. intended_owner: secrets-engine - owner_status: proposed + owner_status: "snapshot parsing admitted; posture admission pending" blocked_on: >- - No engine exposes secret-use evidence. OpenBao is Tooling; same reasoning - as above. + secrets-engine now exposes `secret-use snapshot`, a non-secret local + evidence surface with a 1d heartbeat declaration. Envelope and scoped lane + parsing are admitted under KG-WP-0006; event provenance, heartbeat evidence + and posture admission remain pending under KG-IN-0005. review: "2026-11-28" consequence: "secret-abuse posture lane stays fixture-driven" diff --git a/pyproject.toml b/pyproject.toml index ec8e8ac..7d4538e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ dev = [ # DEV dependency: `dependencies = []` above is load-bearing for the §5 # no-Tooling-client claim and must stay empty. "pyyaml>=6.0,<7.0", + "jsonschema>=4.23,<5", ] [project.scripts] diff --git a/scripts/check_layer_conformance.py b/scripts/check_layer_conformance.py index bb306f5..9126428 100644 --- a/scripts/check_layer_conformance.py +++ b/scripts/check_layer_conformance.py @@ -6,7 +6,7 @@ Read-only. kings-guard's whole position under the standard rests on one claim: it holds no direct client for any Tooling-layer system, and the capabilities that would need one sit at zero instead (§11 blocked-clean) -That claim has been asserted in prose since KG-DEC-2026-001. §11 (v0.6) requires +That claim has been asserted in prose since KG-DEC-2026-001. §11 (v0.7) requires a machine-readable declaration because prose cannot distinguish a declaration from a transcribed review. This script is what makes the claim checkable: it fails if a Tooling client appears in src/ without a matching layer.yaml entry. diff --git a/specs/EmissionCadenceDeclaration.md b/specs/EmissionCadenceDeclaration.md index 09a8a83..5cfe7b0 100644 --- a/specs/EmissionCadenceDeclaration.md +++ b/specs/EmissionCadenceDeclaration.md @@ -16,6 +16,12 @@ classification: Public # Emission-cadence declaration — Taxonomy draft +> Historical handover, frozen after assimilation. InfoTechCanon published +> `standard/emission-cadence` schema 0.1 in canon 0.7.0 (ITC-WP-0018). +> Canonical schema: `info-tech-canon/infospace/schemas/emission-cadence.schema.yaml`. +> Runtime fixtures now consume that shape; NetKingdom owns the security profile +> (NK-WP-0035, still proposed). The original proposal below is provenance. + kings-guard drafts this artifact because it is the only consumer of silence-as-signal and cannot implement §9.6 without a declared cadence. **Ownership stays with Taxonomy.** This file is a handover, not an internal diff --git a/src/kings_guard/adapters/qonto_assistant.py b/src/kings_guard/adapters/qonto_assistant.py index 08ef2be..87390b3 100644 --- a/src/kings_guard/adapters/qonto_assistant.py +++ b/src/kings_guard/adapters/qonto_assistant.py @@ -50,8 +50,16 @@ def observation_from_audit_event( evidence_class=resolved_class, event_class=resolved_event_class, deny_reason=_optional_str(event.get("deny_reason")), - identity_binding=identity_binding, - egress_destination=egress_destination, + identity_binding=( + _optional_str(event.get("identity_binding")) + if "identity_binding" in event + else identity_binding + ), + egress_destination=( + _optional_str(event.get("egress_destination")) + if "egress_destination" in event + else egress_destination + ), latency_ms=_optional_int(event.get("latency_ms")), result_count=_optional_int(event.get("result_count")), policy_version=_optional_int(event.get("policy_version")), diff --git a/src/kings_guard/adapters/secrets_engine.py b/src/kings_guard/adapters/secrets_engine.py new file mode 100644 index 0000000..1517c9b --- /dev/null +++ b/src/kings_guard/adapters/secrets_engine.py @@ -0,0 +1,125 @@ +"""Pure admission of the Engine's non-secret secret-use snapshot envelope. + +Callers supply an Engine response and explicit catalog-to-tenant bindings. +A snapshot is not an authorization event or proof of complete observation. +""" +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import timedelta +from typing import Any + +from kings_guard.cadence import parse_interval, parse_timestamp +from kings_guard.contracts import StreamCompleteness + + +@dataclass(frozen=True, slots=True) +class SecretUseLane: + catalog_id: str + tenant_id: str + subject_id: str + stage: str + kind: str + ready: bool | None + revocation_attempted: bool | None + revocation_succeeded: bool | None + lifecycle_operation: str | None + evidence_kind: str | None + + +@dataclass(frozen=True, slots=True) +class SecretUseSnapshot: + as_of: str + lanes: tuple[SecretUseLane, ...] + heartbeat_interval: timedelta + findings: tuple[str, ...] + completeness: StreamCompleteness = StreamCompleteness.UNKNOWN + source_system: str = "secrets-engine" + + +def snapshot_from_engine( + payload: Mapping[str, Any], + *, + lane_bindings: Mapping[str, tuple[str, str]], + now: str, +) -> SecretUseSnapshot: + """Admit scoped snapshot metadata; freshness never asserts event recency. + + Bindings are caller-owned (tenant_id, subject_id), never inferred from paths. + Omitted evidence stays None. The response's decision id, session handle, + mount, path and arbitrary extra fields are deliberately not retained. + """ + if payload.get("surface") != "secret-use-evidence": + raise ValueError("unsupported Engine snapshot surface") + if payload.get("completeness_claimed") is not False: + raise ValueError("secret-use snapshot must explicitly deny completeness") + as_of = _text(payload, "as_of") + age = parse_timestamp(now) - parse_timestamp(as_of) + cadence = payload.get("cadence") + if not isinstance(cadence, Mapping) or cadence.get("form") != "heartbeat": + raise ValueError("expected source-declared heartbeat cadence") + interval = parse_interval(_text(cadence, "interval")) + if interval <= timedelta(0): + raise ValueError("heartbeat interval must be positive") + raw_lanes = payload.get("lanes") + if not isinstance(raw_lanes, list): + raise ValueError("snapshot lanes must be a list") + lanes: list[SecretUseLane] = [] + seen: set[str] = set() + for row in raw_lanes: + if not isinstance(row, Mapping): + raise ValueError("snapshot lane must be an object") + catalog_id = _text(row, "catalog_id") + if catalog_id in seen or catalog_id not in lane_bindings: + raise ValueError("duplicate or unbound catalog lane") + seen.add(catalog_id) + binding = lane_bindings[catalog_id] + if (not isinstance(binding, tuple) or len(binding) != 2 + or any(not isinstance(value, str) or not value.strip() for value in binding)): + raise ValueError("each lane needs explicit tenant and subject scope") + if parse_timestamp(_text(row, "as_of")) != parse_timestamp(as_of): + raise ValueError("lane timestamp differs from the snapshot envelope") + operation = _optional_text(row, "lifecycle_operation") + if operation not in {None, "suspend", "deactivate", "destroy", "revoke"}: + raise ValueError("unsupported lifecycle operation") + evidence_kind = _optional_text(row, "evidence_kind") + if evidence_kind not in {None, "attributive", "load-bearing", "heartbeat"}: + raise ValueError("unsupported evidence kind") + lanes.append(SecretUseLane( + catalog_id=catalog_id, + tenant_id=binding[0], + subject_id=binding[1], + stage=_text(row, "stage"), + kind=_text(row, "kind"), + ready=_optional_bool(row, "ready"), + revocation_attempted=_optional_bool(row, "revocation_attempted"), + revocation_succeeded=_optional_bool(row, "revocation_succeeded"), + lifecycle_operation=operation, + evidence_kind=evidence_kind, + )) + findings: list[str] = [] + if age < timedelta(0): + findings.append("snapshot:future_timestamp") + elif age > interval: + findings.append("snapshot:stale") + return SecretUseSnapshot(as_of, tuple(lanes), interval, tuple(findings)) + + +def _text(payload: Mapping[str, Any], key: str) -> str: + value = payload.get(key) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{key} must be a nonempty string") + return value + + +def _optional_text(payload: Mapping[str, Any], key: str) -> str | None: + return _text(payload, key) if key in payload else None + + +def _optional_bool(payload: Mapping[str, Any], key: str) -> bool | None: + if key not in payload: + return None + if type(payload[key]) is not bool: + raise ValueError(f"{key} must be a boolean when supplied") + return payload[key] diff --git a/src/kings_guard/cadence.py b/src/kings_guard/cadence.py index 35c6dcb..7c61cb5 100644 --- a/src/kings_guard/cadence.py +++ b/src/kings_guard/cadence.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import re from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import UTC, datetime, timedelta @@ -32,15 +33,10 @@ def parse_interval(value: str | int) -> timedelta: return timedelta(hours=int(text[:-1])) if text.endswith("d") and text[:-1].isdigit(): return timedelta(days=int(text[:-1])) - if text.startswith("pt"): - # Minimal ISO-8601 duration: PT24H, PT1H, PT30M. - amount = text[2:] - if amount.endswith("h") and amount[:-1].isdigit(): - return timedelta(hours=int(amount[:-1])) - if amount.endswith("m") and amount[:-1].isdigit(): - return timedelta(minutes=int(amount[:-1])) - if amount.endswith("s") and amount[:-1].isdigit(): - return timedelta(seconds=int(amount[:-1])) + match = re.fullmatch(r"p(?:(\d+)d)?(?:t(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?)?", text) + if match and any(match.groups()): + days, hours, minutes, seconds = (int(value or 0) for value in match.groups()) + return timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds) raise ValueError(f"unsupported interval: {value!r}") @@ -71,10 +67,10 @@ class ReconciliationCadence: @dataclass(frozen=True, slots=True) class EmissionCadence: - """Runtime view of the Taxonomy draft, loaded from the local worked example. + """Runtime projection of InfoTechCanon standard/emission-cadence 0.1. - This is a consumer of the draft in `specs/EmissionCadenceDeclaration.md`, - not a competing schema. Ownership stays with Taxonomy. + NetKingdom classifications and local provenance live in extensions. + Canonical schema validation belongs to the owner's schema, not a local copy. """ schema_version: str @@ -106,12 +102,27 @@ def load_qonto_assistant_cadence() -> EmissionCadence: return emission_cadence_from_dict(payload) +def load_qonto_assistant_source_cadence() -> EmissionCadence: + """Load qonto-assistant's shipped source-owned emission declaration.""" + payload = json.loads( + files("kings_guard") + .joinpath("fixtures") + .joinpath("qonto_assistant_source_cadence.json") + .read_text(encoding="utf-8") + ) + return emission_cadence_from_dict(payload) + + def emission_cadence_from_dict(data: Mapping[str, Any]) -> EmissionCadence: rates: list[RateCadence] = [] heartbeats: list[HeartbeatCadence] = [] reconciliations: list[ReconciliationCadence] = [] - for item in data.get("sources", ()): - evidence_class = EvidenceClass(str(item["evidence_class"])) + if data.get("schema_version") != "0.1" or not data.get("declaration_id"): + raise ValueError("expected canonical emission-cadence 0.1 declaration") + provenance = data.get("extensions", {}).get("kings-guard", {}) + for item in data["sources"]: + profile = item.get("extensions", {}).get("net-kingdom", {}) + evidence_class = EvidenceClass(str(profile["evidence_class"])) form = str(item["form"]) if form == "expected-rate": rates.append( @@ -150,11 +161,11 @@ def emission_cadence_from_dict(data: Mapping[str, Any]) -> EmissionCadence: ) return EmissionCadence( schema_version=str(data.get("schema_version", "0.1")), - status=str(data.get("status", "taxonomy-draft")), - drafter=str(data.get("drafter", "kings-guard")), - owner=str(data.get("owner", "Taxonomy")), + status=str(provenance.get("status", "source-declared")), + drafter=str(provenance.get("drafter", data["source"])), + owner=str(provenance.get("owner", data["source"])), source_system=str(data.get("source", data.get("source_system", "unknown"))), - reference_instance=str(data.get("reference_instance", "GH-WP-0002-T04")), + reference_instance=str(provenance.get("reference_instance", data["declaration_id"])), rates=tuple(rates), heartbeats=tuple(heartbeats), reconciliations=tuple(reconciliations), diff --git a/src/kings_guard/contracts.py b/src/kings_guard/contracts.py index b4edca5..c7d0c01 100644 --- a/src/kings_guard/contracts.py +++ b/src/kings_guard/contracts.py @@ -140,12 +140,9 @@ class SecurityGenome: permitted_protocols=frozenset(str(item) for item in data["permitted_protocols"]), permitted_egress=frozenset(str(item) for item in data["permitted_egress"]), data_classifications=tuple(str(item) for item in data.get("data_classifications", ())), - tolerances=tuple( - ToleranceRule.from_dict(item) for item in data.get("tolerances", ()) - ), + tolerances=tuple(ToleranceRule.from_dict(item) for item in data.get("tolerances", ())), evidence_sources=tuple( - DeclaredEvidenceSource.from_dict(item) - for item in data.get("evidence_sources", ()) + DeclaredEvidenceSource.from_dict(item) for item in data.get("evidence_sources", ()) ), ) diff --git a/src/kings_guard/fixtures/qonto_assistant_cadence.json b/src/kings_guard/fixtures/qonto_assistant_cadence.json index 6ba4a83..dd21962 100644 --- a/src/kings_guard/fixtures/qonto_assistant_cadence.json +++ b/src/kings_guard/fixtures/qonto_assistant_cadence.json @@ -1,35 +1,31 @@ { "schema_version": "0.1", - "status": "taxonomy-draft", - "drafter": "kings-guard", - "owner": "Taxonomy", "source": "qonto-assistant", - "belongs_alongside": "security_genome", - "reference_instance": "GH-WP-0002-T04", - "reference_source_declaration": "approval-engine/cadence.yaml", "sources": [ { "source_id": "qonto-assistant.audit.allow", "source_system": "qonto-assistant", "event_class": "audit.allow", - "evidence_class": "attributive", "form": "expected-rate", - "window": "24h", "window_seconds": 86400, "expected_min": 1, "drop_below": "finding", - "note": "Worked example of the volume form. qonto-assistant is called sporadically, so this rate is a SHOULD illustration, not a claim that completeness of allows is currently meaningful." + "extensions": { + "net-kingdom": { + "evidence_class": "attributive" + }, + "kings-guard": { + "note": "Worked example of the volume form. qonto-assistant is called sporadically, so this rate is a SHOULD illustration, not a claim that completeness of allows is currently meaningful." + } + } }, { "source_id": "qonto-assistant.audit.deny", "source_system": "qonto-assistant", "event_class": "audit.deny", - "evidence_class": "load-bearing", "form": "heartbeat-or-reconciliation", - "rate_monitoring": "forbidden", "heartbeat": { "event_class": "audit.heartbeat", - "interval": "24h", "interval_seconds": 86400, "assertion": "nothing-to-report", "missing": "finding" @@ -38,7 +34,24 @@ "compare_local": "source_transition_counts.audit.deny", "compare_observed": "evidence_counts.audit.deny", "divergence": "finding" + }, + "extensions": { + "net-kingdom": { + "evidence_class": "load-bearing", + "rate_monitoring": "forbidden" + } } } - ] + ], + "declaration_id": "kings-guard.qonto-worked-example", + "extensions": { + "kings-guard": { + "status": "taxonomy-draft", + "drafter": "kings-guard", + "owner": "Taxonomy", + "belongs_alongside": "security_genome", + "reference_instance": "GH-WP-0002-T04", + "reference_source_declaration": "approval-engine/cadence.yaml" + } + } } diff --git a/src/kings_guard/fixtures/qonto_assistant_source_cadence.json b/src/kings_guard/fixtures/qonto_assistant_source_cadence.json new file mode 100644 index 0000000..3e0e45a --- /dev/null +++ b/src/kings_guard/fixtures/qonto_assistant_source_cadence.json @@ -0,0 +1,38 @@ +{ + "schema_version": "0.1", + "source": "qonto-assistant", + "sources": [ + { + "source_id": "qonto-assistant.audit.deny", + "source_system": "qonto-assistant", + "event_class": "audit.deny", + "form": "heartbeat-or-reconciliation", + "heartbeat": { + "event_class": "audit.heartbeat", + "interval_seconds": 86400, + "assertion": "nothing-to-report", + "missing": "finding" + }, + "reconciliation": { + "compare_local": "source_transition_counts.audit.deny", + "compare_observed": "evidence_counts.audit.deny", + "divergence": "finding" + }, + "extensions": { + "net-kingdom": { + "evidence_class": "load-bearing", + "rate_monitoring": "forbidden" + } + } + } + ], + "declaration_id": "qonto-assistant.audit-source-cadence", + "extensions": { + "kings-guard": { + "status": "source-declared", + "drafter": "qonto-assistant", + "owner": "qonto-assistant", + "reference_instance": "QONTO-WP-0005" + } + } +} diff --git a/src/kings_guard/live.py b/src/kings_guard/live.py index a50028d..c3f8d99 100644 --- a/src/kings_guard/live.py +++ b/src/kings_guard/live.py @@ -8,12 +8,15 @@ from pathlib import Path from typing import Any from kings_guard.adapters import observation_from_audit_event -from kings_guard.contracts import ImmuneObservation, SecurityGenome +from kings_guard.contracts import ( + ImmuneObservation, + ReconciliationView, + SecurityGenome, + StreamHeartbeat, +) QONTO_ASSISTANT_SRC = Path("/home/worsch/qonto-assistant/src") -QONTO_POLICY = ( - Path("/home/worsch/qonto-assistant/src/qonto_assistant/policy/qonto-v1.yaml") -) +QONTO_POLICY = Path("/home/worsch/qonto-assistant/src/qonto_assistant/policy/qonto-v1.yaml") QONTO_FIXTURES = Path("/home/worsch/qonto-assistant/tests/fixtures/qonto") @@ -21,8 +24,12 @@ QONTO_FIXTURES = Path("/home/worsch/qonto-assistant/tests/fixtures/qonto") class LiveQontoCapture: events: tuple[dict[str, Any], ...] observations: tuple[ImmuneObservation, ...] + heartbeats: tuple[StreamHeartbeat, ...] + reconciliation: ReconciliationView mapping_notes: tuple[str, ...] corrections_for_source: tuple[str, ...] + stream_notes: tuple[str, ...] + stream_corrections: tuple[str, ...] def qonto_assistant_available() -> bool: @@ -36,9 +43,7 @@ def capture_qonto_assistant_events(genome: SecurityGenome) -> LiveQontoCapture: client is opened; the source publishes its own stream. """ if not qonto_assistant_available(): - raise FileNotFoundError( - f"qonto-assistant checkout not found at {QONTO_ASSISTANT_SRC}" - ) + raise FileNotFoundError(f"qonto-assistant checkout not found at {QONTO_ASSISTANT_SRC}") src = str(QONTO_ASSISTANT_SRC) if src not in sys.path: @@ -53,6 +58,8 @@ def capture_qonto_assistant_events(genome: SecurityGenome) -> LiveQontoCapture: from qonto_assistant.service import CapabilityService events: list[dict[str, Any]] = [] + audit_logger = AuditLogger(sink=events.append) + audit_logger.emit_heartbeat(reason="startup") service = CapabilityService( client=FixtureQontoClient(fixture_dir=QONTO_FIXTURES), policy=PolicyEngine.from_file( @@ -60,7 +67,7 @@ def capture_qonto_assistant_events(genome: SecurityGenome) -> LiveQontoCapture: required_scope="finance.qonto.read", enforce_scope=False, ), - audit_logger=AuditLogger(sink=events.append), + audit_logger=audit_logger, rate_limiter=RateLimiter(limit=100, window_seconds=60), concurrency_limiter=ConcurrencyLimiter(limit=4), ) @@ -89,8 +96,13 @@ def capture_qonto_assistant_events(genome: SecurityGenome) -> LiveQontoCapture: else: raise RuntimeError("expected qonto-assistant to emit a deny for oversized page_size") - if len(events) < 2: - raise RuntimeError(f"expected allow and deny audit events, got {len(events)}") + audit_logger.emit_heartbeat(reason="periodic") + reconciliation_payload = audit_logger.reconciliation_snapshot() + request_events = tuple( + event for event in events if event.get("event_class") in {"audit.allow", "audit.deny"} + ) + if len(request_events) < 2: + raise RuntimeError(f"expected allow and deny audit events, got {len(request_events)}") observations = tuple( observation_from_audit_event( @@ -101,14 +113,25 @@ def capture_qonto_assistant_events(genome: SecurityGenome) -> LiveQontoCapture: egress_destination="qonto-thirdparty-api", genome=genome, ) - for event in events + for event in request_events ) - mapping_notes, corrections = _confirm_mapping(events, observations) + mapping_notes, corrections = _confirm_mapping(request_events, observations) + heartbeats, reconciliation, stream_notes, stream_corrections = _confirm_stream( + events, reconciliation_payload + ) + if corrections or stream_corrections: + raise ValueError("source capture failed admission: " + "; ".join( + (*corrections, *stream_corrections) + )) return LiveQontoCapture( - events=tuple(events), + events=request_events, observations=observations, + heartbeats=heartbeats, + reconciliation=reconciliation, mapping_notes=mapping_notes, corrections_for_source=corrections, + stream_notes=stream_notes, + stream_corrections=stream_corrections, ) @@ -166,6 +189,14 @@ def _confirm_mapping( corrections.append("actor mapping drifted") if observation.decision.value != str(event["decision"]): corrections.append("decision mapping drifted") + if "identity_binding" in event and observation.identity_binding != ( + event["identity_binding"] or None + ): + corrections.append("identity_binding mapping drifted") + if "egress_destination" in event and observation.egress_destination != ( + event["egress_destination"] or None + ): + corrections.append("egress_destination mapping drifted") notes.append( f"{event['request_id']}: decision={event['decision']} " f"capability={event['capability']} protocol={event['protocol']} " @@ -196,3 +227,95 @@ def _read_jsonl(path: Path) -> list[dict[str, Any]]: continue events.append(json.loads(text)) return events + + +def _confirm_stream( + events: Sequence[Mapping[str, Any]], + snapshot: Mapping[str, Any], +) -> tuple[ + tuple[StreamHeartbeat, ...], + ReconciliationView, + tuple[str, ...], + tuple[str, ...], +]: + notes: list[str] = [] + corrections: list[str] = [] + if not events: + raise ValueError("cannot validate an empty source capture") + instance_ids = {str(event.get("stream_instance_id")) for event in events} + if len(instance_ids) != 1 or "None" in instance_ids: + corrections.append("stream_instance_id is missing or changed within one capture") + + if any( + type(event.get("stream_sequence")) is not int or event["stream_sequence"] < 1 + for event in events + ): + raise ValueError("stream_sequence must be a positive integer") + if snapshot.get("stream_instance_id") not in instance_ids: + corrections.append("reconciliation stream_instance_id does not match the stream") + stream_ids = {event.get("stream_id") for event in events} + if len(stream_ids) != 1 or None in stream_ids or snapshot.get("stream_id") not in stream_ids: + corrections.append("reconciliation stream_id is missing or does not match the stream") + sequences = [event["stream_sequence"] for event in events] + expected_sequences = list(range(1, len(events) + 1)) + if sequences != expected_sequences: + corrections.append("stream_sequence is not contiguous from 1 within the instance") + if (type(snapshot.get("last_stream_sequence")) is not int + or snapshot["last_stream_sequence"] != sequences[-1]): + corrections.append("reconciliation last_stream_sequence does not match the stream") + + heartbeat_events = [event for event in events if event.get("event_class") == "audit.heartbeat"] + heartbeats = tuple( + StreamHeartbeat( + source_system="qonto-assistant", + timestamp=str(event["timestamp"]), + event_class="audit.heartbeat", + assertion=str(event["assertion"]), + counts=_transition_counts(event), + ) + for event in heartbeat_events + ) + if not heartbeats: + corrections.append("source emitted no audit.heartbeat record") + running_counts = {"audit.allow": 0, "audit.deny": 0} + for event in events: + event_class = event.get("event_class") + if event_class in running_counts: + running_counts[event_class] += 1 + elif event_class == "audit.heartbeat": + if _transition_counts(event) != running_counts: + corrections.append("heartbeat counts do not match preceding request events") + if event.get("assertion") not in {"nothing-to-report", "transitions-reported"}: + corrections.append("heartbeat assertion is not a recognized positive claim") + else: + corrections.append("source emitted an undeclared event class") + + source_counts = _transition_counts(snapshot) + evidence_counts = { + event_class: sum(1 for event in events if event.get("event_class") == event_class) + for event_class in ("audit.allow", "audit.deny") + } + reconciliation = ReconciliationView( + source_counts=source_counts, + evidence_counts=evidence_counts, + ) + if source_counts != evidence_counts: + corrections.append("reconciliation source counts do not match captured request events") + + if events: + notes.append( + f"instance={next(iter(instance_ids))} sequence=1..{sequences[-1]} " + f"heartbeats={len(heartbeats)}" + ) + notes.append(f"source_counts={source_counts} evidence_counts={evidence_counts}") + return heartbeats, reconciliation, tuple(notes), tuple(dict.fromkeys(corrections)) + + +def _transition_counts(payload: Mapping[str, Any]) -> dict[str, int]: + counts = payload.get("source_transition_counts") + if not isinstance(counts, Mapping) or any( + type(counts.get(key)) is not int or counts[key] < 0 + for key in ("audit.allow", "audit.deny") + ): + raise ValueError("source transition counts must include nonnegative allow and deny counts") + return {key: counts[key] for key in ("audit.allow", "audit.deny")} diff --git a/src/kings_guard/posture.py b/src/kings_guard/posture.py index c07c643..84b85e7 100644 --- a/src/kings_guard/posture.py +++ b/src/kings_guard/posture.py @@ -146,7 +146,9 @@ class PostureEvaluator: findings.append("unexpected_egress_destination") if observation.identity_binding and observation.identity_binding != "verified_token": - tolerance = _matching_tolerance(genome, "identity_binding", observation.identity_binding) + tolerance = _matching_tolerance( + genome, "identity_binding", observation.identity_binding + ) if tolerance is None: findings.append("unverified_identity_binding") else: @@ -357,9 +359,8 @@ def _build_rationale( ) -> str: if posture is PostureLevel.HEALTHY: if tolerated_findings: - content = ( - "Healthy posture with tolerated deviations only: " - + ", ".join(tolerated_findings) + content = "Healthy posture with tolerated deviations only: " + ", ".join( + tolerated_findings ) else: content = "Healthy posture: observation is compatible with declared intent." @@ -375,10 +376,7 @@ def _build_rationale( if completeness is StreamCompleteness.COMPLETE: stream_text = "Stream completeness is complete." elif completeness is StreamCompleteness.DEGRADED: - stream_text = ( - "This judgment rests on a stream I cannot vouch for " - f"({completeness_reason})." - ) + stream_text = f"This judgment rests on a stream I cannot vouch for ({completeness_reason})." else: stream_text = ( "Stream completeness is unknown; confidence scores the record, " diff --git a/src/kings_guard/stream.py b/src/kings_guard/stream.py index 308d913..a8542a7 100644 --- a/src/kings_guard/stream.py +++ b/src/kings_guard/stream.py @@ -31,6 +31,14 @@ def evaluate_stream( now_dt = _as_datetime(now) findings: list[str] = [] reasons: list[str] = [] + observations = tuple( + item for item in observations + if item.source_system == cadence.source_system and parse_timestamp(item.timestamp) <= now_dt + ) + heartbeats = tuple( + item for item in heartbeats + if item.source_system == cadence.source_system and parse_timestamp(item.timestamp) <= now_dt + ) observed_counts = _count_observations(observations) watch_start = _as_datetime(watching_since) if watching_since is not None else None @@ -49,6 +57,8 @@ def evaluate_stream( ) observed_counts[rate.event_class] = count watched_long_enough = (now_dt - watch_start) >= rate.window + if not watched_long_enough: + reasons.append(f"rate window for {rate.event_class} has not been fully observed") if watched_long_enough and count < rate.expected_min: findings.append(f"{STREAM_FINDING_PREFIX}cadence_unmet:{rate.event_class}") reasons.append( @@ -78,16 +88,19 @@ def evaluate_stream( if cadence.reconciliations: if reconciliation is None: - reasons.append( - "reconciliation view was not supplied; divergence cannot be ruled out" - ) + reasons.append("reconciliation view was not supplied; divergence cannot be ruled out") else: for spec in cadence.reconciliations: - source_count = int(reconciliation.source_counts.get(spec.covered_event_class, 0)) - evidence_count = int( - reconciliation.evidence_counts.get(spec.covered_event_class, 0) - ) - if evidence_count < source_count: + source_count = reconciliation.source_counts.get(spec.covered_event_class) + evidence_count = reconciliation.evidence_counts.get(spec.covered_event_class) + if any(type(value) is not int or value < 0 + for value in (source_count, evidence_count)): + reasons.append( + "valid reconciliation counts were not supplied for " + f"{spec.covered_event_class}" + ) + continue + if evidence_count != source_count: findings.append( f"{STREAM_FINDING_PREFIX}reconciliation_divergence:{spec.covered_event_class}" ) @@ -98,9 +111,7 @@ def evaluate_stream( completeness = _completeness(findings, reasons, cadence) reason = ( - "; ".join(reasons) - if reasons - else "declared cadence is met and no stream finding is open" + "; ".join(reasons) if reasons else "declared cadence is met and no stream finding is open" ) return StreamAssessment( completeness=completeness, @@ -118,9 +129,7 @@ def _completeness( ) -> StreamCompleteness: if findings: return StreamCompleteness.DEGRADED - heartbeat_pending = any("not yet due" in item for item in reasons) - reconciliation_unsupplied = any("was not supplied" in item for item in reasons) - if heartbeat_pending or reconciliation_unsupplied: + if reasons: return StreamCompleteness.UNKNOWN if not cadence.heartbeats and not cadence.rates and not cadence.reconciliations: return StreamCompleteness.UNKNOWN diff --git a/tests/test_cadence_declaration.py b/tests/test_cadence_declaration.py index aaab850..78e17ba 100644 --- a/tests/test_cadence_declaration.py +++ b/tests/test_cadence_declaration.py @@ -1,7 +1,12 @@ from pathlib import Path +import pytest + +from kings_guard.cadence import load_qonto_assistant_source_cadence + ROOT = Path(__file__).resolve().parents[1] DRAFT = ROOT / "specs" / "EmissionCadenceDeclaration.md" +QONTO_SOURCE_DECLARATION = Path("/home/worsch/qonto-assistant/specs/audit-emission-cadence.yaml") def test_emission_cadence_draft_exists_for_taxonomy_handover() -> None: @@ -17,3 +22,62 @@ def test_emission_cadence_draft_exists_for_taxonomy_handover() -> None: assert "reconciliation" in text.lower() assert "alongside the security genome" in text.lower() or "alongside the genome" in text.lower() assert "security_genome" in text + + +def test_qonto_source_declaration_is_consumed_separately_from_the_taxonomy_draft() -> None: + cadence = load_qonto_assistant_source_cadence() + + assert cadence.status == "source-declared" + assert cadence.owner == "qonto-assistant" + assert cadence.reference_instance == "QONTO-WP-0005" + assert cadence.forms() == {"heartbeat-or-reconciliation"} + assert cadence.rates == () + assert cadence.heartbeats[0].interval.total_seconds() == 86400 + + +@pytest.mark.skipif( + not QONTO_SOURCE_DECLARATION.is_file(), + reason="qonto-assistant source declaration checkout is unavailable", +) +def test_qonto_source_cadence_fixture_matches_the_source_owned_declaration() -> None: + yaml = pytest.importorskip("yaml") + source = yaml.safe_load(QONTO_SOURCE_DECLARATION.read_text(encoding="utf-8"))[ + "audit_emission_cadence" + ] + cadence = load_qonto_assistant_source_cadence() + + assert cadence.source_system == source["source"] + assert cadence.heartbeats[0].event_class == source["heartbeat"]["event_class"] + assert ( + cadence.heartbeats[0].interval.total_seconds() + == source["heartbeat"]["default_interval_seconds"] + ) + assert source["event_classes"]["audit.deny"]["form"] in cadence.forms() + assert source["event_classes"]["audit.allow"]["completeness_claimed"] is False + assert cadence.rates == () + + +@pytest.mark.parametrize("filename", [ + "qonto_assistant_cadence.json", "qonto_assistant_source_cadence.json", +]) +def test_cadence_fixtures_conform_to_owner_schema(filename): + import json + + import jsonschema + import yaml + + schema_path = Path( + "/home/worsch/info-tech-canon/infospace/schemas/emission-cadence.schema.yaml" + ) + if not schema_path.is_file(): + pytest.skip("canonical InfoTechCanon schema checkout is unavailable") + schema = yaml.safe_load(schema_path.read_text()) + payload = json.loads((ROOT / "src/kings_guard/fixtures" / filename).read_text()) + jsonschema.Draft202012Validator(schema).validate(payload) + + +@pytest.mark.parametrize(("value", "seconds"), [("P1DT2H30M5S", 95405), ("P1D", 86400)]) +def test_canonical_compound_durations_are_consumed(value, seconds): + from kings_guard.cadence import parse_interval + + assert parse_interval(value).total_seconds() == seconds diff --git a/tests/test_completeness.py b/tests/test_completeness.py index acddb1e..ffe4f0e 100644 --- a/tests/test_completeness.py +++ b/tests/test_completeness.py @@ -1,3 +1,5 @@ +from helpers import load_pilot, observation_from_fixture + from kings_guard.cadence import load_qonto_assistant_cadence from kings_guard.contracts import ( PostureAssessment, @@ -9,8 +11,6 @@ from kings_guard.contracts import ( ) from kings_guard.posture import PostureEvaluator -from helpers import load_pilot, observation_from_fixture - def _healthy_assessment(*, completeness: StreamCompleteness, confidence: int) -> PostureAssessment: return PostureAssessment( @@ -33,7 +33,10 @@ def test_completeness_is_separated_from_record_richness() -> None: assert evaluation.assessment.confidence_score >= 85 assert evaluation.assessment.stream_completeness is StreamCompleteness.UNKNOWN assert "record" in evaluation.assessment.completeness_reason - assert "cannot vouch" in evaluation.assessment.rationale or "unknown" in evaluation.assessment.rationale + assert ( + "cannot vouch" in evaluation.assessment.rationale + or "unknown" in evaluation.assessment.rationale + ) def test_unmet_cadence_degrades_completeness_and_says_so_in_words() -> None: @@ -47,7 +50,9 @@ def test_unmet_cadence_degrades_completeness_and_says_so_in_words() -> None: now="2026-07-24T10:10:00Z", watching_since="2026-07-22T09:10:00Z", heartbeats=(), - reconciliation=ReconciliationView(source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1}), + reconciliation=ReconciliationView( + source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1} + ), ) assert evaluation.assessment.stream_completeness is StreamCompleteness.DEGRADED @@ -56,12 +61,8 @@ def test_unmet_cadence_degrades_completeness_and_says_so_in_words() -> None: def test_incomplete_stream_is_never_more_trustworthy_than_a_complete_one() -> None: - rich_incomplete = _healthy_assessment( - completeness=StreamCompleteness.DEGRADED, confidence=95 - ) - sparse_complete = _healthy_assessment( - completeness=StreamCompleteness.COMPLETE, confidence=70 - ) + rich_incomplete = _healthy_assessment(completeness=StreamCompleteness.DEGRADED, confidence=95) + sparse_complete = _healthy_assessment(completeness=StreamCompleteness.COMPLETE, confidence=70) unknown = _healthy_assessment(completeness=StreamCompleteness.UNKNOWN, confidence=95) assert assessment_trust_key(rich_incomplete) < assessment_trust_key(sparse_complete) @@ -103,7 +104,9 @@ def test_complete_stream_keeps_richness_and_states_completeness() -> None: counts={"audit.deny": 1}, ), ), - reconciliation=ReconciliationView(source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1}), + reconciliation=ReconciliationView( + source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1} + ), ) assert evaluation.assessment.stream_completeness is StreamCompleteness.COMPLETE diff --git a/tests/test_contracts.py b/tests/test_contracts.py index 2f73f64..909d9b0 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -1,8 +1,8 @@ +from helpers import load_pilot, observation_from_fixture + from kings_guard.contracts import PostureLevel, SecurityGenome, as_jsonable from kings_guard.posture import PostureEvaluator -from helpers import load_pilot, observation_from_fixture - def test_qonto_fixture_loads_a_normalized_genome() -> None: fixture = load_pilot() @@ -25,6 +25,18 @@ def test_qonto_audit_event_normalizes_to_immune_observation() -> None: assert observation.deny_reason == "credential_exfil" +def test_qonto_source_identity_and_egress_override_legacy_mapping_hints() -> None: + fixture = load_pilot() + event = dict(fixture.audit_event) + event["identity_binding"] = "key_cape_jwt" + event["egress_destination"] = "qonto-proxy" + + observation = observation_from_fixture(fixture, event) + + assert observation.identity_binding == "key_cape_jwt" + assert observation.egress_destination == "qonto-proxy" + + def test_posture_evaluation_is_jsonable() -> None: fixture = load_pilot() observation = observation_from_fixture(fixture) @@ -34,3 +46,11 @@ def test_posture_evaluation_is_jsonable() -> None: assert payload["assessment"]["posture"] == PostureLevel.INFLAMED.value assert payload["signals"][0]["signal_kind"] == "posture_hint" assert payload["assessment"]["stream_completeness"] == "unknown" + + +def test_explicitly_missing_source_context_does_not_recover_legacy_hints() -> None: + fixture = load_pilot() + event = dict(fixture.audit_event, identity_binding=None, egress_destination=None) + observation = observation_from_fixture(fixture, event) + assert observation.identity_binding is None + assert observation.egress_destination is None diff --git a/tests/test_effector_origin.py b/tests/test_effector_origin.py index 7dab239..acb3549 100644 --- a/tests/test_effector_origin.py +++ b/tests/test_effector_origin.py @@ -1,6 +1,6 @@ -from kings_guard.contracts import StreamCompleteness - from helpers import load_pilot, observation_from_fixture + +from kings_guard.contracts import StreamCompleteness from kings_guard.posture import PostureEvaluator diff --git a/tests/test_evidence_class.py b/tests/test_evidence_class.py index a774cd6..33924bb 100644 --- a/tests/test_evidence_class.py +++ b/tests/test_evidence_class.py @@ -1,10 +1,9 @@ import pytest +from helpers import load_pilot, observation_from_fixture from kings_guard.adapters import observation_from_audit_event from kings_guard.contracts import DeclaredEvidenceSource, EvidenceClass -from helpers import load_pilot, observation_from_fixture - def test_genome_declares_both_evidence_classes() -> None: genome = load_pilot().genome diff --git a/tests/test_layer_conformance.py b/tests/test_layer_conformance.py index f57b831..9ed4674 100644 --- a/tests/test_layer_conformance.py +++ b/tests/test_layer_conformance.py @@ -5,6 +5,7 @@ Tooling-layer client. A test that only ran the checker against a clean tree would prove nothing — it would pass just as happily if the checker were broken. So the negative case is exercised too, on a synthetic tree. """ + from __future__ import annotations import subprocess diff --git a/tests/test_live_qonto_observation.py b/tests/test_live_qonto_observation.py index 00d6bf5..e751d04 100644 --- a/tests/test_live_qonto_observation.py +++ b/tests/test_live_qonto_observation.py @@ -1,12 +1,13 @@ from __future__ import annotations import pytest +from helpers import load_pilot, observation_from_fixture +from kings_guard.cadence import load_qonto_assistant_source_cadence +from kings_guard.contracts import StreamCompleteness from kings_guard.live import capture_qonto_assistant_events, qonto_assistant_available from kings_guard.posture import PostureEvaluator -from helpers import load_pilot, observation_from_fixture - pytestmark = pytest.mark.skipif( not qonto_assistant_available(), reason="qonto-assistant checkout is required for live observation", @@ -28,7 +29,10 @@ def test_real_emitted_qonto_events_reach_the_evaluator() -> None: assert "allow" in decisions assert "deny" in decisions assert capture.mapping_notes - assert all(observation.source_system == "qonto-assistant" for observation in capture.observations) + assert capture.stream_notes + assert all( + observation.source_system == "qonto-assistant" for observation in capture.observations + ) deny = next(item for item in capture.observations if item.decision.value == "deny") evaluation = PostureEvaluator().evaluate(fixture.genome, deny) @@ -39,8 +43,25 @@ def test_real_emitted_qonto_events_reach_the_evaluator() -> None: for request in signal.effector_requests: assert request.authority_boundary in {"advisory_only", "metadata_only"} - # Mapping confirmed against the real emit path; remaining gaps are source omissions, - # not adapter drift. - assert any("identity_binding" in item for item in capture.corrections_for_source) - assert any("egress_destination" in item for item in capture.corrections_for_source) - assert not any("mapping drifted" in item for item in capture.corrections_for_source) + # The source now emits identity binding and egress directly. The adapter must + # preserve those values instead of masking them with its legacy mapping hints. + for event, observation in zip(capture.events, capture.observations, strict=True): + assert observation.identity_binding == event["identity_binding"] + assert observation.egress_destination == event["egress_destination"] + assert capture.corrections_for_source == () + assert capture.stream_corrections == () + + cadence = load_qonto_assistant_source_cadence() + stream_evaluation = PostureEvaluator().evaluate_with_stream( + fixture.genome, + deny, + cadence, + now=capture.heartbeats[-1].timestamp, + watching_since=capture.heartbeats[0].timestamp, + observations=capture.observations, + heartbeats=capture.heartbeats, + reconciliation=capture.reconciliation, + ) + assert stream_evaluation.assessment.stream_completeness is StreamCompleteness.COMPLETE + assert stream_evaluation.stream is not None + assert stream_evaluation.stream.findings == () diff --git a/tests/test_posture.py b/tests/test_posture.py index d2ec179..74b3841 100644 --- a/tests/test_posture.py +++ b/tests/test_posture.py @@ -1,7 +1,7 @@ -from kings_guard.posture import PostureEvaluator - from helpers import load_pilot, observation_from_fixture +from kings_guard.posture import PostureEvaluator + def test_qonto_pilot_produces_inflamed_posture_with_tolerance_context() -> None: fixture = load_pilot() diff --git a/tests/test_secrets_engine_snapshot.py b/tests/test_secrets_engine_snapshot.py new file mode 100644 index 0000000..a83864a --- /dev/null +++ b/tests/test_secrets_engine_snapshot.py @@ -0,0 +1,83 @@ +from copy import deepcopy +from dataclasses import asdict + +import pytest + +from kings_guard.adapters.secrets_engine import snapshot_from_engine +from kings_guard.contracts import StreamCompleteness + +NOW = "2026-09-05T12:00:00Z" +PAYLOAD = { + "surface": "secret-use-evidence", "as_of": NOW, "completeness_claimed": False, + "cadence": {"form": "heartbeat", "interval": "1d"}, + "lanes": [{"as_of": NOW, "catalog_id": "lane-1", "stage": "test", "kind": "kv"}], +} +BINDINGS = {"lane-1": ("tenant-1", "subject-1")} + + +def admit(payload, now=NOW, bindings=BINDINGS): + return snapshot_from_engine(payload, lane_bindings=bindings, now=now) + + +def test_catalog_only_snapshot_does_not_invent_evidence_or_completeness(): + result = admit(PAYLOAD) + assert result.lanes[0].ready is None + assert result.lanes[0].revocation_succeeded is None + assert result.lanes[0].evidence_kind is None + assert result.completeness is StreamCompleteness.UNKNOWN + assert result.heartbeat_interval.total_seconds() == 86400 + + +def test_evidence_preserves_false_and_drops_authorization_and_sensitive_fields(): + payload = deepcopy(PAYLOAD) + payload["lanes"][0].update(ready=False, revocation_succeeded=False, + decision_id="dec-1", session_handle="session-1", + secret="never-retained", lifecycle_operation="revoke") + result = admit(payload) + assert result.lanes[0].ready is False + assert result.lanes[0].revocation_succeeded is False + assert result.lanes[0].lifecycle_operation == "revoke" + fields = asdict(result.lanes[0]) + assert {"decision_id", "session_handle", "secret"}.isdisjoint(fields) + assert result.completeness is StreamCompleteness.UNKNOWN + + +@pytest.mark.parametrize(("now", "findings"), [ + ("2026-09-06T12:00:00Z", ()), + ("2026-09-06T12:00:01Z", ("snapshot:stale",)), + ("2026-09-05T11:59:59Z", ("snapshot:future_timestamp",)), +]) +def test_freshness_does_not_upgrade_completeness(now, findings): + result = admit(PAYLOAD, now) + assert result.findings == findings + assert result.completeness is StreamCompleteness.UNKNOWN + + +def test_unbound_or_duplicate_lanes_cannot_cross_scope(): + with pytest.raises(ValueError, match="unbound"): + admit(PAYLOAD, bindings={}) + payload = deepcopy(PAYLOAD) + payload["lanes"].append(payload["lanes"][0]) + with pytest.raises(ValueError, match="duplicate"): + admit(payload) + + +@pytest.mark.parametrize("change", [ + {"completeness_claimed": True}, {"completeness_claimed": None}, + {"cadence": {"form": "heartbeat", "interval": "0s"}}, + {"surface": "authorization"}, {"lanes": None}, +]) +def test_invalid_envelope_is_rejected(change): + with pytest.raises(ValueError): + admit({**PAYLOAD, **change}) + + +@pytest.mark.parametrize("change", [ + {"ready": "false"}, {"revocation_attempted": 0}, + {"as_of": "2026-09-04T12:00:00Z"}, {"evidence_kind": "guessed"}, +]) +def test_invalid_lane_is_rejected(change): + payload = deepcopy(PAYLOAD) + payload["lanes"][0].update(change) + with pytest.raises(ValueError): + admit(payload) diff --git a/tests/test_stream.py b/tests/test_stream.py index 7d9490e..3d44a4b 100644 --- a/tests/test_stream.py +++ b/tests/test_stream.py @@ -1,3 +1,5 @@ +from helpers import load_pilot, observation_from_fixture + from kings_guard.cadence import load_qonto_assistant_cadence from kings_guard.contracts import ( ReconciliationView, @@ -7,8 +9,6 @@ from kings_guard.contracts import ( ) from kings_guard.posture import PostureEvaluator, is_stream_finding -from helpers import load_pilot, observation_from_fixture - def test_cadence_draft_covers_both_forms_against_qonto() -> None: cadence = load_qonto_assistant_cadence() @@ -36,7 +36,9 @@ def test_unmet_declared_rate_is_a_stream_finding() -> None: counts={"audit.deny": 0}, ), ), - reconciliation=ReconciliationView(source_counts={"audit.deny": 0}, evidence_counts={"audit.deny": 0}), + reconciliation=ReconciliationView( + source_counts={"audit.deny": 0}, evidence_counts={"audit.deny": 0} + ), ) assert any(item.startswith("stream:cadence_unmet:audit.allow") for item in stream.findings) @@ -54,7 +56,9 @@ def test_missing_heartbeat_is_a_stream_finding() -> None: now="2026-07-24T10:10:00Z", watching_since="2026-07-22T09:10:00Z", heartbeats=(), - reconciliation=ReconciliationView(source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1}), + reconciliation=ReconciliationView( + source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1} + ), ) assert any(item.startswith("stream:heartbeat_missing:audit.deny") for item in stream.findings) @@ -103,7 +107,9 @@ def test_stream_findings_are_distinguishable_from_content_findings() -> None: now="2026-07-24T10:10:00Z", watching_since="2026-07-22T09:10:00Z", heartbeats=(), - reconciliation=ReconciliationView(source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1}), + reconciliation=ReconciliationView( + source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1} + ), ) content_findings = set(evaluation.assessment.findings) @@ -127,7 +133,9 @@ def test_heartbeat_not_yet_due_does_not_false_alarm() -> None: now=observation.timestamp, watching_since=observation.timestamp, heartbeats=(), - reconciliation=ReconciliationView(source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1}), + reconciliation=ReconciliationView( + source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1} + ), ) assert not any(item.startswith("stream:heartbeat_missing") for item in stream.findings) diff --git a/tests/test_stream_admission.py b/tests/test_stream_admission.py new file mode 100644 index 0000000..2ed00bb --- /dev/null +++ b/tests/test_stream_admission.py @@ -0,0 +1,88 @@ +from copy import deepcopy + +import pytest + +from kings_guard.cadence import load_qonto_assistant_source_cadence +from kings_guard.contracts import ReconciliationView, StreamCompleteness, StreamHeartbeat +from kings_guard.live import _confirm_stream +from kings_guard.stream import evaluate_stream + +NOW = "2026-09-05T12:00:00Z" +COUNTS = {"audit.allow": 0, "audit.deny": 0} +EVENT = { + "stream_id": "qonto.audit", "stream_instance_id": "instance-1", "stream_sequence": 1, + "event_class": "audit.heartbeat", "timestamp": NOW, "assertion": "nothing-to-report", + "source_transition_counts": COUNTS, +} +SNAPSHOT = { + "stream_id": "qonto.audit", "stream_instance_id": "instance-1", + "last_stream_sequence": 1, "source_transition_counts": COUNTS, +} + + +@pytest.mark.parametrize("change", [ + {"stream_instance_id": "instance-2"}, {"stream_id": "other.audit"}, + {"last_stream_sequence": 2}, {"last_stream_sequence": True}, +]) +def test_reconciliation_must_belong_to_captured_stream(change): + assert _confirm_stream([EVENT], {**SNAPSHOT, **change})[-1] + + +@pytest.mark.parametrize("sequence", [None, True, "1", -1]) +def test_malformed_sequences_are_rejected(sequence): + with pytest.raises(ValueError, match="stream_sequence"): + _confirm_stream([{**EVENT, "stream_sequence": sequence}], SNAPSHOT) + + +def test_empty_capture_is_rejected_without_index_error(): + with pytest.raises(ValueError, match="empty"): + _confirm_stream([], SNAPSHOT) + + +def test_missing_counts_cannot_be_normalized_to_zero(): + snapshot = deepcopy(SNAPSHOT) + snapshot["source_transition_counts"] = {"audit.allow": 0} + with pytest.raises(ValueError, match="counts"): + _confirm_stream([EVENT], snapshot) + + +@pytest.mark.parametrize("counts", [{}, {"audit.deny": -1}, {"audit.deny": True}]) +def test_absent_or_invalid_reconciliation_counts_leave_completeness_unknown(counts): + result = evaluate_stream( + (), load_qonto_assistant_source_cadence(), now=NOW, + heartbeats=[StreamHeartbeat("qonto-assistant", NOW, "audit.heartbeat", + "nothing-to-report")], + reconciliation=ReconciliationView(counts, counts), + ) + assert result.completeness is StreamCompleteness.UNKNOWN + + +@pytest.mark.parametrize(("source", "timestamp"), [ + ("other-source", NOW), ("qonto-assistant", "2026-09-07T12:00:00Z"), +]) +def test_unrelated_or_future_heartbeat_cannot_satisfy_cadence(source, timestamp): + result = evaluate_stream( + (), load_qonto_assistant_source_cadence(), now=NOW, + watching_since="2026-09-03T12:00:00Z", + heartbeats=[StreamHeartbeat(source, timestamp, "audit.heartbeat", "nothing-to-report")], + reconciliation=ReconciliationView(COUNTS, COUNTS), + ) + assert result.completeness is StreamCompleteness.DEGRADED + assert "stream:heartbeat_missing:audit.deny" in result.findings + + +def test_excess_evidence_is_also_reconciliation_divergence(): + result = evaluate_stream( + (), load_qonto_assistant_source_cadence(), now=NOW, + heartbeats=[StreamHeartbeat("qonto-assistant", NOW, "audit.heartbeat", + "nothing-to-report")], + reconciliation=ReconciliationView({"audit.deny": 0}, {"audit.deny": 1}), + ) + assert "stream:reconciliation_divergence:audit.deny" in result.findings + + +def test_heartbeat_counts_must_match_events_already_captured(): + event = {**EVENT, "source_transition_counts": {"audit.allow": 0, "audit.deny": 1}} + assert "heartbeat counts do not match preceding request events" in ( + _confirm_stream([event], SNAPSHOT)[-1] + ) diff --git a/workplans/ADHOC-2026-09-04.md b/workplans/ADHOC-2026-09-04.md new file mode 100644 index 0000000..b9313c3 --- /dev/null +++ b/workplans/ADHOC-2026-09-04.md @@ -0,0 +1,34 @@ +--- +id: KG-WP-ADHOC-2026-09-04 +type: workplan +title: "2026-09-04 repository consistency maintenance" +domain: infotech +repo: kings-guard +status: finished +owner: codex +topic_slug: netkingdom +created: "2026-09-04" +updated: "2026-09-04" +finished: "2026-09-04" +--- + +# 2026-09-04 Repository Consistency Maintenance + +## Reconcile the accepted layer declaration and the shipped secret-use surface + +```task +id: KG-WP-ADHOC-2026-09-04-T01 +status: done +priority: low +``` + +The completed KG-WP-0004 sweep left current boundary prose and checker +commentary pointing at v0.6 while the authoritative declaration and architecture +already use accepted v0.7. Align those references and make `SCOPE.md` say +explicitly that containment is proposed rather than owned. + +Also consume the 2026-09-04 coordination update from `secrets-engine`: record +that its non-secret `secret-use snapshot` surface and 1d heartbeat declaration +have shipped. Observation-input admission changes a posture lane and requires +mapping and contract decisions, so hand that residual to live intake +`KG-IN-0004` rather than silently treating the unmodeled surface as consumed. diff --git a/workplans/KG-WP-0005-qonto-source-cadence-admission.md b/workplans/KG-WP-0005-qonto-source-cadence-admission.md new file mode 100644 index 0000000..407a96a --- /dev/null +++ b/workplans/KG-WP-0005-qonto-source-cadence-admission.md @@ -0,0 +1,71 @@ +--- +id: KG-WP-0005 +type: workplan +title: "Admit and validate qonto-assistant source completeness evidence" +domain: infotech +repo: kings-guard +status: active +owner: codex +topic_slug: netkingdom +created: "2026-09-04" +updated: "2026-09-05" +origin: cross-repo +origin_ref: QONTO-WP-0005 +--- + +# Admit and Validate Qonto-Assistant Source Completeness Evidence + +Qonto-assistant has shipped the source-side contract requested by KG-WP-0003: +source-owned cadence, instance and sequence fields, startup/periodic/shutdown +heartbeats, and a reconciliation view. Kings-guard must consume and validate +those claims without confusing source transition counts with authorization or +claiming completeness beyond an observed process instance. + +## Consume source-emitted observation context + +```task +id: KG-WP-0005-T01 +status: done +priority: high +``` + +Prefer source-emitted `identity_binding` and `egress_destination` over legacy +normalization hints, retain the hints for old records, and prove the live +mapping does not drift. + +## Validate cadence, heartbeat, sequence, and reconciliation locally + +```task +id: KG-WP-0005-T02 +status: done +priority: high +``` + +Consume the source-owned cadence separately from kings-guard's Taxonomy draft. +Drive qonto-assistant's real AuditLogger emit path, require one stream instance +with contiguous sequence numbers, translate heartbeat claims, compare source +transition counts with captured request evidence, and prove the resulting deny +stream evaluates complete without findings. + +## Validate the deployed stream + +```task +id: KG-WP-0005-T03 +status: wait +priority: medium +``` + +Observe an authorized deployed qonto-assistant process from startup through at +least one allow/deny transition and reconciliation read. Confirm the deployed +configuration emits periodic heartbeats at the declared interval, preserves +sequence/instance semantics, and produces no unexplained divergence. This is +operational evidence; the local source-path proof does not substitute for it. + +**Review 2026-09-05:** Local source-path tests pass with source context, +heartbeat counters, sequence and reconciliation checks. The canonical cadence +fixture migration and rejection-path hardening are tracked in KG-WP-0006. +T03 remains waiting: the source workplan explicitly leaves deployment with the +runtime owner; this repository supplies no authorized deployed stream endpoint +or process-lifecycle observation route. Required input is a runtime-owner +capture from startup, a request transition, periodic heartbeat timing, and a +same-instance reconciliation snapshot. No cluster or Tooling client was used. diff --git a/workplans/KG-WP-0006-observation-input-review.md b/workplans/KG-WP-0006-observation-input-review.md new file mode 100644 index 0000000..1ee8fa9 --- /dev/null +++ b/workplans/KG-WP-0006-observation-input-review.md @@ -0,0 +1,77 @@ +--- +id: KG-WP-0006 +type: workplan +title: "Review and admit canonical cadence and secret-use snapshot inputs" +domain: infotech +repo: kings-guard +status: finished +owner: codex +topic_slug: netkingdom +created: "2026-09-05" +updated: "2026-09-05" +origin: cross-repo +origin_ref: KG-IN-0004 +--- + +# Observation input review + +## Migrate the cadence consumer to the published generic contract + +```task +id: KG-WP-0006-T01 +status: done +priority: high +``` + +Consume InfoTechCanon standard/emission-cadence 0.1. Move profile and +provenance fields into namespaced extensions; validate examples against the +owner's schema without copying it. Keep the handover draft historical. + +## Harden Qonto stream admission and completeness checks + +```task +id: KG-WP-0006-T02 +status: done +priority: high +``` + +Reject mismatched instances, malformed sequences and counters; preserve missing +context; prevent absent reconciliation fields or unrelated/future heartbeats +from manufacturing completeness. Cover failure paths with regression tests. + +## Admit secret-use snapshots with explicit scope and unknown completeness + +```task +id: KG-WP-0006-T03 +status: done +priority: medium +``` + +Map the Engine envelope into a separate snapshot observation: it has no event +actor, event timestamp, or authorization outcome. Require explicit lane scope, +retain only typed metadata, preserve omitted evidence fields, translate the +heartbeat interval as a freshness bound, and never promote snapshot freshness +to stream completeness. Record residual source evidence needed for posture. + +## Completion evidence + +Reviewed the original pending Qonto implementation and added negative-path +coverage for scoped streams, absent/invalid counters, snapshot instance binding, +sequence validation, and heartbeat reconciliation. Both migrated fixtures pass +InfoTechCanon's external schema 0.1. Snapshot mapping and limitations are in +`docs/SecretUseSnapshotAdmission.md`. + +Residual KG-IN-0005 was recorded before completion for secret-use event +provenance and operational evidence. Deployed Qonto acceptance remains explicitly +waiting in KG-WP-0005-T03; local source-path tests do not close it. The NetKingdom +profile remains proposed in its owner's repository; no adoption is asserted. + +Validation: 78 tests passed (including external canonical schema and local +Qonto emit-path tests), lint and layer conformance passed, pilot demo passed, +and `git diff --check` was clean. Decision: KG-DEC-2026-003. + +State Hub: decision KG-DEC-2026-003 was recorded successfully. Two required +`statehub fix-consistency --repo kings-guard` attempts (including an unrestricted +retry) failed with C-00: repository API lookup timed out. Workplan/intake files +remain authoritative; the operator must rerun synchronization when the repository +lookup is available. This is a synchronization failure, not implementation evidence.