From 3c549d9b7873c270a0ce93bc952bc21095b0d9f9 Mon Sep 17 00:00:00 2001 From: codex Date: Fri, 24 Jul 2026 00:24:53 +0200 Subject: [PATCH] Implement KG-WP-0002 posture pilot scaffold --- .gitignore | 7 + AGENTS.md | 32 +- Makefile | 15 + README.md | 46 ++- SCOPE.md | 13 +- WORK-RECORDS.md | 10 +- docs/AdjacentSystemBoundary.md | 77 +++++ docs/pilots/QontoAssistantPosturePilot.md | 119 ++++++++ pyproject.toml | 43 +++ specs/ImmuneContracts.md | 276 ++++++++++++++++++ src/kings_guard/__init__.py | 1 + src/kings_guard/adapters/__init__.py | 3 + src/kings_guard/adapters/qonto_assistant.py | 49 ++++ src/kings_guard/contracts.py | 213 ++++++++++++++ src/kings_guard/fixtures.py | 31 ++ .../fixtures/qonto_assistant_pilot.json | 58 ++++ src/kings_guard/main.py | 49 ++++ src/kings_guard/posture.py | 242 +++++++++++++++ tests/conftest.py | 8 + tests/test_contracts.py | 48 +++ tests/test_posture.py | 42 +++ ...ical-immune-contracts-and-posture-pilot.md | 61 +++- 22 files changed, 1418 insertions(+), 25 deletions(-) create mode 100644 Makefile create mode 100644 docs/AdjacentSystemBoundary.md create mode 100644 docs/pilots/QontoAssistantPosturePilot.md create mode 100644 pyproject.toml create mode 100644 specs/ImmuneContracts.md create mode 100644 src/kings_guard/__init__.py create mode 100644 src/kings_guard/adapters/__init__.py create mode 100644 src/kings_guard/adapters/qonto_assistant.py create mode 100644 src/kings_guard/contracts.py create mode 100644 src/kings_guard/fixtures.py create mode 100644 src/kings_guard/fixtures/qonto_assistant_pilot.json create mode 100644 src/kings_guard/main.py create mode 100644 src/kings_guard/posture.py create mode 100644 tests/conftest.py create mode 100644 tests/test_contracts.py create mode 100644 tests/test_posture.py diff --git a/.gitignore b/.gitignore index e4e0199..867b97e 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,10 @@ .claude/* !.claude/rules/ !.claude/rules/*.md + +__pycache__/ +.pytest_cache/ +.ruff_cache/ +build/ +dist/ +*.egg-info/ diff --git a/AGENTS.md b/AGENTS.md index 965ceae..fe1a80c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -119,9 +119,9 @@ curl -s -X PATCH "http://127.0.0.1:8000/tasks/" \ ## Repo-Specific Notes -This repo is currently **docs-first**. There is no application runtime, package -manifest, or test suite yet. Do not invent build/test commands that do not -exist; add them only when the corresponding implementation lands. +This repo is no longer docs-only. It now has a **small Python 3.12 scaffold** +for contract modeling and posture evaluation, but it is still **not** a +long-running control-plane service. ## Working Set @@ -131,25 +131,39 @@ Start with these files: cat INTENT.md cat SCOPE.md sed -n '1,260p' specs/NetKingdomImmuneArchitecture.md -sed -n '1,220p' history/InitialExploration.md +sed -n '1,260p' specs/ImmuneContracts.md +sed -n '1,260p' docs/AdjacentSystemBoundary.md +sed -n '1,260p' docs/pilots/QontoAssistantPosturePilot.md +find src tests -maxdepth 3 -type f | sort ls workplans/ ``` ## Verification -Current verification is structural, not runtime: +Use the repo-local scaffold commands: ```bash -# Check markdown/frontmatter edits and workplan formatting +# Preferred, if make + pip are available +make install-dev + +# Test and lint the scaffold +make test +make lint + +# Demo the current pilot path +make run-demo + +# Direct shell fallback used in minimal environments +python3 -m pytest -q +PYTHONPATH=src python3 -m kings_guard.main --pilot qonto-assistant + +# Check markdown/frontmatter edits too git diff --check # Sync workplan/task state into State Hub after workplan changes cd /home/worsch/state-hub && ./.venv/bin/statehub fix-consistency --repo kings-guard ``` -When this repo gains executable code, extend this section with the real -install/test/lint/run commands in the same change. - --- ## Workplan Convention (ADR-001) diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..d0a1ed0 --- /dev/null +++ b/Makefile @@ -0,0 +1,15 @@ +PYTHON ?= python3 + +.PHONY: install-dev test lint run-demo + +install-dev: + $(PYTHON) -m pip install -e ".[dev]" + +test: + PYTHONPATH=src $(PYTHON) -m pytest -q + +lint: + $(PYTHON) -m ruff check src tests + +run-demo: + PYTHONPATH=src $(PYTHON) -m kings_guard.main --pilot qonto-assistant diff --git a/README.md b/README.md index 4b4c26b..e8bb048 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,47 @@ # kings-guard -Adaptive immune security architecture for netkingdom \ No newline at end of file +Adaptive security contracts and posture-evaluation scaffold for NetKingdom's +immune-architecture work. + +## Current slice + +This repository now contains four aligned pieces: + +- `INTENT.md` / `SCOPE.md` for the repo's stable boundary +- `specs/NetKingdomImmuneArchitecture.md` for the reference architecture +- `specs/ImmuneContracts.md` for the first canonical contract layer +- `src/kings_guard/` plus `tests/` for a minimal posture loop scaffold + +The current implementation is intentionally narrow. It does **not** provide a +running security control plane yet. 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. + +## Repo layout + +- `specs/NetKingdomImmuneArchitecture.md` +- `specs/ImmuneContracts.md` +- `docs/AdjacentSystemBoundary.md` +- `docs/pilots/QontoAssistantPosturePilot.md` +- `src/kings_guard/` +- `tests/` +- `workplans/` + +## Dev commands + +```bash +# preferred, if make + pip are available +make install-dev +make test +make lint +make run-demo + +# direct shell fallback used in minimal environments +python3 -m pytest -q +PYTHONPATH=src python3 -m kings_guard.main --pilot qonto-assistant +``` diff --git a/SCOPE.md b/SCOPE.md index 7d4b9a9..6a5f839 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -49,11 +49,14 @@ without replacing those systems' primary authority. ## Current State -- Markdown-first exploration repo. Current canon is `INTENT.md`, - `specs/NetKingdomImmuneArchitecture.md`, and `history/InitialExploration.md`. -- No executable service, schemas, or integration adapters exist yet. -- The first implementation strand should establish canonical contracts and a - minimal posture pilot before broader integrations. +- Canon now includes `specs/ImmuneContracts.md`, + `docs/AdjacentSystemBoundary.md`, and + `docs/pilots/QontoAssistantPosturePilot.md`. +- A minimal Python reference scaffold exists under `src/kings_guard/` with + fixture-driven tests under `tests/`. +- The implementation currently evaluates normalized observations and emits + posture/signal results for one bounded pilot lane; it is not yet a running + control plane or integrated enforcement service. --- diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 81bb915..1618f86 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -9,11 +9,11 @@ | Kind | ID | Status | Lane | Source | | --- | --- | --- | --- | --- | | workplan | KG-WP-0001 | finished | — | workplans/KG-WP-0001-statehub-bootstrap.md | -| workplan | KG-WP-0002 | active | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | +| workplan | KG-WP-0002 | finished | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.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 | -| task | KG-WP-0002-T01 | todo | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | -| task | KG-WP-0002-T02 | todo | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | -| task | KG-WP-0002-T03 | todo | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | -| task | KG-WP-0002-T04 | todo | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | +| task | KG-WP-0002-T01 | done | — | workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md | +| 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 | diff --git a/docs/AdjacentSystemBoundary.md b/docs/AdjacentSystemBoundary.md new file mode 100644 index 0000000..240f934 --- /dev/null +++ b/docs/AdjacentSystemBoundary.md @@ -0,0 +1,77 @@ +--- +title: Kings Guard Adjacent-System Boundary +version: 0.1.0 +status: Draft +date: 2026-07-23 +repo: kings-guard +classification: Public +--- + +# Kings Guard Adjacent-System Boundary + +## 1. Purpose + +This document records how `kings-guard` interacts with adjacent security and +platform systems in the current NetKingdom stack. + +It exists to prevent overlap drift: `kings-guard` consumes evidence, evaluates +posture, and emits bounded signals. It does not absorb the primary authority of +identity, authorization, secret custody, or operations systems. + +## 2. Cross-System Rules + +The following rules apply to every integration below: + +1. `kings-guard` consumes **non-secret evidence** or references to secret + operations, never raw secret values. +2. `kings-guard` contributes **risk and posture context**, never final + authorization or identity decisions. +3. `kings-guard` emits **bounded signals and effector requests** that remain + advisory unless the owning system already delegates a narrow action lane. +4. `kings-guard` must preserve **tenant isolation**: any retained evidence or + memory must stay bounded by declared confidentiality rules. + +## 3. System-by-System Boundary + +| Adjacent system | Primary authority | Evidence consumed by `kings-guard` | Signals returned by `kings-guard` | Non-goals | +| --- | --- | --- | --- | --- | +| `key-cape` | verified identity, login, MFA, assurance claims | token assurance level, attestation outcomes, claim provenance, authentication anomalies | posture hints about assurance drift or identity inconsistency | minting tokens, login flow, MFA lifecycle | +| `flex-auth` | resource authorization policy, decision logs, protected-system registry | live allow/deny decisions, decision explanations, policy version, decision-rate anomalies | risk context or posture hints that a protected system may choose to include in an auth check | becoming the PDP, registering resources, final allow/deny | +| `secrets-engine` | approved secret workflow and scoped capability delivery | metadata-only records of reads, leases, deliveries, rotations, revocations | posture hints about abnormal secret-use patterns or repeated denied delivery attempts | holding secret values, issuing leases, bypassing approval | +| `railiance-platform` / OpenBao | runtime secret custody, secret engines, dynamic credentials | engine/mount metadata, lease/revocation evidence, platform-secret posture | posture hints for secret-authority consumers or operators | root-token custody, secret-engine configuration ownership | +| `ops-warden` | operational SSH certificate issuance and routing guidance | signing attempts, TTL/principal anomalies, access-routing misuse evidence | sign-request posture hints, actor-scrutiny hints, metadata-only evidence requests | issuing certificates, routing non-SSH credentials, host trust config | +| Railiance runtime layers | provisioning, deployment, isolation, workload lifecycle | deployment provenance, runtime health, workload/network/storage events, recovery outcomes | bounded requests for isolate, rebuild, or validate, always within the runtime's own authority | becoming the deployment repo or cluster operator | +| Governed domain assistants (example: `qonto-assistant`) | service-local business logic, service-local fast loops, customer/domain-facing policy enforcement points | audit events, policy denies, local safety loop outcomes, service-owned genome records | service-local posture hints and metadata-only evidence requests | replacing local policy kernels or business logic | +| `state-hub` | live coordination state and work-record indexing | repo/workplan/task/progress metadata, incident references | non-secret progress events, posture evidence, incident notes | becoming canon, storing secret payloads, driving irreversible response | + +## 4. Governed Domain Assistants + +`qonto-assistant` introduces an important pattern that deserves to be named +explicitly in the boundary model: + +- it is a workload protected by the platform, +- it is also a policy enforcement point in its own right, +- and it already emits an audit stream and operates a local fast loop. + +That class of system is not identical to a generic workload behind +`flex-auth`, and it is not an infrastructure control plane either. + +`kings-guard`'s relationship to this class is: + +- consume its service-local audit stream as `immune_observation`; +- consume its declared healthy intent as `security_genome`; +- emit posture hints that the service may adopt through its own local control + logic; +- never take over its business logic, credentials, or final access policy. + +## 5. Current First-Class Integrations + +The first executable `kings-guard` slice targets one concrete adjacent-system +pattern: + +- **Chosen pilot:** governed domain assistant (`qonto-assistant`) +- **Why first:** already has a real genome record, audit stream, and fast local + loop; lets `kings-guard` validate contracts without waiting for new platform + control paths + +See `docs/pilots/QontoAssistantPosturePilot.md`. diff --git a/docs/pilots/QontoAssistantPosturePilot.md b/docs/pilots/QontoAssistantPosturePilot.md new file mode 100644 index 0000000..fa1e747 --- /dev/null +++ b/docs/pilots/QontoAssistantPosturePilot.md @@ -0,0 +1,119 @@ +--- +title: Qonto Assistant Posture Pilot +version: 0.1.0 +status: Draft +date: 2026-07-23 +repo: kings-guard +classification: Public +--- + +# Qonto Assistant Posture Pilot + +## 1. Decision + +`qonto-assistant` is selected as the first pilot lane for `KG-WP-0002-T04`. + +This is a deliberate override of the original preference order +(`ops-warden` -> `secrets-engine` -> Railiance reconstitution), not because +those lanes are unimportant, but because `qonto-assistant` already provides all +three prerequisites needed for a meaningful first pilot: + +1. a declared `security_genome`; +2. a real audit stream that maps cleanly to `immune_observation`; +3. a service-local fast loop that can consume an advisory posture hint. + +That makes it a better first validation target for the contract layer than a +more abstract platform lane. + +## 2. Why This Lane + +`qonto-assistant` is: + +- reachable by external harness sessions, not only by cluster-internal callers; +- the sole holder of a real company bank credential; +- already instrumented with a deny-escalation local loop; +- already documented as a future `kings-guard` consumer in + `qonto-assistant/docs/SecurityPractice.md`. + +This gives the pilot real risk, real evidence, and a bounded local response +path without granting `kings-guard` any new secret or authorization authority. + +## 3. Pilot Question + +Can `kings-guard` consume a normalized observation derived from +`qonto-assistant`'s existing audit stream, compare it against a declared +security genome, and emit a bounded **posture hint** that the service can use +for its own local fast loop? + +## 4. Request/Response Flow + +```text +client request + -> qonto-assistant policy kernel denies suspicious action + -> qonto-assistant emits audit event + -> kings-guard normalizes event into immune_observation + -> kings-guard evaluates posture against qonto security_genome + -> kings-guard emits posture_hint signal + -> qonto-assistant chooses whether to activate local deny-escalation lockout + -> state-hub receives metadata-only evidence +``` + +## 5. Input Mapping + +| qonto-assistant field | Kings Guard field | Notes | +| --- | --- | --- | +| `request_id` | `observation_id` | Stable correlation id | +| `timestamp` | `timestamp` | Direct mapping | +| `actor` | `actor_id` | Current pilot still tolerates self-asserted identity binding | +| `tenant_id` | `tenant_id` | Direct mapping | +| `capability` | `resource_scope` | Fine-grained route/capability name such as `org_summary` | +| derived constant | `capability` | Coarse scope: `finance.qonto.read` | +| `protocol` | `protocol` | `rest` or `mcp` | +| `decision` | `decision` | `allow` / `deny` | +| `deny_reason` | `deny_reason` | Pilot currently exercises `credential_exfil` | +| derived constant | `identity_binding` | `self_asserted` until key-cape integration lands | +| derived constant | `egress_destination` | `qonto-thirdparty-api` for this pilot | + +## 6. Output Shape + +The pilot emits one `immune_signal` of kind `posture_hint` when posture is +above `healthy`. + +For the initial `credential_exfil` scenario: + +- posture: `inflamed` +- primary finding: `credential_exfil_probe` +- bounded local action hint: `lock_actor_temporarily` +- evidence sink hint: `state-hub` metadata-only incident recording + +## 7. Authority Boundary + +`kings-guard` does **not**: + +- authorize or deny the client request; +- fetch or hold the Qonto credential; +- change `qonto-assistant` policy rules directly; +- mutate OpenBao, `flex-auth`, or `key-cape`. + +`qonto-assistant` keeps those responsibilities. `kings-guard` only emits an +advisory posture hint and metadata-only evidence guidance. + +## 8. Rollback / Safety + +The pilot is easy to disable: + +- if `kings-guard` is unavailable, `qonto-assistant` continues with its current + local policy and deny-escalation behavior; +- if the signal consumer is disabled, audit events still exist and no secret + or authorization authority has moved; +- no secret values are copied into `kings-guard` or `state-hub`. + +## 9. Follow-On Lanes + +Choosing `qonto-assistant` first does **not** reject the original preference +order. It sequences them: + +1. `qonto-assistant` proves the contract layer with a real consumer. +2. `ops-warden` sign-request posture hinting is the next likely platform lane. +3. `secrets-engine` delivery posture hinting follows once the contract is + stable enough to reuse. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..88daa7b --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,43 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "kings-guard" +version = "0.1.0" +description = "Adaptive security contract and posture-evaluation scaffold for complex multi-tenant cloud environments." +readme = "README.md" +requires-python = ">=3.12" +license = { file = "LICENSE" } +authors = [{ name = "Coulomb" }] +dependencies = [] + +[project.optional-dependencies] +dev = [ + "pytest>=8.2,<9.0", + "ruff>=0.6,<1.0", +] + +[project.scripts] +kings-guard-demo = "kings_guard.main:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.package-data] +kings_guard = ["fixtures/*.json"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = [ + "--strict-markers", + "--disable-warnings", + "--tb=short", +] + +[tool.ruff] +line-length = 100 +target-version = "py312" + +[tool.ruff.lint] +select = ["E", "F", "I", "B"] diff --git a/specs/ImmuneContracts.md b/specs/ImmuneContracts.md new file mode 100644 index 0000000..9a6da62 --- /dev/null +++ b/specs/ImmuneContracts.md @@ -0,0 +1,276 @@ +--- +title: Kings Guard Immune Contracts +document_id: KG-CONTRACTS-IMMUNE +version: 0.1.0 +status: Draft +date: 2026-07-23 +repo: kings-guard +classification: Public +--- + +# Kings Guard Immune Contracts + +## 1. Purpose + +This document defines the first canonical contract layer for `kings-guard`. +The goal is to stabilize the shared vocabulary that adjacent systems can target +before a larger decision plane or runtime exists. + +The contracts are deliberately: + +- implementation-neutral; +- small enough to map from existing services and audit streams; +- explicit about producer and consumer responsibility; +- safe to adopt without granting `kings-guard` identity, authorization, or + secret-custody authority. + +The current reference implementation for these contracts lives in +`src/kings_guard/contracts.py`. + +## 2. Contract Principles + +### 2.1 Observation is not interpretation + +An `immune_observation` is a normalized statement about a security-relevant +event or state. It should stay as close as possible to what a source system +knows directly. + +Interpretation happens later through `security_phenotype`, +`posture_assessment`, and `immune_signal`. + +### 2.2 Intent is explicit + +Posture must be evaluated against a declared `security_genome`, not only +against statistical surprise. + +### 2.3 Response stays bounded + +`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 + +`immune_memory_entry` records reusable learning without becoming a raw secret +store or a place to centralize tenant-confidential payloads. + +## 3. Contract Map + +| Contract | Meaning | Typical producers | Typical consumers | +| --- | --- | --- | --- | +| `security_genome` | Canonical healthy intent for a subject or compartment | workload/service owners, platform architects | posture evaluators, admission/review tooling | +| `immune_observation` | Normalized evidence record | services, audit streams, workload sensors, policy kernels | phenotype derivation, posture evaluators, memory pipelines | +| `security_phenotype` | Derived observable state at one moment | `kings-guard` adapters/evaluators, local sentinels | posture evaluators, signal emitters | +| `posture_assessment` | Risk/confidence judgment over current state | `kings-guard` evaluators | signal emitters, human reviewers, future responders | +| `immune_signal` | Typed coordination message about posture or response | `kings-guard` evaluators, local sentinels | local fast loops, owning services, State Hub, future coordinators | +| `effector_request` | Bounded action hint or request | signal emitters | owning services, policy gates, human operators | +| `tolerance` | Explicitly permitted deviation | workload/service owners, governance owners | posture evaluators | +| `inflammation` | Elevated defensive posture mode | posture evaluators, local loops | owning services, operators | +| `immune_memory_entry` | Durable, governed learning artifact | post-incident review, evaluators, future analytics | future evaluators, policy authors | + +## 4. Contract Definitions + +### 4.1 `security_genome` + +The `security_genome` describes intended healthy operation for a subject or +compartment. + +Minimum fields in v0.1: + +- identity of the subject (`subject_id`, `tenant_id`); +- declared purpose; +- permitted capability scope; +- permitted protocols and egress destinations; +- data classifications; +- declared tolerances. + +Produced by: + +- workload/service owners; +- platform architects; +- future admission or deployment tooling. + +Consumed by: + +- `kings-guard` posture evaluators; +- future review, admission, or governance tooling. + +### 4.2 `immune_observation` + +The `immune_observation` is the normalized input record for posture work. + +Minimum fields in v0.1: + +- source system; +- timestamp; +- subject and actor identifiers; +- tenant context; +- capability or resource scope; +- protocol; +- decision/outcome; +- optional deny reason, identity-binding mode, egress destination, and policy + version. + +Produced by: + +- service audit streams; +- workload sensors; +- local policy kernels; +- future platform/runtime sensors. + +Consumed by: + +- phenotype derivation; +- posture evaluation; +- future memory pipelines. + +### 4.3 `security_phenotype` + +The `security_phenotype` is the derived observable state of a subject at a +point in time. + +In v0.1 it is intentionally small: + +- observed capability; +- protocol and decision; +- active findings; +- tolerated findings. + +Produced by: + +- `kings-guard` evaluators or adapters. + +Consumed by: + +- posture assessment; +- signal emission; +- future visualization and debugging surfaces. + +### 4.4 `posture_assessment` + +The `posture_assessment` is the evaluator's judgment. + +Minimum fields in v0.1: + +- posture level (`healthy`, `elevated`, `inflamed`, `compromised`); +- risk score; +- confidence score; +- findings and tolerated findings; +- human-readable rationale. + +Produced by: + +- `kings-guard` evaluators. + +Consumed by: + +- signal emitters; +- future human review or approval surfaces. + +### 4.5 `immune_signal` + +The `immune_signal` is a typed coordination message emitted after assessment. + +Minimum fields in v0.1: + +- signal id; +- signal kind; +- posture; +- summary; +- target system; +- findings; +- optional effector requests and metadata. + +Produced by: + +- `kings-guard` evaluators; +- future local sentinels. + +Consumed by: + +- owning services with local fast loops; +- State Hub for metadata-only evidence; +- future response coordinators. + +### 4.6 `effector_request` + +The `effector_request` is a bounded action request or hint attached to a +signal. + +Minimum fields in v0.1: + +- target system; +- action name; +- authority boundary; +- reason; +- whether human approval is required. + +Produced by: + +- `immune_signal` emitters. + +Consumed by: + +- the system that already owns the action; +- human operators when approval is required. + +### 4.7 `tolerance` + +A `tolerance` is an explicitly declared deviation that should not trigger an +inappropriate response on its own. + +In v0.1 it is represented as a simple match rule: + +- `match_field`; +- `match_value`; +- description; +- effect (`monitor` or `ignore`). + +### 4.8 `inflammation` + +`Inflammation` is not a standalone object in v0.1. It is represented as the +`inflamed` posture level plus one or more `immune_signal` / `effector_request` +records. + +This keeps the first scaffold small while preserving the operating concept. + +### 4.9 `immune_memory_entry` + +An `immune_memory_entry` is the durable, governed output of defensive learning. + +Minimum fields in v0.1: + +- memory id; +- subject scope; +- summary; +- records it was derived from; +- recommended countermeasures; +- confidentiality class. + +Produced by: + +- post-incident review; +- future evaluator pipelines. + +Consumed by: + +- future contract revisions; +- policy authors; +- future cross-run learning surfaces. + +## 5. Current Reference Slice + +`kings-guard`'s first executable slice demonstrates these contracts with one +real pilot: + +- service: `qonto-assistant` +- source evidence: audit event shaped from its current `AuditEvent` +- intent source: normalized genome derived from its existing security-genome + record +- output: posture assessment and advisory-only signal with bounded effector + requests + +See: + +- `docs/AdjacentSystemBoundary.md` +- `docs/pilots/QontoAssistantPosturePilot.md` +- `src/kings_guard/` diff --git a/src/kings_guard/__init__.py b/src/kings_guard/__init__.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/src/kings_guard/__init__.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/src/kings_guard/adapters/__init__.py b/src/kings_guard/adapters/__init__.py new file mode 100644 index 0000000..eeac34d --- /dev/null +++ b/src/kings_guard/adapters/__init__.py @@ -0,0 +1,3 @@ +from kings_guard.adapters.qonto_assistant import observation_from_audit_event + +__all__ = ["observation_from_audit_event"] diff --git a/src/kings_guard/adapters/qonto_assistant.py b/src/kings_guard/adapters/qonto_assistant.py new file mode 100644 index 0000000..5a93e1f --- /dev/null +++ b/src/kings_guard/adapters/qonto_assistant.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from kings_guard.contracts import ImmuneObservation, ObservationDecision + + +def observation_from_audit_event( + event: Mapping[str, Any], + *, + subject_id: str, + capability_scope: str, + identity_binding: str, + egress_destination: str | None, +) -> ImmuneObservation: + """Normalize qonto-assistant's audit stream into Kings Guard's observation contract.""" + return ImmuneObservation( + observation_id=str(event["request_id"]), + source_system="qonto-assistant", + timestamp=str(event["timestamp"]), + tenant_id=str(event["tenant_id"]), + subject_id=subject_id, + actor_id=str(event["actor"]), + capability=capability_scope, + resource_scope=_optional_str(event.get("capability")), + protocol=str(event["protocol"]), + decision=ObservationDecision(str(event["decision"])), + deny_reason=_optional_str(event.get("deny_reason")), + identity_binding=identity_binding, + egress_destination=egress_destination, + latency_ms=_optional_int(event.get("latency_ms")), + result_count=_optional_int(event.get("result_count")), + policy_version=_optional_int(event.get("policy_version")), + upstream_status=_optional_int(event.get("qonto_http_status")), + ) + + +def _optional_int(value: Any) -> int | None: + if value is None: + return None + return int(value) + + +def _optional_str(value: Any) -> str | None: + if value is None: + return None + text = str(value) + return text if text else None diff --git a/src/kings_guard/contracts.py b/src/kings_guard/contracts.py new file mode 100644 index 0000000..7047da0 --- /dev/null +++ b/src/kings_guard/contracts.py @@ -0,0 +1,213 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import asdict, dataclass, field, is_dataclass +from enum import Enum +from typing import Any, Literal + +ToleranceEffect = Literal["monitor", "ignore"] +AuthorityBoundary = Literal[ + "advisory_only", + "metadata_only", + "local_service_owned", + "requires_human_approval", +] + + +class ObservationDecision(str, Enum): + ALLOW = "allow" + DENY = "deny" + ERROR = "error" + + +class PostureLevel(str, Enum): + HEALTHY = "healthy" + ELEVATED = "elevated" + INFLAMED = "inflamed" + COMPROMISED = "compromised" + + +class SignalKind(str, Enum): + POSTURE_HINT = "posture_hint" + OBSERVATION_ALERT = "observation_alert" + RECOVERY_REQUEST = "recovery_request" + + +@dataclass(frozen=True, slots=True) +class ToleranceRule: + tolerance_id: str + match_field: str + match_value: str + description: str + effect: ToleranceEffect = "monitor" + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "ToleranceRule": + return cls( + tolerance_id=str(data["tolerance_id"]), + match_field=str(data["match_field"]), + match_value=str(data["match_value"]), + description=str(data["description"]), + effect=str(data.get("effect", "monitor")), + ) + + +@dataclass(frozen=True, slots=True) +class SecurityGenome: + genome_id: str + version: str + subject_id: str + tenant_id: str + intended_purpose: str + permitted_capabilities: frozenset[str] + permitted_protocols: frozenset[str] + permitted_egress: frozenset[str] + data_classifications: tuple[str, ...] = () + tolerances: tuple[ToleranceRule, ...] = () + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "SecurityGenome": + return cls( + genome_id=str(data["genome_id"]), + version=str(data["version"]), + subject_id=str(data["subject_id"]), + tenant_id=str(data["tenant_id"]), + intended_purpose=str(data["intended_purpose"]), + permitted_capabilities=frozenset(str(item) for item in data["permitted_capabilities"]), + permitted_protocols=frozenset(str(item) for item in data["permitted_protocols"]), + permitted_egress=frozenset(str(item) for item in data["permitted_egress"]), + data_classifications=tuple(str(item) for item in data.get("data_classifications", ())), + tolerances=tuple( + ToleranceRule.from_dict(item) for item in data.get("tolerances", ()) + ), + ) + + +@dataclass(frozen=True, slots=True) +class ImmuneObservation: + observation_id: str + source_system: str + timestamp: str + tenant_id: str + subject_id: str + actor_id: str + capability: str + resource_scope: str | None + protocol: str + decision: ObservationDecision + deny_reason: str | None = None + identity_binding: str | None = None + egress_destination: str | None = None + latency_ms: int | None = None + result_count: int | None = None + policy_version: int | None = None + upstream_status: int | None = None + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "ImmuneObservation": + return cls( + observation_id=str(data["observation_id"]), + source_system=str(data["source_system"]), + timestamp=str(data["timestamp"]), + tenant_id=str(data["tenant_id"]), + subject_id=str(data["subject_id"]), + actor_id=str(data["actor_id"]), + capability=str(data["capability"]), + resource_scope=_optional_str(data.get("resource_scope")), + protocol=str(data["protocol"]), + decision=ObservationDecision(str(data["decision"])), + deny_reason=_optional_str(data.get("deny_reason")), + identity_binding=_optional_str(data.get("identity_binding")), + egress_destination=_optional_str(data.get("egress_destination")), + latency_ms=_optional_int(data.get("latency_ms")), + result_count=_optional_int(data.get("result_count")), + policy_version=_optional_int(data.get("policy_version")), + upstream_status=_optional_int(data.get("upstream_status")), + ) + + +@dataclass(frozen=True, slots=True) +class SecurityPhenotype: + subject_id: str + tenant_id: str + observed_capability: str + protocol: str + decision: ObservationDecision + active_findings: tuple[str, ...] + tolerated_findings: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class PostureAssessment: + posture: PostureLevel + risk_score: int + confidence_score: int + findings: tuple[str, ...] + tolerated_findings: tuple[str, ...] + rationale: str + + +@dataclass(frozen=True, slots=True) +class EffectorRequest: + target_system: str + action: str + authority_boundary: AuthorityBoundary + reason: str + requires_human_approval: bool + + +@dataclass(frozen=True, slots=True) +class ImmuneSignal: + signal_id: str + signal_kind: SignalKind + posture: PostureLevel + summary: str + target_system: str + findings: tuple[str, ...] + effector_requests: tuple[EffectorRequest, ...] = () + metadata: dict[str, str] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class ImmuneMemoryEntry: + memory_id: str + subject_scope: str + summary: str + derived_from: tuple[str, ...] + recommended_countermeasures: tuple[str, ...] + confidentiality: str = "non-secret" + + +@dataclass(frozen=True, slots=True) +class PostureEvaluation: + phenotype: SecurityPhenotype + assessment: PostureAssessment + signals: tuple[ImmuneSignal, ...] + + +def as_jsonable(value: Any) -> Any: + """Convert contract objects into JSON-safe primitives.""" + if isinstance(value, Enum): + return value.value + if is_dataclass(value): + 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()} + if isinstance(value, (list, tuple)): + return [as_jsonable(item) for item in value] + if isinstance(value, (set, frozenset)): + return [as_jsonable(item) for item in sorted(value)] + return value + + +def _optional_int(value: Any) -> int | None: + if value is None: + return None + return int(value) + + +def _optional_str(value: Any) -> str | None: + if value is None: + return None + text = str(value) + return text if text else None diff --git a/src/kings_guard/fixtures.py b/src/kings_guard/fixtures.py new file mode 100644 index 0000000..60b17f4 --- /dev/null +++ b/src/kings_guard/fixtures.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from importlib.resources import files +from typing import Any + +from kings_guard.contracts import SecurityGenome + + +@dataclass(frozen=True, slots=True) +class QontoAssistantPilotFixture: + genome: SecurityGenome + audit_event: dict[str, Any] + normalization_hints: dict[str, str] + source_notes: tuple[str, ...] + + +def load_qonto_assistant_pilot() -> QontoAssistantPilotFixture: + payload = _load_json_fixture("qonto_assistant_pilot.json") + 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()}, + source_notes=tuple(str(item) for item in payload.get("source_notes", ())), + ) + + +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_pilot.json b/src/kings_guard/fixtures/qonto_assistant_pilot.json new file mode 100644 index 0000000..4e2889d --- /dev/null +++ b/src/kings_guard/fixtures/qonto_assistant_pilot.json @@ -0,0 +1,58 @@ +{ + "normalized_genome": { + "genome_id": "kg:genome:qonto-assistant", + "version": "0.1.0", + "subject_id": "qonto-assistant", + "tenant_id": "binky", + "intended_purpose": "Hold the company Qonto credential behind a governed read-only finance surface for authorized operators and agent harnesses.", + "permitted_capabilities": [ + "finance.qonto.read" + ], + "permitted_protocols": [ + "rest", + "mcp" + ], + "permitted_egress": [ + "openbao", + "qonto-thirdparty-api" + ], + "data_classifications": [ + "tenant-confidential", + "financial" + ], + "tolerances": [ + { + "tolerance_id": "self-asserted-actor-claims", + "match_field": "identity_binding", + "match_value": "self_asserted", + "description": "Actor identity is still self-asserted until key-cape integration lands.", + "effect": "monitor" + } + ] + }, + "qonto_audit_event": { + "request_id": "req-qonto-deny-credential-exfil", + "timestamp": "2026-07-23T09:10:00Z", + "actor": "agt-laptop-risky", + "tenant_id": "binky", + "capability": "org_summary", + "protocol": "mcp", + "decision": "deny", + "deny_reason": "credential_exfil", + "policy_version": 1, + "latency_ms": 12, + "qonto_http_status": null, + "result_count": null + }, + "normalization_hints": { + "subject_id": "qonto-assistant", + "capability_scope": "finance.qonto.read", + "identity_binding": "self_asserted", + "egress_destination": "qonto-thirdparty-api" + }, + "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." + ] +} diff --git a/src/kings_guard/main.py b/src/kings_guard/main.py new file mode 100644 index 0000000..4144ab9 --- /dev/null +++ b/src/kings_guard/main.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import argparse +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.posture import PostureEvaluator + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run the Kings Guard pilot posture demo.") + parser.add_argument( + "--pilot", + default="qonto-assistant", + choices=["qonto-assistant"], + help="Pilot bundle to evaluate.", + ) + 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, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/src/kings_guard/posture.py b/src/kings_guard/posture.py new file mode 100644 index 0000000..e68d8fb --- /dev/null +++ b/src/kings_guard/posture.py @@ -0,0 +1,242 @@ +from __future__ import annotations + +from kings_guard.contracts import ( + EffectorRequest, + ImmuneObservation, + ImmuneSignal, + PostureAssessment, + PostureEvaluation, + PostureLevel, + SecurityGenome, + SecurityPhenotype, + SignalKind, +) + +CRITICAL_FINDINGS = frozenset( + { + "tenant_mismatch", + "undeclared_capability", + "unexpected_egress_destination", + "credential_exfil_probe", + } +) +ELEVATED_FINDINGS = frozenset( + { + "constraint_probe_or_boundary_trip", + "unexpected_protocol", + "unverified_identity_binding", + "control_plane_error", + } +) +RISK_WEIGHTS = { + "tenant_mismatch": 50, + "undeclared_capability": 40, + "unexpected_egress_destination": 50, + "credential_exfil_probe": 60, + "constraint_probe_or_boundary_trip": 25, + "unexpected_protocol": 15, + "unverified_identity_binding": 20, + "control_plane_error": 20, +} + + +class PostureEvaluator: + def evaluate(self, genome: SecurityGenome, observation: ImmuneObservation) -> 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) + + def _derive_phenotype( + self, + genome: SecurityGenome, + observation: ImmuneObservation, + ) -> SecurityPhenotype: + findings: list[str] = [] + tolerated: list[str] = [] + + if observation.tenant_id != genome.tenant_id: + findings.append("tenant_mismatch") + if observation.capability not in genome.permitted_capabilities: + findings.append("undeclared_capability") + if observation.protocol not in genome.permitted_protocols: + findings.append("unexpected_protocol") + if ( + observation.egress_destination is not None + and observation.egress_destination not in genome.permitted_egress + ): + findings.append("unexpected_egress_destination") + + if observation.identity_binding and observation.identity_binding != "verified_token": + tolerance = _matching_tolerance(genome, "identity_binding", observation.identity_binding) + if tolerance is None: + findings.append("unverified_identity_binding") + else: + tolerated.append(f"tolerated:{tolerance.tolerance_id}") + + if observation.decision.value == "deny" and observation.deny_reason == "credential_exfil": + findings.append("credential_exfil_probe") + elif observation.decision.value == "deny" and observation.deny_reason == "arg_constraint": + findings.append("constraint_probe_or_boundary_trip") + elif observation.decision.value == "error": + findings.append("control_plane_error") + + return SecurityPhenotype( + subject_id=observation.subject_id, + tenant_id=observation.tenant_id, + observed_capability=observation.capability, + protocol=observation.protocol, + decision=observation.decision, + active_findings=tuple(findings), + tolerated_findings=tuple(tolerated), + ) + + def _assess( + self, + phenotype: SecurityPhenotype, + observation: ImmuneObservation, + ) -> PostureAssessment: + findings = set(phenotype.active_findings) + + posture = PostureLevel.HEALTHY + if findings & CRITICAL_FINDINGS: + posture = PostureLevel.INFLAMED + elif findings: + posture = PostureLevel.ELEVATED + if observation.decision.value == "allow" and "unexpected_egress_destination" in findings: + posture = PostureLevel.COMPROMISED + + risk_score = 5 + sum(RISK_WEIGHTS.get(item, 10) for item in findings) + if posture is PostureLevel.ELEVATED: + risk_score = max(risk_score, 40) + elif posture is PostureLevel.INFLAMED: + risk_score = max(risk_score, 80) + elif posture is PostureLevel.COMPROMISED: + risk_score = max(risk_score, 95) + risk_score = min(risk_score, 100) + + confidence_score = 70 + if observation.policy_version is not None: + confidence_score += 10 + if observation.latency_ms is not None: + confidence_score += 5 + if observation.resource_scope: + confidence_score += 5 + confidence_score = min(confidence_score, 95) + + rationale = _build_rationale( + posture=posture, + findings=phenotype.active_findings, + tolerated_findings=phenotype.tolerated_findings, + ) + return PostureAssessment( + posture=posture, + risk_score=risk_score, + confidence_score=confidence_score, + findings=phenotype.active_findings, + tolerated_findings=phenotype.tolerated_findings, + rationale=rationale, + ) + + def _build_signals( + self, + observation: ImmuneObservation, + assessment: PostureAssessment, + ) -> tuple[ImmuneSignal, ...]: + if assessment.posture is PostureLevel.HEALTHY: + return () + + 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, + posture=assessment.posture, + summary=assessment.rationale, + target_system=observation.source_system, + findings=assessment.findings, + metadata={"source_system": observation.source_system}, + ) + return (signal,) + + def _build_qonto_pilot_signal( + self, + observation: ImmuneObservation, + assessment: PostureAssessment, + ) -> ImmuneSignal: + if "credential_exfil_probe" in assessment.findings: + action = "lock_actor_temporarily" + 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" + reason = ( + "Observed repeated policy-boundary pressure; qonto-assistant should tighten " + "local scrutiny without delegating final authorization to kings-guard." + ) + + return ImmuneSignal( + signal_id=f"sig:{observation.observation_id}", + signal_kind=SignalKind.POSTURE_HINT, + posture=assessment.posture, + summary=reason, + target_system="qonto-assistant", + findings=assessment.findings, + effector_requests=( + EffectorRequest( + target_system="qonto-assistant", + action=action, + authority_boundary="advisory_only", + reason=reason, + requires_human_approval=False, + ), + EffectorRequest( + target_system="state-hub", + action="record_non_secret_incident_evidence", + authority_boundary="metadata_only", + reason="Preserve posture evidence without copying secret values.", + requires_human_approval=False, + ), + ), + metadata={ + "pilot_lane": "qonto-assistant", + "resource_scope": observation.resource_scope or "unknown", + }, + ) + + +def _matching_tolerance( + genome: SecurityGenome, + field_name: str, + field_value: str, +): + for tolerance in genome.tolerances: + if tolerance.match_field == field_name and tolerance.match_value == field_value: + return tolerance + return None + + +def _build_rationale( + *, + posture: PostureLevel, + findings: tuple[str, ...], + tolerated_findings: tuple[str, ...], +) -> str: + if posture is PostureLevel.HEALTHY: + if tolerated_findings: + return ( + "Healthy posture with tolerated deviations only: " + + ", ".join(tolerated_findings) + ) + return "Healthy posture: observation is compatible with declared intent." + + 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}" diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..de705b1 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,8 @@ +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)) diff --git a/tests/test_contracts.py b/tests/test_contracts.py new file mode 100644 index 0000000..2e3eb07 --- /dev/null +++ b/tests/test_contracts.py @@ -0,0 +1,48 @@ +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 + + +def test_qonto_fixture_loads_a_normalized_genome() -> None: + fixture = load_qonto_assistant_pilot() + + assert isinstance(fixture.genome, SecurityGenome) + assert fixture.genome.genome_id == "kg:genome:qonto-assistant" + assert fixture.genome.tenant_id == "binky" + assert "finance.qonto.read" in fixture.genome.permitted_capabilities + assert fixture.normalization_hints["identity_binding"] == "self_asserted" + + +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"], + ) + + assert observation.source_system == "qonto-assistant" + assert observation.capability == "finance.qonto.read" + assert observation.resource_scope == "org_summary" + assert observation.protocol == "mcp" + assert observation.decision.value == "deny" + assert observation.deny_reason == "credential_exfil" + + +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"], + ) + 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" diff --git a/tests/test_posture.py b/tests/test_posture.py new file mode 100644 index 0000000..b1c94bf --- /dev/null +++ b/tests/test_posture.py @@ -0,0 +1,42 @@ +from kings_guard.adapters import observation_from_audit_event +from kings_guard.fixtures import load_qonto_assistant_pilot +from kings_guard.posture import PostureEvaluator + + +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"], + ) + + evaluation = PostureEvaluator().evaluate(fixture.genome, observation) + + assert evaluation.assessment.posture.value == "inflamed" + assert "credential_exfil_probe" in evaluation.assessment.findings + assert "tolerated:self-asserted-actor-claims" in evaluation.assessment.tolerated_findings + assert evaluation.assessment.risk_score >= 80 + + +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"], + ) + + evaluation = PostureEvaluator().evaluate(fixture.genome, observation) + signal = evaluation.signals[0] + + assert signal.target_system == "qonto-assistant" + assert signal.signal_kind.value == "posture_hint" + assert signal.effector_requests[0].authority_boundary == "advisory_only" + assert signal.effector_requests[0].action == "lock_actor_temporarily" + assert signal.effector_requests[1].target_system == "state-hub" + assert signal.effector_requests[1].authority_boundary == "metadata_only" diff --git a/workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md b/workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md index 1b2d663..163f46d 100644 --- a/workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md +++ b/workplans/KG-WP-0002-canonical-immune-contracts-and-posture-pilot.md @@ -4,7 +4,7 @@ type: workplan title: "Canonical immune contracts and first posture pilot" domain: infotech repo: kings-guard -status: active +status: finished owner: codex topic_slug: netkingdom created: "2026-07-23" @@ -59,7 +59,7 @@ T01 canonical contracts ```task id: KG-WP-0002-T01 -status: todo +status: done priority: high state_hub_task_id: "9982a3b4-1e65-493a-9b61-322f23d4fd2d" ``` @@ -75,11 +75,19 @@ Done when: - producer/consumer expectations are named for each contract; - the contracts are usable without requiring one particular product stack. +**Done 2026-07-23:** Added [ImmuneContracts.md](/home/worsch/kings-guard/specs/ImmuneContracts.md), +which defines the v0.1 contract layer for `security_genome`, +`immune_observation`, `security_phenotype`, `posture_assessment`, +`immune_signal`, `effector_request`, `tolerance`, `inflammation`, and +`immune_memory_entry`. The doc names producer/consumer expectations for every +contract and keeps the field set intentionally implementation-neutral. The +matching reference dataclasses live in `src/kings_guard/contracts.py`. + ## Task: Write adjacent-system boundary contract ```task id: KG-WP-0002-T02 -status: todo +status: done priority: high state_hub_task_id: "0c44035e-b1b8-4f5d-8a6c-e6514b4bc897" ``` @@ -94,11 +102,20 @@ Done when: - `kings-guard` inputs, outputs, and non-goals are named per system; - tenant-isolation and non-secret evidence rules are captured. +**Done 2026-07-23:** Added +[AdjacentSystemBoundary.md](/home/worsch/kings-guard/docs/AdjacentSystemBoundary.md), +which records the primary authority, evidence inputs, bounded outputs, and +non-goals for `key-cape`, `flex-auth`, `secrets-engine`, +`railiance-platform` / OpenBao, `ops-warden`, Railiance runtime layers, +`state-hub`, and the governed domain-assistant pattern represented by +`qonto-assistant`. The document also names the cross-system rules for +tenant isolation and metadata-only evidence handling. + ## Task: Scaffold a minimal posture loop ```task id: KG-WP-0002-T03 -status: todo +status: done priority: high state_hub_task_id: "c88a7da6-a9ff-4bd9-ba47-7c199321666b" ``` @@ -114,11 +131,34 @@ Done when: signal; - tests or fixture-driven validation prove the contract shape is stable. +**Done 2026-07-23:** Added the first executable scaffold: + +- `pyproject.toml`, `Makefile`, `README.md`, `AGENTS.md`, and `.gitignore` + updates; +- `src/kings_guard/` with contract dataclasses, a qonto audit-event adapter, + fixture loader, posture evaluator, and CLI demo runner; +- `tests/` with fixture-driven tests for normalization, JSON-safe contract + serialization, posture assessment, and bounded signal emission. + +The sample end-to-end path is: qonto audit event fixture -> +`observation_from_audit_event()` -> `PostureEvaluator.evaluate()` -> +advisory `posture_hint` signal. Verified in this shell with: + +- `git diff --check` +- `python3 -m pytest -q` -> 5 passed +- `python3 -m compileall src tests` +- `PYTHONPATH=src python3 -m kings_guard.main --pilot qonto-assistant` + +`ruff` is declared in the repo's dev dependencies and lint command, but the +current shell lacked both `pip` and the `ruff` module, so source verification +in this session used tests + compile + demo execution rather than the lint +step. + ## Task: Choose and specify the first pilot lane ```task id: KG-WP-0002-T04 -status: todo +status: done priority: medium state_hub_task_id: "77e1dc69-9902-4381-8028-ce1cfac7e9d5" ``` @@ -136,3 +176,14 @@ Done when: - the pilot can run without granting `kings-guard` secret, identity, or final authorization authority; - bounded-response and rollback expectations are documented. + +**Done 2026-07-23:** Selected `qonto-assistant` as the first pilot lane and +documented it in +[QontoAssistantPosturePilot.md](/home/worsch/kings-guard/docs/pilots/QontoAssistantPosturePilot.md). +This deliberately overrides the original preference order because +`qonto-assistant` already exposes all three things the first pilot needs: a +real genome record, a real audit stream, and a real local fast loop. The pilot +flow is concrete and bounded: `kings-guard` consumes normalized observation +evidence, emits an advisory-only posture hint, and may request metadata-only +incident evidence in `state-hub`; it does not gain final authorization, +identity, or secret-custody authority.