diff --git a/.gitignore b/.gitignore index 867b97e..5220c9f 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,12 @@ __pycache__/ build/ dist/ *.egg-info/ +.env +.env.* +credentials.json +secrets.yaml +secrets.yml +id_rsa +id_ed25519 +*.pem +*.key diff --git a/INTENT.md b/INTENT.md index 4caa01b..dbafc28 100644 --- a/INTENT.md +++ b/INTENT.md @@ -292,10 +292,11 @@ The repository should evolve through clear layers: immune memory. 2. **Assessment loop:** provide a minimal service that ingests observations, evaluates posture against declared intent, and produces typed signals — - against **real emitted events**, not fixtures. §12's fourth step - ("kings-guard observes it in operation") is the estate's, and it is - unstaffed until this stage is live. Completeness of the stream is part of - the judgment, not an assumption about it (§9.6). + against **real emitted events**, not only fixtures. §12's fourth step + ("kings-guard observes it in operation") is staffed for the + `qonto-assistant` audit stream. Completeness of the stream is part of + the judgment, not an assumption about it (§9.6). The deny class still + lacks a source-published heartbeat, so completeness is not claimed. 3. **Bounded proposal:** emit containment *requests* — isolation, throttling, revocation, reconstitution — as typed, authority-bounded proposals to the engine that renders them. kings-guard never actuates (§9.2); the actuation diff --git a/README.md b/README.md index 2f743fc..bcff268 100644 --- a/README.md +++ b/README.md @@ -21,10 +21,12 @@ anything. It provides: - typed contracts for security genome, phenotype, observation, posture, signal, effector request, and immune memory entry; -- a minimal posture evaluator that turns a normalized observation into a - posture assessment and bounded response hints; -- a fixture-driven pilot based on `qonto-assistant`, chosen because it already - exposes a security genome record, an audit stream, and a fast local loop. +- evidence-class and stream-completeness fields, so silence is a finding and + record richness is not mistaken for a complete stream; +- a posture evaluator that judges one observation and, separately, the stream + it came from; +- a `qonto-assistant` pilot that still has a fixture regression case and can + also consume real events from that service's own AuditLogger. ## Repo layout diff --git a/SCOPE.md b/SCOPE.md index 59fd958..22c8978 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -96,13 +96,16 @@ does it decide and act?** The first is in scope. The second is another layer's. - The implementation currently evaluates normalized observations and emits posture/signal results for one bounded pilot lane; it is not an enforcement service and will not become one. +- Evidence class, stream-completeness evaluation, origin-linked proposals, and + the emission-cadence Taxonomy draft landed in `KG-WP-0003`. - `layer.yaml`, `scripts/check_layer_conformance.py` and - `tests/test_layer_conformance.py` make the no-Tooling-client claim executable; - the companion cites them estate-wide as the reference for a repository with no - Tooling contacts at all. -- **Every input is still a fixture.** No real emitted event has reached the - evaluator, so statute §12's fourth step remains unstaffed. Closing that is - `KG-WP-0003`. + `tests/test_layer_conformance.py` make the no-Tooling-client claim and the + 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. --- diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 97e5ae8..94e9c32 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -10,7 +10,7 @@ | --- | --- | --- | --- | --- | | workplan | KG-WP-0001 | finished | — | workplans/KG-WP-0001-statehub-bootstrap.md | | workplan | KG-WP-0002 | finished | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | -| workplan | KG-WP-0003 | ready | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | +| workplan | KG-WP-0003 | finished | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | | task | KG-WP-0001-T01 | done | — | workplans/KG-WP-0001-statehub-bootstrap.md | | task | KG-WP-0001-T02 | done | — | workplans/KG-WP-0001-statehub-bootstrap.md | | task | KG-WP-0001-T03 | done | — | workplans/KG-WP-0001-statehub-bootstrap.md | @@ -18,15 +18,16 @@ | task | KG-WP-0002-T02 | done | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | | task | KG-WP-0002-T03 | done | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | | task | KG-WP-0002-T04 | done | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | -| task | KG-WP-0003-T01 | todo | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | -| task | KG-WP-0003-T02 | todo | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | -| task | KG-WP-0003-T03 | todo | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | -| task | KG-WP-0003-T04 | todo | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | -| task | KG-WP-0003-T05 | todo | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | -| task | KG-WP-0003-T06 | todo | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | -| task | KG-WP-0003-T07 | todo | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | -| task | KG-WP-0003-T08 | todo | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | +| task | KG-WP-0003-T01 | done | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | +| task | KG-WP-0003-T02 | done | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | +| task | KG-WP-0003-T03 | done | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | +| task | KG-WP-0003-T04 | done | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | +| task | KG-WP-0003-T05 | done | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | +| task | KG-WP-0003-T06 | done | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | +| task | KG-WP-0003-T07 | done | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | +| task | KG-WP-0003-T08 | done | — | workplans/KG-WP-0003-evidence-completeness-and-live-observation.md | | intake | KG-IN-0001 | closed | — | intakes/intakes.md | | intake | KG-IN-0002 | open | — | intakes/intakes.md | -| intake | KG-IN-0003 | open | — | intakes/intakes.md | +| intake | KG-IN-0003 | closed | — | intakes/intakes.md | | decision | KG-DEC-2026-001 | resolved | — | decisions/decisions.md | +| decision | KG-DEC-2026-002 | resolved | — | decisions/decisions.md | diff --git a/docs/AdjacentSystemBoundary.md b/docs/AdjacentSystemBoundary.md index d4188b7..31f08f3 100644 --- a/docs/AdjacentSystemBoundary.md +++ b/docs/AdjacentSystemBoundary.md @@ -37,6 +37,15 @@ The following rules apply to every integration below: the lane stays fixture-driven and the gap is declared in `INTENT.md`. 6. `kings-guard` never renders or caches an authorization decision. `access-engine` is the estate's only decision point (layer model §6). +7. Every containment **proposal** carries the originating observation and + signal identity, the stream-completeness state, and a restrictive + direction. It does not carry credentials, secret values, or a direct + actuation instruction. The receiving Engine (today: `access-engine` as + decision point; actuation still unowned) MUST retain those origin + references on the eventual decision record so a containment action is + reconstructable as a decision, not a side channel (statute §9.2; + `gate-house/docs/contracts/posture-findings-return.md`). Origin linkage + does not widen authority. ## 3. System-by-System Boundary @@ -97,4 +106,7 @@ Mechanically checkable, per layer model §10: no OpenBao client, no Kubernetes client; - every `EffectorRequest` carries an explicit `authority_boundary`, and the values in use are `advisory_only` and `metadata_only`; +- every `EffectorRequest` also carries `originating_observation_id` and + `originating_signal_id` so a later decision record can name what was + proposed and against which observation; - no module exposes an authorization decision surface. diff --git a/docs/pilots/QontoAssistantPosturePilot.md b/docs/pilots/QontoAssistantPosturePilot.md index fa1e747..f23e207 100644 --- a/docs/pilots/QontoAssistantPosturePilot.md +++ b/docs/pilots/QontoAssistantPosturePilot.md @@ -73,6 +73,7 @@ client request | `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 | +| 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/layer.yaml b/layer.yaml index 70b7746..24c8f4d 100644 --- a/layer.yaml +++ b/layer.yaml @@ -32,6 +32,28 @@ agent_principal_rules: memory_is_not_a_state_plane: true reconstructable_as_caller: true +# Honest split between what is mechanically checked and what remains an +# assertion. A claim in this file is not a test; the `form` column is. +agent_principal_rule_checks: + no_standing_credential: + form: test + checked_by: scripts/check_layer_conformance.py + tool_use_shapes: + form: assertion + note: >- + No third route is claimed. The Tooling-client scan covers the Tooling + half of "no third route". Distinguishing an Engine API call from an + ordinary Python call is not mechanical here; remaining assertion. + memory_is_not_a_state_plane: + form: test + checked_by: tests/test_immune_memory.py + reconstructable_as_caller: + form: mixed + tested: EffectorRequest origin fields (KG-WP-0003-T06) + remainder: >- + Full reconstructability of every agent action as the caller's is still + assertion; session and harness traces are glas-harness's. + catalog_entry: owns: - adaptive defence and judgment diff --git a/pyproject.toml b/pyproject.toml index 7d3cf5d..ec8e8ac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ where = ["src"] [tool.setuptools.package-data] kings_guard = ["fixtures/*.json"] + [tool.pytest.ini_options] testpaths = ["tests"] addopts = [ diff --git a/scripts/check_layer_conformance.py b/scripts/check_layer_conformance.py index 2b7ffdb..bb306f5 100644 --- a/scripts/check_layer_conformance.py +++ b/scripts/check_layer_conformance.py @@ -25,6 +25,7 @@ from __future__ import annotations import argparse import ast +import re import sys from datetime import date from pathlib import Path @@ -53,6 +54,28 @@ TOOLING_IMPORTS = { "docker": "container runtime client", } +# §3.4 rule 1 — no standing credential held in the repository or its +# configuration. Filenames that would be a standing secret, and assignments +# of well-known secret env vars to string literals in src/. +CREDENTIAL_FILENAMES = { + ".env", + ".env.local", + ".env.production", + "credentials.json", + "secrets.yaml", + "secrets.yml", + "id_rsa", + "id_ed25519", + "id_ecdsa", +} +CREDENTIAL_LITERAL = re.compile( + r"""(?x) + \b(?:VAULT_TOKEN|OPENBAO_TOKEN|BAO_TOKEN|AWS_SECRET_ACCESS_KEY| + PRIVATE_KEY|BEGIN\ (?:RSA\ )?PRIVATE\ KEY) + """ +) +SKIP_CREDENTIAL_SCAN_DIRS = {".git", ".venv", "__pycache__", ".pytest_cache", ".ruff_cache"} + def load_declaration() -> dict: if not DECL.exists(): @@ -98,6 +121,27 @@ def scan() -> list[tuple[Path, str, str]]: return hits +def scan_standing_credentials() -> list[tuple[Path, str]]: + """§3.4 rule 1: no standing credential in the repository or its config.""" + hits: list[tuple[Path, str]] = [] + for path in ROOT.rglob("*"): + if not path.is_file(): + continue + if any(part in SKIP_CREDENTIAL_SCAN_DIRS for part in path.parts): + continue + if path.name in CREDENTIAL_FILENAMES: + hits.append((path, f"credential-shaped file {path.name}")) + continue + if path.suffix in {".pem", ".key"} and "test" not in path.parts: + hits.append((path, f"key material file {path.name}")) + if SRC.is_dir(): + for path in sorted(SRC.rglob("*.py")): + text = path.read_text(encoding="utf-8") + if CREDENTIAL_LITERAL.search(text): + hits.append((path, "standing-credential literal or private-key block")) + return hits + + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--report", action="store_true", help="print the declaration summary") @@ -106,8 +150,11 @@ def main() -> int: decl = load_declaration() declared = {c.get("id") for c in decl.get("tooling_contacts") or []} hits = scan() + credential_hits = scan_standing_credentials() undeclared = [h for h in hits if h[1] not in declared] + rules = decl.get("agent_principal_rules") or {} + checks = decl.get("agent_principal_rule_checks") or {} if args.report: print(f"kings-guard — layer {decl['layer']}, standard v{decl['standard_version']}") @@ -121,6 +168,10 @@ def main() -> int: if review and date.fromisoformat(str(review)) < today: stale = " [REVIEW OVERDUE]" print(f" - {cap['id']}: {cap.get('owner_status', '?')}{stale}") + print(" agent-principal rule checks (§3.4):") + for name, meta in checks.items(): + form = meta.get("form", "unspecified") if isinstance(meta, dict) else "unspecified" + print(f" - {name}: {form} (claimed={rules.get(name)})") if undeclared: print("", file=sys.stderr) @@ -134,8 +185,29 @@ def main() -> int: print(" engine gap — do not declare this to make the check pass.", file=sys.stderr) return 1 + if credential_hits: + print("", file=sys.stderr) + print("FAIL: standing credential material (§3.4 rule 1)", file=sys.stderr) + for path, what in credential_hits: + try: + rel = path.relative_to(ROOT) + except ValueError: + rel = path + print(f" {rel}: {what}", file=sys.stderr) + print("", file=sys.stderr) + print(" Authority is per task, time-bounded, and attributable to the", file=sys.stderr) + print(" principal acted for. Do not hold a standing secret here.", file=sys.stderr) + return 1 + + if rules.get("no_standing_credential") is not True: + print("FAIL: layer.yaml does not claim no_standing_credential (§3.4 rule 1)", file=sys.stderr) + return 2 + if not args.report: - print(f"OK: no direct Tooling client in {SRC.relative_to(ROOT)} (§5, §11)") + print( + f"OK: no direct Tooling client in {SRC.relative_to(ROOT)}; " + "no standing credential (§5, §11, §3.4 rule 1)" + ) return 0 diff --git a/specs/EmissionCadenceDeclaration.md b/specs/EmissionCadenceDeclaration.md new file mode 100644 index 0000000..09a8a83 --- /dev/null +++ b/specs/EmissionCadenceDeclaration.md @@ -0,0 +1,134 @@ +--- +title: Emission-cadence declaration (Taxonomy draft) +document_id: KG-DRAFT-EMISSION-CADENCE +version: 0.1.0 +status: Draft-for-handover +date: 2026-09-01 +drafter: kings-guard +owner: Taxonomy +worked_example: qonto-assistant +reference_instance: GH-WP-0002-T04 +reference_source_declaration: approval-engine/cadence.yaml +statute: net-kingdom/canon/standards/security-layer-model_v0.7.md +sections: ["9.6", "17"] +classification: Public +--- + +# Emission-cadence declaration — Taxonomy draft + +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 +schema. Inventing a local shape and keeping it here would be the drift +§17 exists to prevent. + +Proposed Taxonomy homes, neither of which has assented (§17): + +- `info-tech-canon` — ecosystem-wide semantic contracts +- `net-kingdom` — NetKingdom standards of record + +gate-house already named the heartbeat form as doctrine in +`docs/contracts/approval-emission-detection.md` (`GH-WP-0002-T04`). The +reference source declaration is `approval-engine/cadence.yaml`. This draft +covers **both** forms, because one does not substitute for the other. + +## 1. Why this belongs alongside the security genome + +A source already declares intended healthy operation as a `security_genome`: +purpose, permitted capabilities, egress, tolerances. Expected emission cadence +is a claim of the same kind — a statement of intent about the evidence the +source will publish, not a measurement kings-guard takes of it. + +Putting the declaration next to the genome keeps three properties: + +1. **The class is the source's declaration.** Load-bearing versus attributive + is not inferred by the observer from event contents. +2. **The form follows the class.** Volume classes declare an expected rate. + Low-volume load-bearing classes declare a heartbeat or reconciliation. + Rate monitoring is forbidden for the latter: a suppressed month of + revocations is indistinguishable from a quiet one. +3. **Silence becomes observable without Tooling contact.** The source + publishes the stream, including the positive claim that can itself go + missing. kings-guard compares what arrived against what was declared. + +## 2. Both forms + +### 2.1 Expected rate — volume classes + +Use when the class has a rate that can drop. + +```yaml +form: expected-rate +event_class: +evidence_class: attributive | load-bearing +window: 24h +expected_min: +drop_below: finding +``` + +A drop below `expected_min` in `window` is a finding about the **stream**, +not about any record's contents. + +Load-bearing volume classes MUST declare this form (or the heartbeat form +if they are in fact low-volume). Attributive sources SHOULD. + +### 2.2 Heartbeat or reconciliation — low-volume load-bearing classes + +Use when the valuable event is rare: revocations, denials, containment +actions. Rate monitoring cannot work. The required property is a **positive +claim that can itself go missing**. + +Reference instance: `GH-WP-0002-T04` / `approval-engine/cadence.yaml`. +Form is heartbeat **plus** reconciliation, not rate monitoring. + +```yaml +form: heartbeat-or-reconciliation +event_class: +evidence_class: load-bearing +rate_monitoring: forbidden +heartbeat: + event_class: .heartbeat # or a dedicated heartbeat class + interval: 24h + assertion: nothing-to-report + missing: finding +reconciliation: + compare_local: source transition counts per class + compare_observed: evidence-engine counts per class + divergence: finding + undrained_local: lag-not-divergence +``` + +A missing heartbeat, or a reconciliation divergence, is a finding about the +stream. None of these are informational logs. + +## 3. Worked example — `qonto-assistant` + +The one real source this draft is written against. + +| Event class | Evidence class | Form | Why | +| --- | --- | --- | --- | +| `audit.allow` | attributive | expected-rate (SHOULD) | Forensic reconstruction; no control branches on presence. Completeness is not claimed. The rate row is here so the volume form is specified, not because qonto traffic is currently a meaningful volume class. | +| `audit.deny` | load-bearing | heartbeat-or-reconciliation (MUST) | Deny-escalation and posture branch on deny presence/absence. Statute §9.6 names denials as load-bearing. Denials are infrequent, so rate monitoring is the wrong form. | +| `audit.heartbeat` | load-bearing | the positive claim | `nothing-to-report` plus per-class counts since the previous heartbeat. | + +Machine-readable worked example consumed by the local evaluator (draft-shaped, +not a competing schema): `src/kings_guard/fixtures/qonto_assistant_cadence.json`. + +Until `qonto-assistant` publishes the declaration and emits heartbeats, no +argument may assume this surface is live on that source. kings-guard can +still observe the audit events it does emit; stream completeness for the +deny class stays `unknown` or `degraded` until the positive claim exists. + +## 4. Consumer obligations (kings-guard) + +- Copy the declared evidence class onto each observation; do not infer it. +- Evaluate the stream against the declaration, not only each record. +- Carry completeness separately from record richness in posture output. +- Treat stream findings as distinct from content findings. +- Open no Tooling client to obtain the stream. The source publishes it. + +## 5. Handover + +Sent to `gate-house`, `net-kingdom`, and `info-tech-canon` for ownership. +kings-guard will consume whatever Taxonomy publishes; it will not fork this +draft into a private schema. diff --git a/specs/ImmuneContracts.md b/specs/ImmuneContracts.md index 9a6da62..7d8f383 100644 --- a/specs/ImmuneContracts.md +++ b/specs/ImmuneContracts.md @@ -1,9 +1,9 @@ --- title: Kings Guard Immune Contracts document_id: KG-CONTRACTS-IMMUNE -version: 0.1.0 +version: 0.2.0 status: Draft -date: 2026-07-23 +date: 2026-09-01 repo: kings-guard classification: Public --- @@ -48,11 +48,36 @@ against statistical surprise. `immune_signal` and `effector_request` may recommend or request response, but they do not silently acquire authority owned by adjacent systems. -### 2.4 Memory is governed +### 2.4 Memory is governed and is not a state plane `immune_memory_entry` records reusable learning without becoming a raw secret store or a place to centralize tenant-confidential payloads. +Under statute §3.4 rule 3, immune memory informs kings-guard's own judgment +and may be published as evidence. No engine, PEP, or workload may read it as +a runtime input. Making it one is a §4 Tooling catalog change, not a quiet +integration. The reference type carries +`runtime_input_for_other_layers: forbidden`. + +### 2.5 Evidence class is declared, not inferred + +Statute §9.6 attaches different obligations to load-bearing and attributive +evidence. The class is the **source's declaration**, recorded on the +`security_genome`'s `evidence_sources` and copied onto each +`immune_observation`. kings-guard does not infer the class from event +contents. + +Obligation difference: + +| Class | Cadence | Completeness | +| --- | --- | --- | +| **Load-bearing** | MUST declare a cadence. Volume classes use expected rate. Low-volume classes (revocations, denials, containment) use heartbeat or reconciliation — rate monitoring is the wrong form. | Absence of a required event, unmet cadence, missing heartbeat, or reconciliation divergence is a finding about the stream. | +| **Attributive** | SHOULD declare a cadence. Atomicity SHOULD be sought; a deliberate trade-away MUST be declared. | Completeness MUST NOT be claimed where atomicity was traded away. | + +The emission-cadence declaration itself is a Taxonomy artifact (§17). +kings-guard drafts it and consumes it; it does not own it. See +`specs/EmissionCadenceDeclaration.md`. + ## 3. Contract Map | Contract | Meaning | Typical producers | Typical consumers | @@ -81,7 +106,10 @@ Minimum fields in v0.1: - permitted capability scope; - permitted protocols and egress destinations; - data classifications; -- declared tolerances. +- declared tolerances; +- declared evidence sources, each with an evidence class (load-bearing or + attributive) and optional cadence form. The class is the source's + declaration. Produced by: @@ -107,6 +135,8 @@ Minimum fields in v0.1: - capability or resource scope; - protocol; - decision/outcome; +- evidence class, copied from the source declaration; +- event class (for example `audit.deny`, `audit.allow`, `audit.heartbeat`); - optional deny reason, identity-binding mode, egress destination, and policy version. @@ -153,9 +183,19 @@ Minimum fields in v0.1: - posture level (`healthy`, `elevated`, `inflamed`, `compromised`); - risk score; -- confidence score; +- confidence score — richness of the **record** received, never of the stream; +- stream completeness (`complete`, `degraded`, `unknown`) and a reason in + words — an unmet cadence or missing heartbeat degrades this dimension, and + a posture derived from an incomplete stream can never read as more + trustworthy than one derived from a complete one; - findings and tolerated findings; -- human-readable rationale. +- human-readable rationale, which must state completeness in words, not only + as a number. + +Stream findings (`stream:cadence_unmet`, `stream:heartbeat_missing`, +`stream:reconciliation_divergence`) are findings about the stream observed, +not about a record's contents. They MUST be distinguishable from content +findings such as `credential_exfil_probe`. Produced by: @@ -202,7 +242,16 @@ Minimum fields in v0.1: - action name; - authority boundary; - reason; -- whether human approval is required. +- whether human approval is required; +- originating observation identifier; +- originating signal identifier; +- stream-completeness state at emission; +- requested restrictive direction (`reduce_authority`, `require_step_up`, + `request_containment`, or `none`). + +These origin fields exist so the eventual decision record can name what the +proposal was rendered for (§9.2). They do not widen authority. A containment +proposal is not an authorization; kings-guard never actuates. Produced by: @@ -244,7 +293,8 @@ Minimum fields in v0.1: - summary; - records it was derived from; - recommended countermeasures; -- confidentiality class. +- confidentiality class; +- `runtime_input_for_other_layers`, which is `forbidden`. Produced by: @@ -253,9 +303,14 @@ Produced by: Consumed by: -- future contract revisions; -- policy authors; -- future cross-run learning surfaces. +- kings-guard's own later judgment; +- published evidence surfaces. + +Not consumed by: + +- any engine, PEP, or workload as a runtime input. That would make immune + memory a state plane, which §3.4 rule 3 forbids unless the memory is + catalogued as Tooling under §4. ## 5. Current Reference Slice diff --git a/src/kings_guard/adapters/qonto_assistant.py b/src/kings_guard/adapters/qonto_assistant.py index 5a93e1f..08ef2be 100644 --- a/src/kings_guard/adapters/qonto_assistant.py +++ b/src/kings_guard/adapters/qonto_assistant.py @@ -3,7 +3,12 @@ from __future__ import annotations from collections.abc import Mapping from typing import Any -from kings_guard.contracts import ImmuneObservation, ObservationDecision +from kings_guard.contracts import ( + EvidenceClass, + ImmuneObservation, + ObservationDecision, + SecurityGenome, +) def observation_from_audit_event( @@ -13,8 +18,24 @@ def observation_from_audit_event( capability_scope: str, identity_binding: str, egress_destination: str | None, + genome: SecurityGenome | None = None, + evidence_class: EvidenceClass | str | None = None, + event_class: str | None = None, ) -> ImmuneObservation: - """Normalize qonto-assistant's audit stream into Kings Guard's observation contract.""" + """Normalize qonto-assistant's audit stream into Kings Guard's observation contract. + + Evidence class is copied from the source's declaration (the genome) or an + explicit caller-supplied declaration. It is never inferred from the event + body — a deny is load-bearing because the source said so, not because + kings-guard recognized the string "deny". + """ + decision = ObservationDecision(str(event["decision"])) + resolved_event_class = event_class or f"audit.{decision.value}" + resolved_class = _declared_evidence_class( + genome=genome, + event_class=resolved_event_class, + evidence_class=evidence_class, + ) return ImmuneObservation( observation_id=str(event["request_id"]), source_system="qonto-assistant", @@ -25,7 +46,9 @@ def observation_from_audit_event( capability=capability_scope, resource_scope=_optional_str(event.get("capability")), protocol=str(event["protocol"]), - decision=ObservationDecision(str(event["decision"])), + decision=decision, + 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, @@ -36,6 +59,36 @@ def observation_from_audit_event( ) +def _declared_evidence_class( + *, + genome: SecurityGenome | None, + event_class: str, + evidence_class: EvidenceClass | str | None, +) -> EvidenceClass: + if evidence_class is not None: + declared = ( + evidence_class + if isinstance(evidence_class, EvidenceClass) + else EvidenceClass(str(evidence_class)) + ) + if genome is not None: + source = genome.source_for(event_class) + if source is not None and source.evidence_class != declared: + raise ValueError( + f"supplied evidence class {declared.value!r} does not match " + f"source declaration {source.evidence_class.value!r} for {event_class}" + ) + return declared + if genome is not None: + source = genome.source_for(event_class) + if source is not None: + return source.evidence_class + raise ValueError( + "evidence class must be declared by the source (genome) or supplied " + "explicitly; kings-guard does not infer it from event contents" + ) + + def _optional_int(value: Any) -> int | None: if value is None: return None diff --git a/src/kings_guard/cadence.py b/src/kings_guard/cadence.py new file mode 100644 index 0000000..35c6dcb --- /dev/null +++ b/src/kings_guard/cadence.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from importlib.resources import files +from typing import Any + +from kings_guard.contracts import CadenceForm, EvidenceClass + + +def parse_timestamp(value: str) -> datetime: + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + parsed = datetime.fromisoformat(text) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=UTC) + return parsed.astimezone(UTC) + + +def parse_interval(value: str | int) -> timedelta: + if isinstance(value, int): + return timedelta(seconds=value) + text = str(value).strip().lower() + if text.endswith("s") and text[:-1].isdigit(): + return timedelta(seconds=int(text[:-1])) + if text.endswith("m") and text[:-1].isdigit(): + return timedelta(minutes=int(text[:-1])) + if text.endswith("h") and text[:-1].isdigit(): + 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])) + raise ValueError(f"unsupported interval: {value!r}") + + +@dataclass(frozen=True, slots=True) +class RateCadence: + event_class: str + evidence_class: EvidenceClass + window: timedelta + expected_min: int + + +@dataclass(frozen=True, slots=True) +class HeartbeatCadence: + event_class: str + covered_event_class: str + evidence_class: EvidenceClass + interval: timedelta + assertion: str + + +@dataclass(frozen=True, slots=True) +class ReconciliationCadence: + covered_event_class: str + evidence_class: EvidenceClass + local_field: str + observed_field: str + + +@dataclass(frozen=True, slots=True) +class EmissionCadence: + """Runtime view of the Taxonomy draft, loaded from the local worked example. + + This is a consumer of the draft in `specs/EmissionCadenceDeclaration.md`, + not a competing schema. Ownership stays with Taxonomy. + """ + + schema_version: str + status: str + drafter: str + owner: str + source_system: str + reference_instance: str + rates: tuple[RateCadence, ...] + heartbeats: tuple[HeartbeatCadence, ...] + reconciliations: tuple[ReconciliationCadence, ...] + + def forms(self) -> frozenset[CadenceForm]: + forms: set[CadenceForm] = set() + if self.rates: + forms.add("expected-rate") + if self.heartbeats or self.reconciliations: + forms.add("heartbeat-or-reconciliation") + return frozenset(forms) + + +def load_qonto_assistant_cadence() -> EmissionCadence: + payload = json.loads( + files("kings_guard") + .joinpath("fixtures") + .joinpath("qonto_assistant_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"])) + form = str(item["form"]) + if form == "expected-rate": + rates.append( + RateCadence( + event_class=str(item["event_class"]), + evidence_class=evidence_class, + window=parse_interval(item.get("window_seconds", item.get("window"))), + expected_min=int(item["expected_min"]), + ) + ) + continue + if form != "heartbeat-or-reconciliation": + raise ValueError(f"unknown cadence form: {form}") + heartbeat = item.get("heartbeat") or {} + if heartbeat: + heartbeats.append( + HeartbeatCadence( + event_class=str(heartbeat["event_class"]), + covered_event_class=str(item["event_class"]), + evidence_class=evidence_class, + interval=parse_interval( + heartbeat.get("interval_seconds", heartbeat.get("interval")) + ), + assertion=str(heartbeat.get("assertion", "nothing-to-report")), + ) + ) + reconciliation = item.get("reconciliation") or {} + if reconciliation: + reconciliations.append( + ReconciliationCadence( + covered_event_class=str(item["event_class"]), + evidence_class=evidence_class, + local_field=str(reconciliation.get("compare_local", "source_counts")), + observed_field=str(reconciliation.get("compare_observed", "evidence_counts")), + ) + ) + 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")), + source_system=str(data.get("source", data.get("source_system", "unknown"))), + reference_instance=str(data.get("reference_instance", "GH-WP-0002-T04")), + rates=tuple(rates), + heartbeats=tuple(heartbeats), + reconciliations=tuple(reconciliations), + ) + + +def count_event_classes(event_classes: Sequence[str]) -> dict[str, int]: + counts: dict[str, int] = {} + for event_class in event_classes: + counts[event_class] = counts.get(event_class, 0) + 1 + return counts diff --git a/src/kings_guard/contracts.py b/src/kings_guard/contracts.py index 7047da0..b4edca5 100644 --- a/src/kings_guard/contracts.py +++ b/src/kings_guard/contracts.py @@ -12,12 +12,36 @@ AuthorityBoundary = Literal[ "local_service_owned", "requires_human_approval", ] +RestrictiveDirection = Literal[ + "reduce_authority", + "require_step_up", + "request_containment", + "none", +] +CadenceForm = Literal["expected-rate", "heartbeat-or-reconciliation"] +MemoryRuntimeDependency = Literal["forbidden"] + + +class EvidenceClass(str, Enum): + """Source-declared evidence class (§9.6). Not inferred by kings-guard.""" + + LOAD_BEARING = "load-bearing" + ATTRIBUTIVE = "attributive" + + +class StreamCompleteness(str, Enum): + """Completeness of the *stream*, never richness of the record in hand.""" + + COMPLETE = "complete" + DEGRADED = "degraded" + UNKNOWN = "unknown" class ObservationDecision(str, Enum): ALLOW = "allow" DENY = "deny" ERROR = "error" + HEARTBEAT = "heartbeat" class PostureLevel(str, Enum): @@ -31,6 +55,16 @@ class SignalKind(str, Enum): POSTURE_HINT = "posture_hint" OBSERVATION_ALERT = "observation_alert" RECOVERY_REQUEST = "recovery_request" + STREAM_COMPLETENESS = "stream_completeness" + + +STREAM_FINDING_PREFIX = "stream:" + +COMPLETENESS_RANK = { + StreamCompleteness.UNKNOWN: 0, + StreamCompleteness.DEGRADED: 1, + StreamCompleteness.COMPLETE: 2, +} @dataclass(frozen=True, slots=True) @@ -52,6 +86,34 @@ class ToleranceRule: ) +@dataclass(frozen=True, slots=True) +class DeclaredEvidenceSource: + """A source the subject declares, including its evidence class. + + The class is the source's declaration. kings-guard copies it onto + observations; it does not infer it from event contents. + """ + + source_id: str + source_system: str + event_class: str + evidence_class: EvidenceClass + reasoning: str + cadence_form: CadenceForm | None = None + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "DeclaredEvidenceSource": + cadence_form = data.get("cadence_form") + return cls( + source_id=str(data["source_id"]), + source_system=str(data["source_system"]), + event_class=str(data["event_class"]), + evidence_class=EvidenceClass(str(data["evidence_class"])), + reasoning=str(data["reasoning"]), + cadence_form=str(cadence_form) if cadence_form else None, + ) + + @dataclass(frozen=True, slots=True) class SecurityGenome: genome_id: str @@ -64,6 +126,7 @@ class SecurityGenome: permitted_egress: frozenset[str] data_classifications: tuple[str, ...] = () tolerances: tuple[ToleranceRule, ...] = () + evidence_sources: tuple[DeclaredEvidenceSource, ...] = () @classmethod def from_dict(cls, data: Mapping[str, Any]) -> "SecurityGenome": @@ -80,8 +143,18 @@ class SecurityGenome: 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", ()) + ), ) + def source_for(self, event_class: str) -> DeclaredEvidenceSource | None: + for source in self.evidence_sources: + if source.event_class == event_class: + return source + return None + @dataclass(frozen=True, slots=True) class ImmuneObservation: @@ -95,6 +168,8 @@ class ImmuneObservation: resource_scope: str | None protocol: str decision: ObservationDecision + evidence_class: EvidenceClass + event_class: str deny_reason: str | None = None identity_binding: str | None = None egress_destination: str | None = None @@ -116,6 +191,8 @@ class ImmuneObservation: resource_scope=_optional_str(data.get("resource_scope")), protocol=str(data["protocol"]), decision=ObservationDecision(str(data["decision"])), + evidence_class=EvidenceClass(str(data["evidence_class"])), + event_class=str(data["event_class"]), deny_reason=_optional_str(data.get("deny_reason")), identity_binding=_optional_str(data.get("identity_binding")), egress_destination=_optional_str(data.get("egress_destination")), @@ -126,6 +203,25 @@ class ImmuneObservation: ) +@dataclass(frozen=True, slots=True) +class StreamHeartbeat: + """A signed positive claim that can itself go missing (§9.6).""" + + source_system: str + timestamp: str + event_class: str + assertion: str + counts: Mapping[str, int] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class ReconciliationView: + """Source transition counts versus observed evidence counts per event class.""" + + source_counts: Mapping[str, int] + evidence_counts: Mapping[str, int] + + @dataclass(frozen=True, slots=True) class SecurityPhenotype: subject_id: str @@ -142,11 +238,22 @@ class PostureAssessment: posture: PostureLevel risk_score: int confidence_score: int + stream_completeness: StreamCompleteness + completeness_reason: str findings: tuple[str, ...] tolerated_findings: tuple[str, ...] rationale: str +@dataclass(frozen=True, slots=True) +class StreamAssessment: + completeness: StreamCompleteness + reason: str + findings: tuple[str, ...] + observed_counts: Mapping[str, int] + window_end: str + + @dataclass(frozen=True, slots=True) class EffectorRequest: target_system: str @@ -154,6 +261,10 @@ class EffectorRequest: authority_boundary: AuthorityBoundary reason: str requires_human_approval: bool + originating_observation_id: str + originating_signal_id: str + stream_completeness: StreamCompleteness + restrictive_direction: RestrictiveDirection = "none" @dataclass(frozen=True, slots=True) @@ -170,12 +281,15 @@ class ImmuneSignal: @dataclass(frozen=True, slots=True) class ImmuneMemoryEntry: + """Governed defensive learning. Not a runtime input for any other layer.""" + memory_id: str subject_scope: str summary: str derived_from: tuple[str, ...] recommended_countermeasures: tuple[str, ...] confidentiality: str = "non-secret" + runtime_input_for_other_layers: MemoryRuntimeDependency = "forbidden" @dataclass(frozen=True, slots=True) @@ -183,13 +297,23 @@ class PostureEvaluation: phenotype: SecurityPhenotype assessment: PostureAssessment signals: tuple[ImmuneSignal, ...] + stream: StreamAssessment | None = None + + +def assessment_trust_key(assessment: PostureAssessment) -> tuple[int, int]: + """Order judgments so completeness outranks record richness. + + An incomplete stream can never read as more trustworthy than a complete + one, regardless of how well-formed the record in hand is. + """ + return (COMPLETENESS_RANK[assessment.stream_completeness], assessment.confidence_score) def as_jsonable(value: Any) -> Any: """Convert contract objects into JSON-safe primitives.""" if isinstance(value, Enum): return value.value - if is_dataclass(value): + if is_dataclass(value) and not isinstance(value, type): return {key: as_jsonable(item) for key, item in asdict(value).items()} if isinstance(value, Mapping): return {str(key): as_jsonable(item) for key, item in value.items()} diff --git a/src/kings_guard/fixtures.py b/src/kings_guard/fixtures.py index 60b17f4..7c12d96 100644 --- a/src/kings_guard/fixtures.py +++ b/src/kings_guard/fixtures.py @@ -5,6 +5,7 @@ from dataclasses import dataclass from importlib.resources import files from typing import Any +from kings_guard.cadence import EmissionCadence, load_qonto_assistant_cadence from kings_guard.contracts import SecurityGenome @@ -14,6 +15,7 @@ class QontoAssistantPilotFixture: audit_event: dict[str, Any] normalization_hints: dict[str, str] source_notes: tuple[str, ...] + evidence_class_reasoning: tuple[str, ...] def load_qonto_assistant_pilot() -> QontoAssistantPilotFixture: @@ -21,11 +23,20 @@ def load_qonto_assistant_pilot() -> QontoAssistantPilotFixture: return QontoAssistantPilotFixture( genome=SecurityGenome.from_dict(payload["normalized_genome"]), audit_event=dict(payload["qonto_audit_event"]), - normalization_hints={str(key): str(value) for key, value in payload["normalization_hints"].items()}, + normalization_hints={ + str(key): str(value) for key, value in payload["normalization_hints"].items() + }, source_notes=tuple(str(item) for item in payload.get("source_notes", ())), + evidence_class_reasoning=tuple( + str(item) for item in payload.get("evidence_class_reasoning", ()) + ), ) +def load_pilot_cadence() -> EmissionCadence: + return load_qonto_assistant_cadence() + + def _load_json_fixture(name: str) -> dict[str, Any]: fixture_path = files("kings_guard").joinpath("fixtures").joinpath(name) return json.loads(fixture_path.read_text(encoding="utf-8")) diff --git a/src/kings_guard/fixtures/qonto_assistant_cadence.json b/src/kings_guard/fixtures/qonto_assistant_cadence.json new file mode 100644 index 0000000..6ba4a83 --- /dev/null +++ b/src/kings_guard/fixtures/qonto_assistant_cadence.json @@ -0,0 +1,44 @@ +{ + "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." + }, + { + "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" + }, + "reconciliation": { + "compare_local": "source_transition_counts.audit.deny", + "compare_observed": "evidence_counts.audit.deny", + "divergence": "finding" + } + } + ] +} diff --git a/src/kings_guard/fixtures/qonto_assistant_pilot.json b/src/kings_guard/fixtures/qonto_assistant_pilot.json index 4e2889d..e0c0f4d 100644 --- a/src/kings_guard/fixtures/qonto_assistant_pilot.json +++ b/src/kings_guard/fixtures/qonto_assistant_pilot.json @@ -28,6 +28,32 @@ "description": "Actor identity is still self-asserted until key-cape integration lands.", "effect": "monitor" } + ], + "evidence_sources": [ + { + "source_id": "qonto-assistant.audit.allow", + "source_system": "qonto-assistant", + "event_class": "audit.allow", + "evidence_class": "attributive", + "cadence_form": "expected-rate", + "reasoning": "Allow records support forensic reconstruction. No control currently branches on an allow being present or absent, so the class is attributive and completeness is not claimed." + }, + { + "source_id": "qonto-assistant.audit.deny", + "source_system": "qonto-assistant", + "event_class": "audit.deny", + "evidence_class": "load-bearing", + "cadence_form": "heartbeat-or-reconciliation", + "reasoning": "qonto-assistant's deny-escalation loop and kings-guard posture both branch on deny presence or absence. Statute §9.6 names denials as load-bearing. The class is this source's declaration, not an inference from the string 'deny'." + }, + { + "source_id": "qonto-assistant.audit.heartbeat", + "source_system": "qonto-assistant", + "event_class": "audit.heartbeat", + "evidence_class": "load-bearing", + "cadence_form": "heartbeat-or-reconciliation", + "reasoning": "The positive nothing-to-report claim for the deny class. Rate monitoring cannot work on infrequent denials; the heartbeat is the claim that can itself go missing." + } ] }, "qonto_audit_event": { @@ -50,9 +76,15 @@ "identity_binding": "self_asserted", "egress_destination": "qonto-thirdparty-api" }, + "evidence_class_reasoning": [ + "audit.deny is load-bearing because deny-escalation and posture branch on it; the source declares that class.", + "audit.allow is attributive: forensic reconstruction only; completeness is not claimed.", + "kings-guard copies the declared class onto the observation and does not infer it from decision=deny." + ], "source_notes": [ "Derived from qonto-assistant/specs/security-genome.yaml", "Derived from qonto-assistant/src/qonto_assistant/contracts.py#AuditEvent", - "Pilot chooses qonto-assistant because it already ships an audit stream, a genome record, and a fast local loop." + "Pilot chooses qonto-assistant because it already ships an audit stream, a genome record, and a fast local loop.", + "This JSON remains the regression fixture. Live emitted events are captured separately by kings_guard.live." ] } diff --git a/src/kings_guard/live.py b/src/kings_guard/live.py new file mode 100644 index 0000000..a50028d --- /dev/null +++ b/src/kings_guard/live.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +import json +import sys +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from kings_guard.adapters import observation_from_audit_event +from kings_guard.contracts import ImmuneObservation, SecurityGenome + +QONTO_ASSISTANT_SRC = Path("/home/worsch/qonto-assistant/src") +QONTO_POLICY = ( + Path("/home/worsch/qonto-assistant/src/qonto_assistant/policy/qonto-v1.yaml") +) +QONTO_FIXTURES = Path("/home/worsch/qonto-assistant/tests/fixtures/qonto") + + +@dataclass(frozen=True, slots=True) +class LiveQontoCapture: + events: tuple[dict[str, Any], ...] + observations: tuple[ImmuneObservation, ...] + mapping_notes: tuple[str, ...] + corrections_for_source: tuple[str, ...] + + +def qonto_assistant_available() -> bool: + return (QONTO_ASSISTANT_SRC / "qonto_assistant" / "service.py").is_file() + + +def capture_qonto_assistant_events(genome: SecurityGenome) -> LiveQontoCapture: + """Drive qonto-assistant's real emit path and capture what it publishes. + + Uses the adjacent checkout and its fixture-backed client. No Tooling + 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}" + ) + + src = str(QONTO_ASSISTANT_SRC) + if src not in sys.path: + sys.path.insert(0, src) + + from qonto_assistant.audit import AuditLogger + from qonto_assistant.contracts import ActorClaims + from qonto_assistant.errors import PolicyDeniedError + from qonto_assistant.policy import PolicyEngine + from qonto_assistant.qonto_client import FixtureQontoClient + from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter + from qonto_assistant.service import CapabilityService + + events: list[dict[str, Any]] = [] + service = CapabilityService( + client=FixtureQontoClient(fixture_dir=QONTO_FIXTURES), + policy=PolicyEngine.from_file( + QONTO_POLICY, + required_scope="finance.qonto.read", + enforce_scope=False, + ), + audit_logger=AuditLogger(sink=events.append), + rate_limiter=RateLimiter(limit=100, window_seconds=60), + concurrency_limiter=ConcurrencyLimiter(limit=4), + ) + claims = ActorClaims(actor_id="kg-live-observer", tenant_id="binky", lane="green") + + allow_payload = service.get_accounts( + claims=claims, request_id="req-kg-live-allow", protocol="rest" + ) + if not isinstance(allow_payload, Mapping): + raise RuntimeError("qonto-assistant allow path did not return a payload") + + try: + service.list_transactions( + claims=claims, + request_id="req-kg-live-deny-arg-constraint", + account_slug=None, + page=1, + page_size=10_000, + window_days=31, + status="completed", + side=None, + protocol="mcp", + ) + except PolicyDeniedError: + pass + 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)}") + + observations = tuple( + observation_from_audit_event( + event, + subject_id=genome.subject_id, + capability_scope="finance.qonto.read", + identity_binding="self_asserted", + egress_destination="qonto-thirdparty-api", + genome=genome, + ) + for event in events + ) + mapping_notes, corrections = _confirm_mapping(events, observations) + return LiveQontoCapture( + events=tuple(events), + observations=observations, + mapping_notes=mapping_notes, + corrections_for_source=corrections, + ) + + +def observations_from_jsonl( + path: Path, + genome: SecurityGenome, + *, + capability_scope: str, + identity_binding: str, + egress_destination: str | None, +) -> tuple[ImmuneObservation, ...]: + events = _read_jsonl(path) + return tuple( + observation_from_audit_event( + event, + subject_id=genome.subject_id, + capability_scope=capability_scope, + identity_binding=identity_binding, + egress_destination=egress_destination, + genome=genome, + ) + for event in events + ) + + +def _confirm_mapping( + events: Sequence[Mapping[str, Any]], + observations: Sequence[ImmuneObservation], +) -> tuple[tuple[str, ...], tuple[str, ...]]: + notes: list[str] = [] + corrections: list[str] = [] + required = ( + "request_id", + "timestamp", + "actor", + "tenant_id", + "capability", + "protocol", + "decision", + "policy_version", + "latency_ms", + ) + for event, observation in zip(events, observations, strict=True): + missing = [field for field in required if field not in event] + if missing: + corrections.append(f"audit event missing fields: {missing}") + continue + if observation.observation_id != str(event["request_id"]): + corrections.append("request_id mapping drifted") + if observation.resource_scope != str(event["capability"]): + corrections.append("capability->resource_scope mapping drifted") + if observation.capability != "finance.qonto.read": + corrections.append("coarse capability_scope mapping drifted") + if observation.actor_id != str(event["actor"]): + corrections.append("actor mapping drifted") + if observation.decision.value != str(event["decision"]): + corrections.append("decision mapping drifted") + notes.append( + f"{event['request_id']}: decision={event['decision']} " + f"capability={event['capability']} protocol={event['protocol']} " + f"maps to observation {observation.observation_id} " + f"event_class={observation.event_class} " + f"evidence_class={observation.evidence_class.value}" + ) + if "identity_binding" not in event: + corrections.append( + "AuditEvent still omits identity_binding; kings-guard continues " + "to use the genome-declared constant self_asserted" + ) + if "egress_destination" not in event: + corrections.append( + "AuditEvent still omits egress_destination; kings-guard continues " + "to use the genome-declared constant qonto-thirdparty-api" + ) + # Deduplicate repeated correction text across events. + unique_corrections = tuple(dict.fromkeys(corrections)) + return tuple(notes), unique_corrections + + +def _read_jsonl(path: Path) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + for line in path.read_text(encoding="utf-8").splitlines(): + text = line.strip() + if not text: + continue + events.append(json.loads(text)) + return events diff --git a/src/kings_guard/main.py b/src/kings_guard/main.py index 4144ab9..fe8c5b4 100644 --- a/src/kings_guard/main.py +++ b/src/kings_guard/main.py @@ -5,7 +5,8 @@ import json from kings_guard.adapters import observation_from_audit_event from kings_guard.contracts import as_jsonable -from kings_guard.fixtures import load_qonto_assistant_pilot +from kings_guard.fixtures import load_pilot_cadence, load_qonto_assistant_pilot +from kings_guard.live import capture_qonto_assistant_events, qonto_assistant_available from kings_guard.posture import PostureEvaluator @@ -17,32 +18,61 @@ def main() -> None: choices=["qonto-assistant"], help="Pilot bundle to evaluate.", ) + parser.add_argument( + "--live", + action="store_true", + help="Observe real events emitted by qonto-assistant's AuditLogger (adjacent checkout).", + ) args = parser.parse_args() if args.pilot != "qonto-assistant": raise SystemExit(f"Unsupported pilot: {args.pilot}") fixture = load_qonto_assistant_pilot() - observation = observation_from_audit_event( - fixture.audit_event, - subject_id=fixture.normalization_hints["subject_id"], - capability_scope=fixture.normalization_hints["capability_scope"], - identity_binding=fixture.normalization_hints["identity_binding"], - egress_destination=fixture.normalization_hints["egress_destination"], - ) - evaluation = PostureEvaluator().evaluate(fixture.genome, observation) - print( - json.dumps( - { - "pilot": args.pilot, - "source_notes": list(fixture.source_notes), - "observation": as_jsonable(observation), - "evaluation": as_jsonable(evaluation), - }, - indent=2, - sort_keys=True, + cadence = load_pilot_cadence() + evaluator = PostureEvaluator() + + if args.live: + if not qonto_assistant_available(): + raise SystemExit("qonto-assistant checkout is not available for live observation") + capture = capture_qonto_assistant_events(fixture.genome) + observation = capture.observations[-1] + now = observation.timestamp + evaluation = evaluator.evaluate_with_stream( + fixture.genome, + observation, + cadence, + now=now, + observations=capture.observations, ) - ) + payload = { + "pilot": args.pilot, + "mode": "live", + "source_notes": list(fixture.source_notes), + "mapping_notes": list(capture.mapping_notes), + "corrections_for_source": list(capture.corrections_for_source), + "observation": as_jsonable(observation), + "evaluation": as_jsonable(evaluation), + } + else: + observation = observation_from_audit_event( + fixture.audit_event, + subject_id=fixture.normalization_hints["subject_id"], + capability_scope=fixture.normalization_hints["capability_scope"], + identity_binding=fixture.normalization_hints["identity_binding"], + egress_destination=fixture.normalization_hints["egress_destination"], + genome=fixture.genome, + ) + evaluation = evaluator.evaluate(fixture.genome, observation) + payload = { + "pilot": args.pilot, + "mode": "fixture", + "source_notes": list(fixture.source_notes), + "observation": as_jsonable(observation), + "evaluation": as_jsonable(evaluation), + } + + print(json.dumps(payload, indent=2, sort_keys=True)) if __name__ == "__main__": diff --git a/src/kings_guard/posture.py b/src/kings_guard/posture.py index e68d8fb..c07c643 100644 --- a/src/kings_guard/posture.py +++ b/src/kings_guard/posture.py @@ -1,16 +1,26 @@ from __future__ import annotations +from collections.abc import Sequence + +from kings_guard.cadence import EmissionCadence from kings_guard.contracts import ( + STREAM_FINDING_PREFIX, EffectorRequest, ImmuneObservation, ImmuneSignal, PostureAssessment, PostureEvaluation, PostureLevel, + ReconciliationView, + RestrictiveDirection, SecurityGenome, SecurityPhenotype, SignalKind, + StreamAssessment, + StreamCompleteness, + StreamHeartbeat, ) +from kings_guard.stream import evaluate_stream as assess_stream CRITICAL_FINDINGS = frozenset( { @@ -39,13 +49,70 @@ RISK_WEIGHTS = { "control_plane_error": 20, } +UNKNOWN_STREAM_REASON = ( + "per-observation evaluation cannot vouch for stream completeness; " + "the record in hand was scored for richness only" +) + class PostureEvaluator: - def evaluate(self, genome: SecurityGenome, observation: ImmuneObservation) -> PostureEvaluation: + def evaluate( + self, + genome: SecurityGenome, + observation: ImmuneObservation, + *, + stream: StreamAssessment | None = None, + ) -> PostureEvaluation: phenotype = self._derive_phenotype(genome, observation) - assessment = self._assess(phenotype, observation) - signals = self._build_signals(observation, assessment) - return PostureEvaluation(phenotype=phenotype, assessment=assessment, signals=signals) + assessment = self._assess(phenotype, observation, stream=stream) + signals = self._build_signals(observation, assessment, stream=stream) + return PostureEvaluation( + phenotype=phenotype, + assessment=assessment, + signals=signals, + stream=stream, + ) + + def evaluate_stream( + self, + observations: Sequence[ImmuneObservation], + cadence: EmissionCadence, + *, + now: str, + watching_since: str | None = None, + heartbeats: Sequence[StreamHeartbeat] = (), + reconciliation: ReconciliationView | None = None, + ) -> StreamAssessment: + return assess_stream( + observations, + cadence, + now=now, + watching_since=watching_since, + heartbeats=heartbeats, + reconciliation=reconciliation, + ) + + def evaluate_with_stream( + self, + genome: SecurityGenome, + observation: ImmuneObservation, + cadence: EmissionCadence, + *, + now: str, + observations: Sequence[ImmuneObservation] | None = None, + watching_since: str | None = None, + heartbeats: Sequence[StreamHeartbeat] = (), + reconciliation: ReconciliationView | None = None, + ) -> PostureEvaluation: + stream = self.evaluate_stream( + observations if observations is not None else (observation,), + cadence, + now=now, + watching_since=watching_since, + heartbeats=heartbeats, + reconciliation=reconciliation, + ) + return self.evaluate(genome, observation, stream=stream) def _derive_phenotype( self, @@ -55,6 +122,17 @@ class PostureEvaluator: findings: list[str] = [] tolerated: list[str] = [] + if observation.decision is observation.decision.HEARTBEAT: + return SecurityPhenotype( + subject_id=observation.subject_id, + tenant_id=observation.tenant_id, + observed_capability=observation.capability, + protocol=observation.protocol, + decision=observation.decision, + active_findings=(), + tolerated_findings=(), + ) + if observation.tenant_id != genome.tenant_id: findings.append("tenant_mismatch") if observation.capability not in genome.permitted_capabilities: @@ -95,6 +173,8 @@ class PostureEvaluator: self, phenotype: SecurityPhenotype, observation: ImmuneObservation, + *, + stream: StreamAssessment | None, ) -> PostureAssessment: findings = set(phenotype.active_findings) @@ -115,6 +195,7 @@ class PostureEvaluator: risk_score = max(risk_score, 95) risk_score = min(risk_score, 100) + # Richness of the record received — never of the stream. confidence_score = 70 if observation.policy_version is not None: confidence_score += 10 @@ -124,15 +205,26 @@ class PostureEvaluator: confidence_score += 5 confidence_score = min(confidence_score, 95) + if stream is None: + completeness = StreamCompleteness.UNKNOWN + completeness_reason = UNKNOWN_STREAM_REASON + else: + completeness = stream.completeness + completeness_reason = stream.reason + rationale = _build_rationale( posture=posture, findings=phenotype.active_findings, tolerated_findings=phenotype.tolerated_findings, + completeness=completeness, + completeness_reason=completeness_reason, ) return PostureAssessment( posture=posture, risk_score=risk_score, confidence_score=confidence_score, + stream_completeness=completeness, + completeness_reason=completeness_reason, findings=phenotype.active_findings, tolerated_findings=phenotype.tolerated_findings, rationale=rationale, @@ -142,23 +234,47 @@ class PostureEvaluator: self, observation: ImmuneObservation, assessment: PostureAssessment, + *, + stream: StreamAssessment | None, ) -> tuple[ImmuneSignal, ...]: - if assessment.posture is PostureLevel.HEALTHY: - return () + signals: list[ImmuneSignal] = [] + if assessment.posture is not PostureLevel.HEALTHY: + if observation.source_system == "qonto-assistant": + signals.append(self._build_qonto_pilot_signal(observation, assessment)) + else: + signals.append( + ImmuneSignal( + signal_id=f"sig:{observation.observation_id}", + signal_kind=SignalKind.OBSERVATION_ALERT, + posture=assessment.posture, + summary=assessment.rationale, + target_system=observation.source_system, + findings=assessment.findings, + metadata={"source_system": observation.source_system}, + ) + ) + if stream is not None and stream.findings: + signals.append(self._build_stream_signal(observation, assessment, stream)) + return tuple(signals) - if observation.source_system == "qonto-assistant": - return (self._build_qonto_pilot_signal(observation, assessment),) - - signal = ImmuneSignal( - signal_id=f"sig:{observation.observation_id}", - signal_kind=SignalKind.OBSERVATION_ALERT, + def _build_stream_signal( + self, + observation: ImmuneObservation, + assessment: PostureAssessment, + stream: StreamAssessment, + ) -> ImmuneSignal: + return ImmuneSignal( + signal_id=f"sig:stream:{observation.observation_id}", + signal_kind=SignalKind.STREAM_COMPLETENESS, posture=assessment.posture, - summary=assessment.rationale, + summary=stream.reason, target_system=observation.source_system, - findings=assessment.findings, - metadata={"source_system": observation.source_system}, + findings=stream.findings, + metadata={ + "stream_completeness": stream.completeness.value, + "finding_class": "stream", + }, ) - return (signal,) def _build_qonto_pilot_signal( self, @@ -167,19 +283,22 @@ class PostureEvaluator: ) -> ImmuneSignal: if "credential_exfil_probe" in assessment.findings: action = "lock_actor_temporarily" + direction: RestrictiveDirection = "reduce_authority" reason = ( "Observed a credential-exfil deny signal; qonto-assistant should activate " "its fast local loop lockout and preserve metadata-only evidence." ) else: action = "tighten_actor_scrutiny" + direction = "require_step_up" reason = ( "Observed repeated policy-boundary pressure; qonto-assistant should tighten " "local scrutiny without delegating final authorization to kings-guard." ) + signal_id = f"sig:{observation.observation_id}" return ImmuneSignal( - signal_id=f"sig:{observation.observation_id}", + signal_id=signal_id, signal_kind=SignalKind.POSTURE_HINT, posture=assessment.posture, summary=reason, @@ -192,6 +311,10 @@ class PostureEvaluator: authority_boundary="advisory_only", reason=reason, requires_human_approval=False, + originating_observation_id=observation.observation_id, + originating_signal_id=signal_id, + stream_completeness=assessment.stream_completeness, + restrictive_direction=direction, ), EffectorRequest( target_system="state-hub", @@ -199,11 +322,16 @@ class PostureEvaluator: authority_boundary="metadata_only", reason="Preserve posture evidence without copying secret values.", requires_human_approval=False, + originating_observation_id=observation.observation_id, + originating_signal_id=signal_id, + stream_completeness=assessment.stream_completeness, + restrictive_direction="none", ), ), metadata={ "pilot_lane": "qonto-assistant", "resource_scope": observation.resource_scope or "unknown", + "originating_observation_id": observation.observation_id, }, ) @@ -224,19 +352,40 @@ def _build_rationale( posture: PostureLevel, findings: tuple[str, ...], tolerated_findings: tuple[str, ...], + completeness: StreamCompleteness, + completeness_reason: str, ) -> str: if posture is PostureLevel.HEALTHY: if tolerated_findings: - return ( + content = ( "Healthy posture with tolerated deviations only: " + ", ".join(tolerated_findings) ) - return "Healthy posture: observation is compatible with declared intent." + else: + content = "Healthy posture: observation is compatible with declared intent." + else: + detail = ", ".join(findings) if findings else "no active findings" + tolerated = ( + f" Tolerated deviations still present: {', '.join(tolerated_findings)}." + if tolerated_findings + else "" + ) + content = f"{posture.value.title()} posture driven by {detail}.{tolerated}" - detail = ", ".join(findings) if findings else "no active findings" - tolerated = ( - f" Tolerated deviations still present: {', '.join(tolerated_findings)}." - if tolerated_findings - else "" - ) - return f"{posture.value.title()} posture driven by {detail}.{tolerated}" + 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})." + ) + else: + stream_text = ( + "Stream completeness is unknown; confidence scores the record, " + f"not the stream ({completeness_reason})." + ) + return f"{content} {stream_text}" + + +def is_stream_finding(finding: str) -> bool: + return finding.startswith(STREAM_FINDING_PREFIX) diff --git a/src/kings_guard/stream.py b/src/kings_guard/stream.py new file mode 100644 index 0000000..308d913 --- /dev/null +++ b/src/kings_guard/stream.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +from collections.abc import Sequence +from datetime import UTC, datetime + +from kings_guard.cadence import EmissionCadence, parse_timestamp +from kings_guard.contracts import ( + STREAM_FINDING_PREFIX, + ImmuneObservation, + ReconciliationView, + StreamAssessment, + StreamCompleteness, + StreamHeartbeat, +) + + +def evaluate_stream( + observations: Sequence[ImmuneObservation], + cadence: EmissionCadence, + *, + now: str | datetime, + watching_since: str | datetime | None = None, + heartbeats: Sequence[StreamHeartbeat] = (), + reconciliation: ReconciliationView | None = None, +) -> StreamAssessment: + """Evaluate the stream, not its contents. + + Findings are prefixed with `stream:` so they cannot be mistaken for + content findings such as `credential_exfil_probe`. + """ + now_dt = _as_datetime(now) + findings: list[str] = [] + reasons: list[str] = [] + observed_counts = _count_observations(observations) + + watch_start = _as_datetime(watching_since) if watching_since is not None else None + if watch_start is None: + timestamps = [parse_timestamp(item.timestamp) for item in observations] + timestamps.extend(parse_timestamp(item.timestamp) for item in heartbeats) + watch_start = min(timestamps) if timestamps else now_dt + + for rate in cadence.rates: + window_start = now_dt - rate.window + count = sum( + 1 + for item in observations + if item.event_class == rate.event_class + and parse_timestamp(item.timestamp) >= window_start + ) + observed_counts[rate.event_class] = count + watched_long_enough = (now_dt - watch_start) >= rate.window + if watched_long_enough and count < rate.expected_min: + findings.append(f"{STREAM_FINDING_PREFIX}cadence_unmet:{rate.event_class}") + reasons.append( + f"declared rate for {rate.event_class} is at least {rate.expected_min} " + f"per {int(rate.window.total_seconds())}s; observed {count}" + ) + + for spec in cadence.heartbeats: + due = (now_dt - watch_start) >= spec.interval + latest = _latest_heartbeat(heartbeats, spec.event_class) + missing = False + if latest is None: + missing = due + else: + missing = (now_dt - parse_timestamp(latest.timestamp)) > spec.interval + if missing: + findings.append(f"{STREAM_FINDING_PREFIX}heartbeat_missing:{spec.covered_event_class}") + reasons.append( + f"no {spec.assertion} heartbeat for {spec.covered_event_class} " + f"within {int(spec.interval.total_seconds())}s" + ) + elif not due and latest is None: + reasons.append( + f"heartbeat for {spec.covered_event_class} is not yet due; " + "the stream cannot be vouched for until a positive claim arrives" + ) + + if cadence.reconciliations: + if reconciliation is None: + 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: + findings.append( + f"{STREAM_FINDING_PREFIX}reconciliation_divergence:{spec.covered_event_class}" + ) + reasons.append( + f"{spec.covered_event_class} source transitions={source_count} " + f"but evidence count={evidence_count}" + ) + + completeness = _completeness(findings, reasons, cadence) + reason = ( + "; ".join(reasons) + if reasons + else "declared cadence is met and no stream finding is open" + ) + return StreamAssessment( + completeness=completeness, + reason=reason, + findings=tuple(findings), + observed_counts=observed_counts, + window_end=now_dt.isoformat().replace("+00:00", "Z"), + ) + + +def _completeness( + findings: Sequence[str], + reasons: Sequence[str], + cadence: EmissionCadence, +) -> 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: + return StreamCompleteness.UNKNOWN + if not cadence.heartbeats and not cadence.rates and not cadence.reconciliations: + return StreamCompleteness.UNKNOWN + return StreamCompleteness.COMPLETE + + +def _count_observations(observations: Sequence[ImmuneObservation]) -> dict[str, int]: + counts: dict[str, int] = {} + for item in observations: + counts[item.event_class] = counts.get(item.event_class, 0) + 1 + return counts + + +def _latest_heartbeat( + heartbeats: Sequence[StreamHeartbeat], + event_class: str, +) -> StreamHeartbeat | None: + matching = [item for item in heartbeats if item.event_class == event_class] + if not matching: + return None + return max(matching, key=lambda item: parse_timestamp(item.timestamp)) + + +def _as_datetime(value: str | datetime) -> datetime: + if isinstance(value, datetime): + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + return parse_timestamp(value) diff --git a/tests/conftest.py b/tests/conftest.py index de705b1..8bdfea5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,6 +3,9 @@ from __future__ import annotations import sys from pathlib import Path -SRC = Path(__file__).resolve().parents[1] / "src" -if str(SRC) not in sys.path: - sys.path.insert(0, str(SRC)) +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +TESTS = Path(__file__).resolve().parent +for path in (SRC, TESTS): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..6d39d0c --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from collections.abc import Mapping + +from kings_guard.adapters import observation_from_audit_event +from kings_guard.contracts import ImmuneObservation +from kings_guard.fixtures import QontoAssistantPilotFixture, load_qonto_assistant_pilot + + +def load_pilot() -> QontoAssistantPilotFixture: + return load_qonto_assistant_pilot() + + +def observation_from_fixture( + fixture: QontoAssistantPilotFixture | None = None, + event: Mapping[str, object] | None = None, +) -> ImmuneObservation: + fixture = fixture or load_qonto_assistant_pilot() + return observation_from_audit_event( + event or fixture.audit_event, + subject_id=fixture.normalization_hints["subject_id"], + capability_scope=fixture.normalization_hints["capability_scope"], + identity_binding=fixture.normalization_hints["identity_binding"], + egress_destination=fixture.normalization_hints["egress_destination"], + genome=fixture.genome, + ) diff --git a/tests/test_cadence_declaration.py b/tests/test_cadence_declaration.py new file mode 100644 index 0000000..aaab850 --- /dev/null +++ b/tests/test_cadence_declaration.py @@ -0,0 +1,19 @@ +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +DRAFT = ROOT / "specs" / "EmissionCadenceDeclaration.md" + + +def test_emission_cadence_draft_exists_for_taxonomy_handover() -> None: + text = DRAFT.read_text(encoding="utf-8") + assert DRAFT.is_file() + assert "owner: Taxonomy" in text + assert "drafter: kings-guard" in text + assert "qonto-assistant" in text + assert "GH-WP-0002-T04" in text + assert "approval-engine/cadence.yaml" in text + assert "expected rate" in text.lower() or "expected-rate" in text + assert "heartbeat" in text.lower() + 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 diff --git a/tests/test_completeness.py b/tests/test_completeness.py new file mode 100644 index 0000000..acddb1e --- /dev/null +++ b/tests/test_completeness.py @@ -0,0 +1,111 @@ +from kings_guard.cadence import load_qonto_assistant_cadence +from kings_guard.contracts import ( + PostureAssessment, + PostureLevel, + ReconciliationView, + StreamCompleteness, + StreamHeartbeat, + assessment_trust_key, +) +from kings_guard.posture import PostureEvaluator + +from helpers import load_pilot, observation_from_fixture + + +def _healthy_assessment(*, completeness: StreamCompleteness, confidence: int) -> PostureAssessment: + return PostureAssessment( + posture=PostureLevel.HEALTHY, + risk_score=5, + confidence_score=confidence, + stream_completeness=completeness, + completeness_reason="test", + findings=(), + tolerated_findings=(), + rationale="test", + ) + + +def test_completeness_is_separated_from_record_richness() -> None: + fixture = load_pilot() + observation = observation_from_fixture(fixture) + evaluation = PostureEvaluator().evaluate(fixture.genome, observation) + + 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 + + +def test_unmet_cadence_degrades_completeness_and_says_so_in_words() -> None: + fixture = load_pilot() + observation = observation_from_fixture(fixture) + cadence = load_qonto_assistant_cadence() + evaluation = PostureEvaluator().evaluate_with_stream( + fixture.genome, + observation, + cadence, + 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}), + ) + + assert evaluation.assessment.stream_completeness is StreamCompleteness.DEGRADED + assert evaluation.assessment.confidence_score >= 85 + assert "cannot vouch" in evaluation.assessment.rationale + + +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 + ) + unknown = _healthy_assessment(completeness=StreamCompleteness.UNKNOWN, confidence=95) + + assert assessment_trust_key(rich_incomplete) < assessment_trust_key(sparse_complete) + assert assessment_trust_key(unknown) < assessment_trust_key(sparse_complete) + assert assessment_trust_key(unknown) < assessment_trust_key(rich_incomplete) + + +def test_complete_stream_keeps_richness_and_states_completeness() -> None: + fixture = load_pilot() + observation = observation_from_fixture(fixture) + cadence = load_qonto_assistant_cadence() + allow_event = dict(fixture.audit_event) + allow_event["decision"] = "allow" + allow_event["deny_reason"] = None + allow_event["request_id"] = "req-allow-volume" + from kings_guard.adapters import observation_from_audit_event + + allow = observation_from_audit_event( + allow_event, + subject_id=fixture.normalization_hints["subject_id"], + capability_scope=fixture.normalization_hints["capability_scope"], + identity_binding=fixture.normalization_hints["identity_binding"], + egress_destination=fixture.normalization_hints["egress_destination"], + genome=fixture.genome, + ) + evaluation = PostureEvaluator().evaluate_with_stream( + fixture.genome, + observation, + cadence, + now="2026-07-24T09:10:00Z", + observations=(observation, allow), + watching_since="2026-07-22T09:10:00Z", + heartbeats=( + StreamHeartbeat( + source_system="qonto-assistant", + timestamp="2026-07-24T09:00:00Z", + event_class="audit.heartbeat", + assertion="nothing-to-report", + counts={"audit.deny": 1}, + ), + ), + reconciliation=ReconciliationView(source_counts={"audit.deny": 1}, evidence_counts={"audit.deny": 1}), + ) + + assert evaluation.assessment.stream_completeness is StreamCompleteness.COMPLETE + assert "Stream completeness is complete." in evaluation.assessment.rationale + assert evaluation.assessment.confidence_score >= 85 diff --git a/tests/test_contracts.py b/tests/test_contracts.py index 2e3eb07..2f73f64 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -1,11 +1,11 @@ -from kings_guard.adapters import observation_from_audit_event from kings_guard.contracts import PostureLevel, SecurityGenome, as_jsonable -from kings_guard.fixtures import load_qonto_assistant_pilot 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_qonto_assistant_pilot() + fixture = load_pilot() assert isinstance(fixture.genome, SecurityGenome) assert fixture.genome.genome_id == "kg:genome:qonto-assistant" @@ -15,14 +15,7 @@ def test_qonto_fixture_loads_a_normalized_genome() -> None: def test_qonto_audit_event_normalizes_to_immune_observation() -> None: - fixture = load_qonto_assistant_pilot() - observation = observation_from_audit_event( - fixture.audit_event, - subject_id=fixture.normalization_hints["subject_id"], - capability_scope=fixture.normalization_hints["capability_scope"], - identity_binding=fixture.normalization_hints["identity_binding"], - egress_destination=fixture.normalization_hints["egress_destination"], - ) + observation = observation_from_fixture() assert observation.source_system == "qonto-assistant" assert observation.capability == "finance.qonto.read" @@ -33,16 +26,11 @@ def test_qonto_audit_event_normalizes_to_immune_observation() -> None: def test_posture_evaluation_is_jsonable() -> None: - fixture = load_qonto_assistant_pilot() - observation = observation_from_audit_event( - fixture.audit_event, - subject_id=fixture.normalization_hints["subject_id"], - capability_scope=fixture.normalization_hints["capability_scope"], - identity_binding=fixture.normalization_hints["identity_binding"], - egress_destination=fixture.normalization_hints["egress_destination"], - ) + fixture = load_pilot() + observation = observation_from_fixture(fixture) evaluation = PostureEvaluator().evaluate(fixture.genome, observation) payload = as_jsonable(evaluation) assert payload["assessment"]["posture"] == PostureLevel.INFLAMED.value assert payload["signals"][0]["signal_kind"] == "posture_hint" + assert payload["assessment"]["stream_completeness"] == "unknown" diff --git a/tests/test_effector_origin.py b/tests/test_effector_origin.py new file mode 100644 index 0000000..7dab239 --- /dev/null +++ b/tests/test_effector_origin.py @@ -0,0 +1,39 @@ +from kings_guard.contracts import StreamCompleteness + +from helpers import load_pilot, observation_from_fixture +from kings_guard.posture import PostureEvaluator + + +def test_effector_requests_carry_origin_and_do_not_widen_authority() -> None: + fixture = load_pilot() + observation = observation_from_fixture(fixture) + evaluation = PostureEvaluator().evaluate(fixture.genome, observation) + signal = evaluation.signals[0] + + assert signal.effector_requests + for request in signal.effector_requests: + assert request.originating_observation_id == observation.observation_id + assert request.originating_signal_id == signal.signal_id + assert request.stream_completeness is StreamCompleteness.UNKNOWN + assert request.authority_boundary in {"advisory_only", "metadata_only"} + assert request.restrictive_direction in { + "reduce_authority", + "require_step_up", + "request_containment", + "none", + } + + assert signal.effector_requests[0].restrictive_direction == "reduce_authority" + assert signal.effector_requests[1].restrictive_direction == "none" + assert signal.metadata["originating_observation_id"] == observation.observation_id + + +def test_adjacent_boundary_states_receiving_side_origin_expectation() -> None: + from pathlib import Path + + text = (Path(__file__).resolve().parents[1] / "docs" / "AdjacentSystemBoundary.md").read_text( + encoding="utf-8" + ) + assert "originating observation" in text + assert "decision record" in text + assert "does not widen authority" in text diff --git a/tests/test_evidence_class.py b/tests/test_evidence_class.py new file mode 100644 index 0000000..a774cd6 --- /dev/null +++ b/tests/test_evidence_class.py @@ -0,0 +1,91 @@ +import pytest + +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 + classes = {source.event_class: source.evidence_class for source in genome.evidence_sources} + + assert classes["audit.deny"] is EvidenceClass.LOAD_BEARING + assert classes["audit.allow"] is EvidenceClass.ATTRIBUTIVE + assert classes["audit.heartbeat"] is EvidenceClass.LOAD_BEARING + + +def test_observation_copies_declared_class_and_does_not_infer_it() -> None: + fixture = load_pilot() + observation = observation_from_fixture(fixture) + + declared = fixture.genome.source_for("audit.deny") + assert declared is not None + assert observation.evidence_class is declared.evidence_class + assert observation.evidence_class is EvidenceClass.LOAD_BEARING + joined = " ".join(fixture.evidence_class_reasoning) + assert "source declares" in joined + assert "does not infer" in joined + assert all(source.reasoning for source in fixture.genome.evidence_sources) + + +def test_allow_event_is_attributive_because_the_source_said_so() -> None: + fixture = load_pilot() + event = dict(fixture.audit_event) + event["decision"] = "allow" + event["deny_reason"] = None + event["request_id"] = "req-allow" + + observation = observation_from_audit_event( + event, + subject_id=fixture.normalization_hints["subject_id"], + capability_scope=fixture.normalization_hints["capability_scope"], + identity_binding=fixture.normalization_hints["identity_binding"], + egress_destination=fixture.normalization_hints["egress_destination"], + genome=fixture.genome, + ) + + assert observation.event_class == "audit.allow" + assert observation.evidence_class is EvidenceClass.ATTRIBUTIVE + + +def test_class_mismatch_with_source_declaration_is_rejected() -> None: + fixture = load_pilot() + with pytest.raises(ValueError, match="does not match source declaration"): + observation_from_audit_event( + fixture.audit_event, + subject_id=fixture.normalization_hints["subject_id"], + capability_scope=fixture.normalization_hints["capability_scope"], + identity_binding=fixture.normalization_hints["identity_binding"], + egress_destination=fixture.normalization_hints["egress_destination"], + genome=fixture.genome, + evidence_class=EvidenceClass.ATTRIBUTIVE, + ) + + +def test_class_cannot_be_inferred_from_event_contents() -> None: + fixture = load_pilot() + with pytest.raises(ValueError, match="does not infer"): + observation_from_audit_event( + fixture.audit_event, + subject_id=fixture.normalization_hints["subject_id"], + capability_scope=fixture.normalization_hints["capability_scope"], + identity_binding=fixture.normalization_hints["identity_binding"], + egress_destination=fixture.normalization_hints["egress_destination"], + ) + + +def test_declared_source_is_expressible_on_genome_and_observation() -> None: + source = DeclaredEvidenceSource( + source_id="example.audit.deny", + source_system="example", + event_class="audit.deny", + evidence_class=EvidenceClass.LOAD_BEARING, + reasoning="denials are load-bearing by statute §9.6", + cadence_form="heartbeat-or-reconciliation", + ) + assert source.evidence_class is EvidenceClass.LOAD_BEARING + assert observation_from_fixture().evidence_class in { + EvidenceClass.LOAD_BEARING, + EvidenceClass.ATTRIBUTIVE, + } diff --git a/tests/test_immune_memory.py b/tests/test_immune_memory.py new file mode 100644 index 0000000..cdc4c77 --- /dev/null +++ b/tests/test_immune_memory.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import ast +import inspect +from pathlib import Path + +from kings_guard.contracts import ImmuneMemoryEntry +from kings_guard.posture import PostureEvaluator + +ROOT = Path(__file__).resolve().parents[1] +INTENT = ROOT / "INTENT.md" +CONTRACTS = ROOT / "specs" / "ImmuneContracts.md" +SRC = ROOT / "src" / "kings_guard" + + +def test_memory_entry_forbids_runtime_dependency_by_other_layers() -> None: + entry = ImmuneMemoryEntry( + memory_id="mem:example", + subject_scope="qonto-assistant", + summary="credential-exfil probe pattern", + derived_from=("req-qonto-deny-credential-exfil",), + recommended_countermeasures=("lock_actor_temporarily",), + ) + assert entry.runtime_input_for_other_layers == "forbidden" + assert entry.confidentiality == "non-secret" + + +def test_posture_evaluator_does_not_accept_memory_as_input() -> None: + for name in ("evaluate", "evaluate_stream", "evaluate_with_stream"): + signature = inspect.signature(getattr(PostureEvaluator, name)) + for parameter in signature.parameters.values(): + annotation = str(parameter.annotation) + assert "ImmuneMemoryEntry" not in annotation, name + + +def test_no_src_function_takes_memory_as_runtime_input() -> None: + """Catch the drift of wiring immune memory into an engine-facing input.""" + hits: list[str] = [] + for path in SRC.rglob("*.py"): + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + for arg in node.args.args + node.args.kwonlyargs: + annotation = ast.unparse(arg.annotation) if arg.annotation is not None else "" + if "ImmuneMemoryEntry" in annotation and arg.arg not in {"entry", "memory"}: + hits.append(f"{path.name}:{node.name}:{arg.arg}") + if "ImmuneMemoryEntry" in annotation and node.name.startswith("evaluate"): + hits.append(f"{path.name}:{node.name} evaluates from memory") + assert hits == [] + + +def test_intent_stage_five_forbids_memory_as_state_plane() -> None: + text = INTENT.read_text(encoding="utf-8") + collapsed = " ".join(text.split()) + assert "Federated memory" in collapsed + assert "without becoming a state plane" in collapsed + assert "no engine, PEP, or workload may read it as" in collapsed + + +def test_immune_contracts_state_the_state_plane_rule() -> None: + text = CONTRACTS.read_text(encoding="utf-8") + assert "not a state plane" in text.lower() + assert "runtime_input_for_other_layers" in text + assert "Tooling catalog change" in text diff --git a/tests/test_layer_conformance.py b/tests/test_layer_conformance.py index bf71aca..f57b831 100644 --- a/tests/test_layer_conformance.py +++ b/tests/test_layer_conformance.py @@ -64,6 +64,37 @@ def test_checker_passes_on_the_real_tree(): assert result.returncode == 0, result.stderr +def test_agent_principal_rule_checks_are_honest(): + """§3.4 claims that are tests, and claims that remain assertions, are named.""" + data = yaml.safe_load(DECL.read_text()) + checks = data["agent_principal_rule_checks"] + assert checks["no_standing_credential"]["form"] == "test" + assert checks["memory_is_not_a_state_plane"]["form"] == "test" + assert checks["tool_use_shapes"]["form"] == "assertion" + assert checks["reconstructable_as_caller"]["form"] == "mixed" + assert data["agent_principal_rules"]["no_standing_credential"] is True + + +def test_checker_catches_a_standing_credential(tmp_path, monkeypatch): + import importlib.util + + spec = importlib.util.spec_from_file_location("check_layer_conformance", SCRIPT) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + fake_src = tmp_path / "src" / "kings_guard" + fake_src.mkdir(parents=True) + (fake_src / "secrets.py").write_text('VAULT_TOKEN = "s.standing-secret"\n') + (tmp_path / ".env").write_text("OPENBAO_TOKEN=s.also-standing\n") + monkeypatch.setattr(module, "SRC", fake_src) + monkeypatch.setattr(module, "ROOT", tmp_path) + + hits = module.scan_standing_credentials() + assert hits, "a standing credential was not detected — the checker is blind" + kinds = " ".join(reason for _, reason in hits) + assert "credential-shaped file" in kinds or "standing-credential" in kinds + + def test_checker_catches_an_undeclared_tooling_client(tmp_path, monkeypatch): """The negative case: a direct OpenBao client must fail the check. diff --git a/tests/test_live_qonto_observation.py b/tests/test_live_qonto_observation.py new file mode 100644 index 0000000..00d6bf5 --- /dev/null +++ b/tests/test_live_qonto_observation.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import pytest + +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", +) + + +def test_fixture_regression_case_is_retained() -> None: + observation = observation_from_fixture() + assert observation.observation_id == "req-qonto-deny-credential-exfil" + assert observation.deny_reason == "credential_exfil" + + +def test_real_emitted_qonto_events_reach_the_evaluator() -> None: + fixture = load_pilot() + capture = capture_qonto_assistant_events(fixture.genome) + + assert len(capture.events) >= 2 + decisions = {event["decision"] for event in capture.events} + assert "allow" in decisions + assert "deny" in decisions + assert capture.mapping_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) + + assert evaluation.assessment.posture.value in {"elevated", "inflamed"} + assert evaluation.signals + for signal in evaluation.signals: + 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) diff --git a/tests/test_posture.py b/tests/test_posture.py index b1c94bf..d2ec179 100644 --- a/tests/test_posture.py +++ b/tests/test_posture.py @@ -1,17 +1,11 @@ -from kings_guard.adapters import observation_from_audit_event -from kings_guard.fixtures import load_qonto_assistant_pilot from kings_guard.posture import PostureEvaluator +from helpers import load_pilot, observation_from_fixture + def test_qonto_pilot_produces_inflamed_posture_with_tolerance_context() -> None: - fixture = load_qonto_assistant_pilot() - observation = observation_from_audit_event( - fixture.audit_event, - subject_id=fixture.normalization_hints["subject_id"], - capability_scope=fixture.normalization_hints["capability_scope"], - identity_binding=fixture.normalization_hints["identity_binding"], - egress_destination=fixture.normalization_hints["egress_destination"], - ) + fixture = load_pilot() + observation = observation_from_fixture(fixture) evaluation = PostureEvaluator().evaluate(fixture.genome, observation) @@ -22,14 +16,8 @@ def test_qonto_pilot_produces_inflamed_posture_with_tolerance_context() -> None: def test_qonto_pilot_emits_advisory_only_effector_requests() -> None: - fixture = load_qonto_assistant_pilot() - observation = observation_from_audit_event( - fixture.audit_event, - subject_id=fixture.normalization_hints["subject_id"], - capability_scope=fixture.normalization_hints["capability_scope"], - identity_binding=fixture.normalization_hints["identity_binding"], - egress_destination=fixture.normalization_hints["egress_destination"], - ) + fixture = load_pilot() + observation = observation_from_fixture(fixture) evaluation = PostureEvaluator().evaluate(fixture.genome, observation) signal = evaluation.signals[0] diff --git a/tests/test_stream.py b/tests/test_stream.py new file mode 100644 index 0000000..7d9490e --- /dev/null +++ b/tests/test_stream.py @@ -0,0 +1,134 @@ +from kings_guard.cadence import load_qonto_assistant_cadence +from kings_guard.contracts import ( + ReconciliationView, + SignalKind, + StreamCompleteness, + StreamHeartbeat, +) +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() + assert cadence.status == "taxonomy-draft" + assert cadence.owner == "Taxonomy" + assert cadence.drafter == "kings-guard" + assert cadence.source_system == "qonto-assistant" + assert cadence.reference_instance == "GH-WP-0002-T04" + assert cadence.forms() == {"expected-rate", "heartbeat-or-reconciliation"} + + +def test_unmet_declared_rate_is_a_stream_finding() -> None: + cadence = load_qonto_assistant_cadence() + stream = PostureEvaluator().evaluate_stream( + (), + cadence, + now="2026-07-24T09:10:00Z", + watching_since="2026-07-22T09:10:00Z", + heartbeats=( + StreamHeartbeat( + source_system="qonto-assistant", + timestamp="2026-07-24T09:00:00Z", + event_class="audit.heartbeat", + assertion="nothing-to-report", + 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) + assert all(is_stream_finding(item) for item in stream.findings) + assert stream.completeness is StreamCompleteness.DEGRADED + + +def test_missing_heartbeat_is_a_stream_finding() -> None: + fixture = load_pilot() + observation = observation_from_fixture(fixture) + cadence = load_qonto_assistant_cadence() + stream = PostureEvaluator().evaluate_stream( + (observation,), + cadence, + 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}), + ) + + assert any(item.startswith("stream:heartbeat_missing:audit.deny") for item in stream.findings) + assert "credential_exfil_probe" not in stream.findings + assert stream.completeness is StreamCompleteness.DEGRADED + + +def test_reconciliation_divergence_is_a_stream_finding() -> None: + fixture = load_pilot() + observation = observation_from_fixture(fixture) + cadence = load_qonto_assistant_cadence() + stream = PostureEvaluator().evaluate_stream( + (observation,), + cadence, + now="2026-07-23T10:10:00Z", + watching_since="2026-07-23T09:10:00Z", + heartbeats=( + StreamHeartbeat( + source_system="qonto-assistant", + timestamp="2026-07-23T10:00:00Z", + event_class="audit.heartbeat", + assertion="nothing-to-report", + counts={"audit.deny": 2}, + ), + ), + reconciliation=ReconciliationView( + source_counts={"audit.deny": 2}, + evidence_counts={"audit.deny": 1}, + ), + ) + + assert any( + item.startswith("stream:reconciliation_divergence:audit.deny") for item in stream.findings + ) + assert stream.completeness is StreamCompleteness.DEGRADED + + +def test_stream_findings_are_distinguishable_from_content_findings() -> None: + fixture = load_pilot() + observation = observation_from_fixture(fixture) + cadence = load_qonto_assistant_cadence() + evaluation = PostureEvaluator().evaluate_with_stream( + fixture.genome, + observation, + cadence, + 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}), + ) + + content_findings = set(evaluation.assessment.findings) + stream_findings = set(evaluation.stream.findings if evaluation.stream else ()) + assert "credential_exfil_probe" in content_findings + assert stream_findings + assert content_findings.isdisjoint(stream_findings) + assert all(is_stream_finding(item) for item in stream_findings) + kinds = {signal.signal_kind for signal in evaluation.signals} + assert SignalKind.POSTURE_HINT in kinds + assert SignalKind.STREAM_COMPLETENESS in kinds + + +def test_heartbeat_not_yet_due_does_not_false_alarm() -> None: + fixture = load_pilot() + observation = observation_from_fixture(fixture) + cadence = load_qonto_assistant_cadence() + stream = PostureEvaluator().evaluate_stream( + (observation,), + cadence, + now=observation.timestamp, + watching_since=observation.timestamp, + heartbeats=(), + 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) + assert stream.completeness is StreamCompleteness.UNKNOWN diff --git a/workplans/KG-WP-0003-evidence-completeness-and-live-observation.md b/workplans/KG-WP-0003-evidence-completeness-and-live-observation.md index aa779ef..a38ed54 100644 --- a/workplans/KG-WP-0003-evidence-completeness-and-live-observation.md +++ b/workplans/KG-WP-0003-evidence-completeness-and-live-observation.md @@ -4,11 +4,11 @@ type: workplan title: "Evidence completeness and live observation" domain: infotech repo: kings-guard -status: ready +status: finished owner: kings-guard topic_slug: netkingdom created: "2026-08-29" -updated: "2026-08-29" +updated: "2026-09-01" source_review: history/2026-08-29-layer-model-v0.7-scope-intent-review.md standard: net-kingdom/canon/standards/security-layer-model_v0.7.md state_hub_workstream_id: "a6c1c760-1386-58fa-9392-d9c3c4dadc41" @@ -51,7 +51,7 @@ T08 agent-principal conformance checks (after T05) ```task id: KG-WP-0003-T01 -status: todo +status: done priority: high state_hub_task_id: "dd660dbf-9148-5f35-aa4a-74deb10e6f8a" ``` @@ -70,11 +70,17 @@ Done when: reasoning recorded; - tests cover both classes. +**Done 2026-09-01:** `DeclaredEvidenceSource` on the genome and `evidence_class` +on `ImmuneObservation`. The adapter copies the source declaration and refuses +to infer class from event contents. qonto `audit.deny` / `audit.heartbeat` are +load-bearing; `audit.allow` is attributive, with reasoning in the fixture. +`tests/test_evidence_class.py` covers both classes. + ## Task: Draft the emission-cadence declaration and hand it to Taxonomy ```task id: KG-WP-0003-T02 -status: todo +status: done priority: high state_hub_task_id: "5b66aec0-a753-5039-bcbe-bc69cf1ad13e" ``` @@ -104,11 +110,19 @@ Done when: - it is sent to gate-house and to the Taxonomy repositories for ownership, and the handover is recorded. +**Done 2026-09-01:** Draft at `specs/EmissionCadenceDeclaration.md`, worked +example against `qonto-assistant`, both forms specified, genome-adjacent +rationale stated, `GH-WP-0002-T04` / `approval-engine/cadence.yaml` cited as +the heartbeat reference. Runtime consumer is +`src/kings_guard/fixtures/qonto_assistant_cadence.json` (draft-shaped, not a +fork). Handover messages sent to `gate-house`, `net-kingdom`, and +`info-tech-canon`. + ## Task: Evaluate the stream, not only the observation ```task id: KG-WP-0003-T03 -status: todo +status: done priority: high state_hub_task_id: "2cd878c0-8b70-5f2b-9f82-c4d74df5833c" ``` @@ -127,11 +141,17 @@ Done when: not its contents; - no Tooling contact is introduced; the source publishes its own stream. +**Done 2026-09-01:** `PostureEvaluator.evaluate_stream` / `evaluate_with_stream` +compare a declared cadence to arrived events. Unmet rate, missing heartbeat, +and reconciliation divergence emit `stream:*` findings, distinct from content +findings, and a `stream_completeness` signal. No Tooling client. Tests in +`tests/test_stream.py`. + ## Task: Carry stream completeness in the posture output ```task id: KG-WP-0003-T04 -status: todo +status: done priority: high state_hub_task_id: "73367acf-d888-5acc-ae85-7231a9cbe9fe" ``` @@ -149,11 +169,17 @@ Done when: - a posture derived from an incomplete stream can never read as more trustworthy than one derived from a complete one. +**Done 2026-09-01:** `PostureAssessment` carries `stream_completeness` and +`completeness_reason` separately from `confidence_score`. Rationale states +completeness in words. `assessment_trust_key` orders completeness above +richness, so a 95-from-degraded stream cannot outrank a 70-from-complete one. +`tests/test_completeness.py`. + ## Task: State and test that immune memory is not a state plane ```task id: KG-WP-0003-T05 -status: todo +status: done priority: medium state_hub_task_id: "d7ed9faf-58ee-59a1-ad1e-b2fcf817317f" ``` @@ -171,11 +197,16 @@ Done when: - a test asserts the constraint rather than leaving it to prose; - the constraint is reflected in the `Direction of Evolution` stage-5 entry. +**Done 2026-09-01:** `ImmuneMemoryEntry.runtime_input_for_other_layers` is +`forbidden`. `specs/ImmuneContracts.md` states the rule. `tests/test_immune_memory.py` +asserts the evaluator does not take memory as input and that INTENT stage 5 +keeps the constraint. + ## Task: Make containment proposals reconstructable to their origin ```task id: KG-WP-0003-T06 -status: todo +status: done priority: medium state_hub_task_id: "98326ed2-ee5f-58ce-9f99-429e621117a6" ``` @@ -192,11 +223,18 @@ Done when: - `docs/AdjacentSystemBoundary.md` states the expectation on the receiving side; - the authority boundary remains explicit and no value widens authority. +**Done 2026-09-01:** `EffectorRequest` carries originating observation and +signal ids, stream completeness, and a restrictive direction. Authority +boundaries in use remain `advisory_only` and `metadata_only`. Applied +`gate-house/docs/contracts/posture-findings-return.md` through this task. +`docs/AdjacentSystemBoundary.md` rule 7 states the receiving-side expectation. +`tests/test_effector_origin.py`. + ## Task: Observe qonto-assistant in operation ```task id: KG-WP-0003-T07 -status: todo +status: done priority: high state_hub_task_id: "ae516be8-5199-5f81-b47f-7843b5993f97" ``` @@ -220,11 +258,20 @@ Done when: - the fixture is retained as a regression case rather than deleted; - gate-house is told that step four is staffed, with what was found. +**Done 2026-09-01:** Real events from `qonto-assistant.audit.AuditLogger` via +`CapabilityService` (allow + deny) reach the evaluator. Mapping confirmed: +`request_id`, `capability`→`resource_scope`, `decision`, `protocol` hold. +Corrections reported to `qonto-assistant`: `AuditEvent` still omits +`identity_binding` and `egress_destination`; no heartbeat is published yet, so +deny-class completeness is unknown. Fixture retained. Output stays advisory. +gate-house notified that §12 step four is staffed for this lane. +`tests/test_live_qonto_observation.py`; `python3 -m kings_guard.main --pilot qonto-assistant --live`. + ## Task: Check the agent-principal rules that can be checked ```task id: KG-WP-0003-T08 -status: todo +status: done priority: medium state_hub_task_id: "e7003fb3-2e7c-5440-b489-b619b3f0c65d" ``` @@ -243,6 +290,11 @@ Done when: `layer.yaml`, honestly, rather than implied to be checked; - `make check-layer` stays green. +**Done 2026-09-01:** `scripts/check_layer_conformance.py` scans for standing +credential files and secret literals (rule 1). Rule 3 is `tests/test_immune_memory.py`. +`layer.yaml` `agent_principal_rule_checks` records rule 2 as assertion and +rule 4 as mixed. `make check-layer` green; `pyproject.toml` `dependencies = []`. + ## Success criteria 1. Every task above is `done`. @@ -252,3 +304,21 @@ Done when: 5. The emission-cadence draft has been handed to Taxonomy and the handover recorded. 6. gate-house has been told that §12's fourth step is staffed. + +## Residuals + +No leftover kings-guard work is parked only in prose. Source-side follow-ups +were sent as coordination messages, not new workplans here: + +- `qonto-assistant` — emit `identity_binding` and `egress_destination` on + `AuditEvent` if they become known; publish the cadence declaration and a + deny-class heartbeat. Until then deny-class completeness stays unknown. +- Taxonomy (`net-kingdom`, `info-tech-canon`) — take ownership of + `specs/EmissionCadenceDeclaration.md`. kings-guard will consume what they + publish and will not fork it. +- `secrets-engine` secret-use evidence contract — assented as a proposed + Lifecycle surface; observation stays at zero until it ships. Not a WP-0003 + residual. + +Pre-existing open intake `KG-IN-0002` (architecture vocabulary sweep) is +unrelated and remains open.