From 4cde4e489a197726d455edebbe8eefd2f0e2b2a0 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 29 Aug 2026 12:54:37 +0200 Subject: [PATCH] Stand up the Engine/PIP surface for MAT-WP-0001 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Declare layer.yaml, add a Python engine over a local SQLite store, and cover deterministic assessment, the §13 gap register, stance-map inventory, claim guardrails, and the gate-house review path with tests. Assistant: grok Assistant-Session: 01a04ceb-150e-7e80-a542-ec8b1372e164 --- .gitignore | 15 + AGENTS.md | 56 +++ INTENT.md | 7 +- Makefile | 15 + README.md | 12 +- SCOPE.md | 97 ++-- docs/claim-contract.md | 56 +++ layer.yaml | 70 +++ pyproject.toml | 40 ++ scripts/check_layer_conformance.py | 142 ++++++ src/maturity_engine/__init__.py | 21 + src/maturity_engine/__main__.py | 3 + src/maturity_engine/claims.py | 82 +++ src/maturity_engine/cli.py | 110 ++++ src/maturity_engine/compute.py | 83 ++++ src/maturity_engine/engine.py | 183 +++++++ src/maturity_engine/errors.py | 25 + src/maturity_engine/ids.py | 20 + src/maturity_engine/models.py | 180 +++++++ src/maturity_engine/scoring.py | 87 ++++ src/maturity_engine/seed.py | 299 +++++++++++ src/maturity_engine/store.py | 470 ++++++++++++++++++ src/maturity_engine/timeutil.py | 19 + tests/conftest.py | 40 ++ tests/test_claims_and_guardrail.py | 72 +++ tests/test_compute.py | 96 ++++ tests/test_layer_conformance.py | 83 ++++ tests/test_register.py | 83 ++++ tests/test_review.py | 51 ++ tests/test_stance_and_readiness.py | 16 + ...P-0001-netkingdom-security-layer-engine.md | 16 +- 31 files changed, 2498 insertions(+), 51 deletions(-) create mode 100644 .gitignore create mode 100644 AGENTS.md create mode 100644 Makefile create mode 100644 docs/claim-contract.md create mode 100644 layer.yaml create mode 100644 pyproject.toml create mode 100644 scripts/check_layer_conformance.py create mode 100644 src/maturity_engine/__init__.py create mode 100644 src/maturity_engine/__main__.py create mode 100644 src/maturity_engine/claims.py create mode 100644 src/maturity_engine/cli.py create mode 100644 src/maturity_engine/compute.py create mode 100644 src/maturity_engine/engine.py create mode 100644 src/maturity_engine/errors.py create mode 100644 src/maturity_engine/ids.py create mode 100644 src/maturity_engine/models.py create mode 100644 src/maturity_engine/scoring.py create mode 100644 src/maturity_engine/seed.py create mode 100644 src/maturity_engine/store.py create mode 100644 src/maturity_engine/timeutil.py create mode 100644 tests/conftest.py create mode 100644 tests/test_claims_and_guardrail.py create mode 100644 tests/test_compute.py create mode 100644 tests/test_layer_conformance.py create mode 100644 tests/test_register.py create mode 100644 tests/test_review.py create mode 100644 tests/test_stance_and_readiness.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..aa60068 --- /dev/null +++ b/.gitignore @@ -0,0 +1,15 @@ +.claude/* +!.claude/rules/ +!.claude/rules/*.md + +__pycache__/ +.pytest_cache/ +.ruff_cache/ +build/ +dist/ +*.egg-info/ +.venv/ +venv/ +data/ +*.sqlite +*.sqlite-journal diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..a475368 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,56 @@ +# maturity-engine — Agent Instructions + +## Repo Identity + +**Purpose:** Compute, deterministically, how far a subject has progressed +against declared criteria and submitted evidence, and remember that +progression. Gap register and capability readiness live here. + +**Layer:** Engine (PIP). NetKingdom Security Layer Model v0.7. Never render +or cache an authorization decision. Never compile a level into registry +content. Never let a consumer branch on a fetched level — emit a claim. + +**Domain:** infotech +**Repo slug:** maturity-engine +**Workplan prefix:** `MAT-WP-` + +Machine-readable declaration: `layer.yaml`, checked by +`scripts/check_layer_conformance.py`. Frontmatter in `INTENT.md` must agree. + +--- + +## Stack and commands + +Python ≥ 3.12. Persistence is this PIP's own SQLite store, not OpenBao. + +```bash +python3 -m pip install -e ".[dev]" +make test +make check-layer +make lint +PYTHONPATH=src python3 -m maturity_engine bootstrap --at 2026-08-29T12:00:00Z +``` + +Default database: `data/maturity-engine.sqlite` (gitignored) or +`MATURITY_ENGINE_DB`. + +## Module boundaries + +| Module | Owns | +| --- | --- | +| `models.py` | ladders, levels, criteria, evidence, assessments, gaps | +| `compute.py` | deterministic level computation | +| `store.py` | SQLite + local outbox | +| `scoring.py` | four conformance states; blocked-clean not below conforming | +| `claims.py` | PIP claim shape and guardrail rejections | +| `seed.py` | §13 snapshot, §13.1 inventory, first two ladders as data | +| `engine.py` | facade; no decide/authorize/may | +| `cli.py` | inspect and compute; does not decide | + +Ladder *content* is registered data. ASM-0…ASM-6 is `gate-house`'s. +PEP-stance publication is `ops-warden`'s. + +## State Hub + +API: `http://127.0.0.1:8000`. Log progress at session close. Update task +status when closing a workplan task (`todo` / `progress` / `done`). diff --git a/INTENT.md b/INTENT.md index 8728cb6..8344041 100644 --- a/INTENT.md +++ b/INTENT.md @@ -28,9 +28,10 @@ companion: net-kingdom/SECURITY-COMPANION.md > **Catalog entry (v0.7 §4):** graded progression against declared criteria and > evidence; the gap register; capability readiness (§9.5). > -> The machine-readable form required by §11 is `INTENT.md` frontmatter now, and -> `layer.yaml` once T01 of the alignment workplan lands. Prose cannot -> distinguish a declaration from a transcribed review; the file is the surface. +> The machine-readable form required by §11 is `layer.yaml`, checked by +> `scripts/check_layer_conformance.py`. `INTENT.md` frontmatter must agree. +> Prose cannot distinguish a declaration from a transcribed review; the file +> is the surface. > > Contest this layer if it is wrong — companion §2 asks for that correction > rather than a polite label. diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..4acc685 --- /dev/null +++ b/Makefile @@ -0,0 +1,15 @@ +PYTHON ?= python3 + +.PHONY: install-dev test lint check-layer + +install-dev: + $(PYTHON) -m pip install -e ".[dev]" + +test: + PYTHONPATH=src $(PYTHON) -m pytest -q + +lint: + $(PYTHON) -m ruff check src tests scripts + +check-layer: + $(PYTHON) scripts/check_layer_conformance.py --report diff --git a/README.md b/README.md index 4ca04d7..eb644cf 100644 --- a/README.md +++ b/README.md @@ -29,5 +29,13 @@ Maturity here is graded, evidence-based, open-ended, and revisable — levels ca fall when evidence expires. The closed, binary, replay-proof state machine of an approval belongs to `approval-engine`; the two engines are deliberate opposites. -This repository is seeded, not yet an engine surface. See [INTENT.md](INTENT.md) -and [SCOPE.md](SCOPE.md). The work to stand up the surface is `MAT-WP-0001`. +See [INTENT.md](INTENT.md) and [SCOPE.md](SCOPE.md). The claim contract is +[docs/claim-contract.md](docs/claim-contract.md). + +```bash +python3 -m pip install -e ".[dev]" # or: PYTHONPATH=src +make test +make check-layer +PYTHONPATH=src python3 -m maturity_engine bootstrap --at 2026-08-29T12:00:00Z +PYTHONPATH=src python3 -m maturity_engine review --subject kings-guard --at 2026-08-29T12:00:00Z +``` diff --git a/SCOPE.md b/SCOPE.md index c90a1ea..9c4d9fe 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -8,28 +8,32 @@ ## One-liner -Seeded NetKingdom Engine (PIP) for deterministic maturity assessment, the -estate gap register, and capability readiness — declared in `INTENT.md`, -not yet implemented. +NetKingdom Engine (PIP) that computes a maturity level from declared +criteria and submitted evidence, holds the estate gap register and PEP +stance-map inventory, and emits levels only as claims. --- ## Current Responsibility -`maturity-engine` currently owns the **declaration** of the Engine/PIP -boundary assigned by NetKingdom Security Layer Model v0.7 §4 and restated -in `INTENT.md`: graded progression against declared criteria and evidence, -the gap register, and capability readiness. +`maturity-engine` owns the Engine/PIP surface assigned by NetKingdom +Security Layer Model v0.7 §4: graded progression, the gap register, and +capability readiness. It declares that layer in `INTENT.md` frontmatter +and in `layer.yaml`. -It does **not** yet own a running engine, a store, an API, a register that -can be queried, or a claim contract `access-engine` can consume. Statute -§13 and §13.1 still hold the gap-register snapshot and the PEP stance-map -inventory because this repository cannot store state. +The current implementation is an in-process Python library and CLI over +this PIP's own SQLite store. It registers ladders as data, accepts +evidence with validity windows, computes a deterministic explainable +level, records progression history (including demotion on expiry), +serves the §13 gap-register snapshot, inventories PEP stance maps, +answers readiness as pending / declared-gap / surface-exists, and +emits a `maturity-level` claim. Assessment and register mutations are +queued in a local outbox in the same transaction; nothing calls +`audit-core` inside that transaction. -This repository now **declares** itself Engine, role PIP, in `INTENT.md` -frontmatter against security-layer-model v0.7 and companion v0.2. The -machine-readable `layer.yaml` form, a total client account, and the -engine surface itself are not yet evidenced here. +It does **not** decide whether an actor may act. It does not compile a +level into a registry. It does not author ladder criteria, observation, +or actuation. --- @@ -120,20 +124,29 @@ engine surface itself are not yet evidenced here. ## Current State -- Status: seeded, not implemented -- On disk: `INTENT.md` (v0.7-aligned declaration) and `README.md` -- Layer declaration: `INTENT.md` frontmatter declares `layer: Engine`, - `role: PIP` against security-layer-model v0.7; `layer.yaml` and a - conformance check are not yet present -- Engine surface: none — no API, store, tests, or runtime -- Gap register: still the statute §13 snapshot -- Stance-map inventory: still statute §13.1 (one published map, one absence) -- Claim contract with `access-engine`: not written -- Evidence emission: not present -- Registered ladders: none -- Layout conformance (ITC-REPO-LAYOUT 0.1.0-RC1): **minimal** once this - `SCOPE.md` exists; `history/` and `workplans/` are in use as defined, - `demand/` and `docs/` are not claimed +- Status: engine surface present; production deploy is not claimed +- Layer declaration: `INTENT.md` frontmatter and `layer.yaml` declare + `layer: Engine` / `role: PIP` against security-layer-model v0.7; + `scripts/check_layer_conformance.py` checks them +- Runtime: Python package `maturity_engine` over SQLite (`make test` is + the documented command) +- Assessment: named versioned ladders, evidence with expiry, deterministic + compute, explainability, progression history +- Gap register: statute §13 snapshot is queryable here, with `state` and + owner-status intact; blocked-clean is not scored below conforming +- Stance-map inventory: `ops-warden` published, `ops-mason` unpublished +- Readiness: pending / declared-gap / surface-exists; actuation stays + pending and is not owned here +- Claim contract: `docs/claim-contract.md`; guardrail tests reject a + registry compile and a consumer branch +- Evidence emission: local outbox; load-bearing assessments; heartbeat + for rare classes; no synchronous `audit-core` dependency +- Registered ladders: `asm` v0.3 (owner `gate-house`) and + `pep-stance-publication` v0.1 (owner `ops-warden`) +- Review path: `Engine.review` / `maturity-engine review` computes and + remembers; it does not judge +- Layout conformance (ITC-REPO-LAYOUT 0.1.0-RC1): **minimal**; `docs/`, + `history/`, and `workplans/` are in use; `demand/` is not claimed --- @@ -148,9 +161,8 @@ Declared conformance: **`minimal`** under ITC-REPO-LAYOUT 0.1.0-RC1. Intentional omissions: -- `demand/`, `docs/`, `research/`, `spec/`, `wiki/`, and `issues/` are not - currently claimed. They should be added only when their distinct semantics - are needed, not as empty structural decoration. +- `docs/` holds the claim contract. `demand/`, `research/`, `spec/`, + `wiki/`, and `issues/` are not currently claimed. - Finished workplan archival, when it exists, will follow the Custodian ADR-001 convention at `workplans/archived/YYMMDD-...` so State Hub discovery remains deterministic. @@ -178,17 +190,26 @@ Intentional omissions: - Operative form: `net-kingdom/SECURITY-COMPANION.md` - Alignment review: `history/260829-demand-netkingdom-security-layer-alignment.md` +- Claim contract: `docs/claim-contract.md` +- Layer declaration: `layer.yaml` - Active work: `workplans/` +- Tests: `make test` --- ## Gap to Intent -The seed declaration is in place. Nothing `INTENT.md` says this engine -computes, remembers, or inventories is evidenced on disk yet. The work to -close that gap is `MAT-WP-0001`, sourced from -`history/260829-demand-netkingdom-security-layer-alignment.md`. +`MAT-WP-0001` closed the seed-to-engine gap: a queryable Engine/PIP exists. +Remaining aspiration, not a hole in this surface: + +- `gate-house` actually performing live conformance review through the + CLI rather than in tests; +- ASM-0…ASM-6 assessed against a real estate with evidence fed by the + systems being assessed, not fixtures; +- `access-engine` consuming the claim in a live decision record; +- draining the local outbox into `audit-core` (the queue is here; the + drain adapter is not); +- a production SQLite (or later catalogued) deployment. Do not host a PDP, compile a level into a registry, score blocked-clean -below conforming, or treat observation or containment as available while -closing it. +below conforming, or treat observation or containment as available. diff --git a/docs/claim-contract.md b/docs/claim-contract.md new file mode 100644 index 0000000..e50b0d7 --- /dev/null +++ b/docs/claim-contract.md @@ -0,0 +1,56 @@ +# Maturity level claim contract + +`maturity-engine` is a PIP. A computed level leaves this engine only as a +**request claim** (or as a versioned policy rule authored elsewhere). It is +not an authorization decision. + +Statute: `security-layer-model` v0.7 §6.2, §9.5. + +## Claim shape + +```json +{ + "kind": "maturity-level", + "issuer": "maturity-engine", + "subject": "ops-warden", + "model_id": "asm", + "model_version": "0.3", + "level": 1, + "level_id": "ASM-1", + "assessed_at": "2026-08-29T12:00:00Z", + "assessment_id": "", + "freshness_rule": "assessment evaluated_at plus limiting evidence valid_until", + "digest": "" +} +``` + +`access-engine` consumes this object as an input claim. Reconstructability +is from the decision record that names the claim digest, not from a registry +row. + +## Forbidden + +- Compiling a level into registry content. Until `access-engine` decision + provenance carries a registry-snapshot digest, a level that reached a + decision through the registry is not reconstructable. The engine raises + `GuardrailError` on `compile_into_registry`. +- A consumer branching on a fetched level (`if level >= 3: allow`). That is + a second decision point. The engine raises `GuardrailError` on + `gate_on_level`. +- Any `authorize` / `decide` / `may` surface on this engine. + +## Evidence bound + +Assessment records are **load-bearing** once consumed as claims: emission is +queued in this engine's local SQLite outbox in the same transaction as the +state change. No synchronous `audit-core` call sits inside that transaction. + +The archive proves records were not altered or truncated after arrival. It +does not prove an event never sent. Absence of a record is not evidence of +non-occurrence. + +Rare load-bearing classes use a **heartbeat** (`maturity-engine heartbeat` +via `Engine.heartbeat`), not a rate. + +Gap-register mutations are **attributive** unless a control's soundness +depends on their presence. diff --git a/layer.yaml b/layer.yaml new file mode 100644 index 0000000..f64db80 --- /dev/null +++ b/layer.yaml @@ -0,0 +1,70 @@ +# maturity-engine — NetKingdom security layer declaration +# +# Framework: net-kingdom/canon/standards/security-layer-model_v0.7.md +# Assent: INTENT.md frontmatter (this repository's own voice, §11) +# Validate: python3 scripts/check_layer_conformance.py +# +# §11 requires a machine-readable declaration: prose cannot distinguish a +# declaration from a transcribed review. Form adapted from ops-warden's +# reference and kings-guard's no-contact Engine-adjacent shape. +# +# This repository is an Engine (PIP). Computing a level is not a protected +# side effect, so there is no pep_stance path. Catalogued Tooling (OpenBao, +# key-cape) is not contacted. Persistence is this PIP's own SQLite store. + +schema_version: "0.1" +framework: netkingdom-security-layer-model +standard_version: "0.7" +repository: maturity-engine +layer: engine +role: pip +declared_by: INTENT.md +declared_at: "2026-08-29" + +# §4 catalog entry, transcribed so drift between the catalog and this file is +# visible. The standard is authoritative for the row; this records what we +# understand ourselves to have been assigned. +catalog_entry: + owns: + - graded progression against declared criteria and evidence + - the gap register + - capability readiness + does_not_own: + - authorization decisions + - approval objects + - observation in production + - actuation / containment + +# §5 / §11: every direct contact with a Tooling-layer system (a §4 Tooling row). +# Empty is a claim. This is an Engine API over state it owns, not Staff +# reaching into OpenBao. +tooling_contacts: [] + +# §11 requires non-Tooling clients to be recorded so the check is total. +# Neither target is a §4 Tooling row. Recorded, not policed as §5. +non_tooling_clients: + - id: state-hub-work-records + target: state-hub + layer: not-catalogued + operation: "HTTP to the Custodian State Hub for work records and progress events" + write: true + note: >- + Outside §5 by the v0.5 scope rule: "Tooling-layer system" means a §4 + Tooling row, and state-hub is not one. Carries no security authority + and no secret payload. + - id: sqlite-own-store + target: sqlite + layer: not-catalogued + operation: "this PIP's own transactional store (assessments, register, local outbox)" + write: true + note: >- + Engine-owned persistence, not Lifecycle over catalogued Tooling. + Consumers read claims through this engine's API, not by opening the + file. Listed so the client account is total. Review: 2027-02-28 + (two intervals from 2026-08-29). + +# No PEP stance: this engine does not cause a protected side effect. +declared_shapes: + "5.1": [] + "5.2": [] + "5.3": [] diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..b175194 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,40 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "maturity-engine" +version = "0.1.0" +description = "Deterministic maturity assessment, gap register, and capability readiness for NetKingdom." +readme = "README.md" +requires-python = ">=3.12" +authors = [{ name = "Coulomb" }] +dependencies = [] + +[project.optional-dependencies] +dev = [ + "pytest>=7.4,<9.0", + "ruff>=0.6,<1.0", + "pyyaml>=6.0,<7.0", +] + +[project.scripts] +maturity-engine = "maturity_engine.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["src"] +addopts = [ + "--strict-markers", + "--tb=short", +] + +[tool.ruff] +line-length = 100 +target-version = "py312" + +[tool.ruff.lint] +select = ["E", "F", "I", "B"] diff --git a/scripts/check_layer_conformance.py b/scripts/check_layer_conformance.py new file mode 100644 index 0000000..099f1c7 --- /dev/null +++ b/scripts/check_layer_conformance.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Check maturity-engine against the NetKingdom security layer model (§5, §11). + +This is an Engine (PIP). The checkable claims: + +- layer.yaml declares layer=engine, role=pip +- INTENT.md frontmatter agrees (case-insensitive) +- no pep_stance path +- no catalogued Tooling client (OpenBao, key-cape, cluster) +- sqlite3 is this PIP's own store and is allowed + +Exit 0 clean, 1 undeclared Tooling contact, 2 declaration malformed. +""" +from __future__ import annotations + +import argparse +import ast +import re +import sys +from pathlib import Path + +import yaml + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" / "maturity_engine" +DECL = ROOT / "layer.yaml" +INTENT = ROOT / "INTENT.md" + +TOOLING_IMPORTS = { + "hvac": "OpenBao / Vault client", + "bao": "OpenBao client", + "kubernetes": "cluster client", + "kubernetes_asyncio": "cluster client", + "ldap3": "direct LDAP client (key-cape tooling)", + "python_ldap": "direct LDAP client (key-cape tooling)", +} + + +def load_declaration() -> dict: + if not DECL.exists(): + print(f"FAIL: no declaration at {DECL.relative_to(ROOT)} (§11)", file=sys.stderr) + raise SystemExit(2) + try: + data = yaml.safe_load(DECL.read_text()) + except yaml.YAMLError as exc: + print(f"FAIL: {DECL.name} is not parseable: {exc}", file=sys.stderr) + raise SystemExit(2) from exc + for key in ("layer", "role", "repository", "tooling_contacts", "standard_version"): + if key not in data: + print(f"FAIL: {DECL.name} missing required key '{key}' (§11)", file=sys.stderr) + raise SystemExit(2) + if str(data["layer"]).lower() != "engine": + print(f"FAIL: declared layer is {data['layer']!r}, expected 'engine'", file=sys.stderr) + raise SystemExit(2) + if str(data["role"]).lower() != "pip": + print(f"FAIL: declared role is {data['role']!r}, expected 'pip'", file=sys.stderr) + raise SystemExit(2) + if data.get("pep_stance"): + print("FAIL: pep_stance is set; this engine is not PEP-shaped", file=sys.stderr) + raise SystemExit(2) + return data + + +def intent_frontmatter() -> dict: + text = INTENT.read_text() + match = re.match(r"^---\n(.*?)\n---\n", text, re.DOTALL) + if not match: + print("FAIL: INTENT.md has no YAML frontmatter (§11)", file=sys.stderr) + raise SystemExit(2) + meta = yaml.safe_load(match.group(1)) + if not isinstance(meta, dict): + print("FAIL: INTENT.md frontmatter is not a mapping", file=sys.stderr) + raise SystemExit(2) + return meta + + +def imported_modules(path: Path) -> set[str]: + try: + tree = ast.parse(path.read_text()) + except SyntaxError: + return set() + found: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + found.update(alias.name.split(".")[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom): + if node.level == 0 and node.module: + found.add(node.module.split(".")[0]) + return found + + +def scan() -> list[tuple[Path, str, str]]: + hits: list[tuple[Path, str, str]] = [] + for path in sorted(SRC.rglob("*.py")): + for module in sorted(imported_modules(path)): + if module in TOOLING_IMPORTS: + hits.append((path, module, TOOLING_IMPORTS[module])) + return hits + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--report", action="store_true") + args = parser.parse_args() + + decl = load_declaration() + intent = intent_frontmatter() + if str(intent.get("layer", "")).lower() != str(decl["layer"]).lower(): + print( + f"FAIL: INTENT.md layer {intent.get('layer')!r} != layer.yaml {decl['layer']!r}", + file=sys.stderr, + ) + return 2 + if str(intent.get("role", "")).lower() != str(decl["role"]).lower(): + print( + f"FAIL: INTENT.md role {intent.get('role')!r} != layer.yaml {decl['role']!r}", + file=sys.stderr, + ) + return 2 + + hits = scan() + if hits: + print("FAIL: catalogued Tooling-layer client in an Engine that does not own it", file=sys.stderr) + for path, module, what in hits: + print(f" {path.relative_to(ROOT)}: imports {module!r} — {what}", file=sys.stderr) + return 1 + + if args.report: + print( + f"maturity-engine — layer {decl['layer']}, role {decl['role']}, " + f"standard v{decl['standard_version']}" + ) + print(f" tooling contacts: {len(decl.get('tooling_contacts') or [])}") + print(f" non-tooling clients: {len(decl.get('non_tooling_clients') or [])}") + print(" pep_stance: none") + else: + print(f"OK: Engine/PIP declaration holds; no catalogued Tooling client in {SRC.relative_to(ROOT)}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/maturity_engine/__init__.py b/src/maturity_engine/__init__.py new file mode 100644 index 0000000..6b7cc31 --- /dev/null +++ b/src/maturity_engine/__init__.py @@ -0,0 +1,21 @@ +"""Deterministic maturity assessment, gap register, and capability readiness.""" + +from maturity_engine.claims import Claim +from maturity_engine.engine import Engine +from maturity_engine.errors import GuardrailError, ModelError, UnevaluableCriterion +from maturity_engine.models import Assessment, Criterion, Evidence, Gap, Ladder, Level, StanceMap + +__all__ = [ + "Assessment", + "Claim", + "Criterion", + "Engine", + "Evidence", + "Gap", + "GuardrailError", + "Ladder", + "Level", + "ModelError", + "StanceMap", + "UnevaluableCriterion", +] diff --git a/src/maturity_engine/__main__.py b/src/maturity_engine/__main__.py new file mode 100644 index 0000000..55aacd4 --- /dev/null +++ b/src/maturity_engine/__main__.py @@ -0,0 +1,3 @@ +from maturity_engine.cli import main + +raise SystemExit(main()) diff --git a/src/maturity_engine/claims.py b/src/maturity_engine/claims.py new file mode 100644 index 0000000..5f32ded --- /dev/null +++ b/src/maturity_engine/claims.py @@ -0,0 +1,82 @@ +"""PIP claim shape for a maturity level. + +A claim is a fact `access-engine` may consume. This module does not decide +whether an action is permitted. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from maturity_engine.errors import GuardrailError +from maturity_engine.ids import digest +from maturity_engine.models import Assessment +from maturity_engine.timeutil import format_instant + +ISSUER = "maturity-engine" +CLAIM_KIND = "maturity-level" +FRESHNESS_RULE = "assessment evaluated_at plus limiting evidence valid_until" + + +@dataclass(frozen=True) +class Claim: + kind: str + issuer: str + subject: str + model_id: str + model_version: str + level: int + level_id: str + assessed_at: str + assessment_id: str + freshness_rule: str + digest: str + + def as_dict(self) -> dict: + return { + "kind": self.kind, + "issuer": self.issuer, + "subject": self.subject, + "model_id": self.model_id, + "model_version": self.model_version, + "level": self.level, + "level_id": self.level_id, + "assessed_at": self.assessed_at, + "assessment_id": self.assessment_id, + "freshness_rule": self.freshness_rule, + "digest": self.digest, + } + + +def to_claim(assessment: Assessment) -> Claim: + body = { + "kind": CLAIM_KIND, + "issuer": ISSUER, + "subject": assessment.subject, + "model_id": assessment.model_id, + "model_version": assessment.model_version, + "level": assessment.level, + "level_id": assessment.level_id, + "assessed_at": format_instant(assessment.evaluated_at), + "assessment_id": assessment.id, + "freshness_rule": FRESHNESS_RULE, + } + return Claim(**body, digest=digest(body)) + + +def compile_into_registry(claim: Claim) -> None: + raise GuardrailError( + "A maturity level MUST NOT be compiled into registry content " + "(statute §9.5). Until access-engine decision provenance carries a " + "registry-snapshot digest, a level reaching a decision through the " + "registry is not reconstructable. Use to_claim() as a request claim " + "or a versioned policy rule." + ) + + +def gate_on_level(claim: Claim, *, minimum: int) -> None: + raise GuardrailError( + "A maturity level MUST NOT gate a decision directly (statute §6.1 / " + f"§9.5). Refusing to proceed because {claim.level_id} < {minimum} is " + "a second decision point. The level reaches access-engine as a claim." + ) diff --git a/src/maturity_engine/cli.py b/src/maturity_engine/cli.py new file mode 100644 index 0000000..df7dba1 --- /dev/null +++ b/src/maturity_engine/cli.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path + +from maturity_engine.engine import Engine +from maturity_engine.timeutil import parse_instant + + +def _db(args: argparse.Namespace) -> Path: + if args.db: + return Path(args.db) + env = os.environ.get("MATURITY_ENGINE_DB") + if env: + return Path(env) + return Path("data/maturity-engine.sqlite") + + +def _engine(args: argparse.Namespace) -> Engine: + return Engine.open(_db(args)) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="maturity-engine", + description="Compute levels, remember gaps. Does not decide.", + ) + parser.add_argument("--db", help="sqlite path (default data/maturity-engine.sqlite)") + sub = parser.add_subparsers(dest="cmd", required=True) + + boot = sub.add_parser("bootstrap", help="register seed ladders, §13 snapshot, stance inventory") + boot.add_argument("--at", required=True, help="timezone-aware instant") + + assess = sub.add_parser("assess", help="compute a level from recorded evidence") + assess.add_argument("--subject", required=True) + assess.add_argument("--model", required=True) + assess.add_argument("--at", required=True) + assess.add_argument("--version") + + hist = sub.add_parser("history", help="progression history for a subject") + hist.add_argument("--subject", required=True) + hist.add_argument("--model") + + sub.add_parser("gaps", help="query the gap register") + ready = sub.add_parser("readiness", help="pending | declared-gap | surface-exists") + ready.add_argument("capability") + sub.add_parser("stance", help="PEP stance-map inventory") + + review = sub.add_parser("review", help="compute-and-remember surface for gate-house") + review.add_argument("--subject", required=True) + review.add_argument("--at", required=True) + + sub.add_parser("outbox", help="list locally queued evidence events") + sub.add_parser("models", help="list registered ladders") + + args = parser.parse_args(argv) + engine = _engine(args) + try: + return _dispatch(engine, args) + finally: + engine.close() + + +def _dispatch(engine: Engine, args: argparse.Namespace) -> int: + if args.cmd == "bootstrap": + engine.bootstrap(at=parse_instant(args.at)) + print("bootstrapped") + return 0 + if args.cmd == "assess": + assessment = engine.assess( + args.subject, args.model, at=parse_instant(args.at), version=args.version + ) + print(json.dumps(assessment.as_explanation(), indent=2)) + return 0 + if args.cmd == "history": + print(json.dumps([item.as_explanation() for item in engine.history(args.subject, args.model)], indent=2)) + return 0 + if args.cmd == "gaps": + print(json.dumps([gap.as_dict() for gap in engine.gaps()], indent=2)) + return 0 + if args.cmd == "readiness": + print(json.dumps(engine.readiness(args.capability), indent=2)) + return 0 + if args.cmd == "stance": + print(json.dumps([item.as_dict() for item in engine.stance_maps()], indent=2)) + return 0 + if args.cmd == "review": + print(json.dumps(engine.review(args.subject, at=parse_instant(args.at)).as_dict(), indent=2)) + return 0 + if args.cmd == "outbox": + print(json.dumps(list(engine.pending_events()), indent=2)) + return 0 + if args.cmd == "models": + print( + json.dumps( + [ + {"id": item.id, "version": item.version, "owner": item.owner, "name": item.name} + for item in engine.models() + ], + indent=2, + ) + ) + return 0 + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/maturity_engine/compute.py b/src/maturity_engine/compute.py new file mode 100644 index 0000000..4cde88c --- /dev/null +++ b/src/maturity_engine/compute.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +from datetime import datetime + +from maturity_engine.ids import canonical_json, named_id +from maturity_engine.models import ( + Assessment, + Evidence, + Ladder, + MetCriterion, + UnmetCriterion, +) +from maturity_engine.timeutil import format_instant + + +def compute_level( + ladder: Ladder, + evidence: tuple[Evidence, ...], + *, + subject: str, + instant: datetime, +) -> Assessment: + """Highest consecutive level whose criteria all have valid evidence. + + Evaluation instant is an input. Same ladder, same evidence, same instant, + same assessment. + """ + ladder.validate() + valid = [item for item in evidence if item.valid_at(instant)] + expired = tuple( + item.id + for item in evidence + if not item.valid_at(instant) and item.valid_until is not None and instant >= item.valid_until + ) + by_kind: dict[str, list[Evidence]] = {} + for item in valid: + by_kind.setdefault(item.kind, []).append(item) + + met: list[MetCriterion] = [] + achieved = -1 + unmet: tuple[UnmetCriterion, ...] = () + next_level_id: str | None = None + + for level in ladder.levels: + missing = [ + UnmetCriterion(criterion.id, criterion.evidence_kind, criterion.description) + for criterion in level.criteria + if criterion.evidence_kind not in by_kind + ] + if missing: + unmet = tuple(missing) + next_level_id = level.id + break + for criterion in level.criteria: + ids = tuple(item.id for item in by_kind[criterion.evidence_kind]) + met.append(MetCriterion(criterion.id, criterion.evidence_kind, ids)) + achieved = level.index + + if achieved < 0: + raise RuntimeError("ladder has no floor level 0") + + level = ladder.level_by_index(achieved) + material = { + "subject": subject, + "model_id": ladder.id, + "model_version": ladder.version, + "evaluated_at": format_instant(instant), + "evidence_ids": sorted(item.id for item in evidence), + } + assessment_id = named_id("assessment", canonical_json(material)) + return Assessment( + id=assessment_id, + subject=subject, + model_id=ladder.id, + model_version=ladder.version, + level=achieved, + level_id=level.id, + evaluated_at=instant, + met=tuple(met), + unmet=unmet, + next_level_id=next_level_id, + expired_evidence_ids=expired, + ) diff --git a/src/maturity_engine/engine.py b/src/maturity_engine/engine.py new file mode 100644 index 0000000..15a5709 --- /dev/null +++ b/src/maturity_engine/engine.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +from maturity_engine.claims import Claim, compile_into_registry, gate_on_level, to_claim +from maturity_engine.compute import compute_level +from maturity_engine.errors import UnknownModel +from maturity_engine.ids import named_id +from maturity_engine.models import Assessment, Evidence, Gap, Ladder, StanceMap +from maturity_engine.scoring import ( + capability_readiness, + score, + states_from_gaps, + worst_state, +) +from maturity_engine.seed import ( + asm_ladder, + pep_stance_publication_ladder, + section_13_1_stances, + section_13_gaps, +) +from maturity_engine.store import Store +from maturity_engine.timeutil import format_instant + + +@dataclass(frozen=True) +class ConformanceReview: + subject: str + evaluated_at: datetime + aggregate_state: str + capabilities: tuple[dict, ...] + assessments: tuple[dict, ...] + note: str = ( + "gate-house judges and proposes; maturity-engine computes and remembers. " + "This review is a computation, not a judgment." + ) + + def as_dict(self) -> dict: + return { + "subject": self.subject, + "evaluated_at": format_instant(self.evaluated_at), + "aggregate_state": self.aggregate_state, + "capabilities": list(self.capabilities), + "assessments": list(self.assessments), + "note": self.note, + } + + +class Engine: + """Deterministic API for graded progression, the gap register, and readiness. + + No authorize / decide / may method exists. A maturity level leaves this + process only as a Claim. + """ + + def __init__(self, store: Store) -> None: + self.store = store + + @classmethod + def open(cls, path: Path) -> Engine: + return cls(Store(path)) + + def close(self) -> None: + self.store.close() + + def bootstrap(self, *, at: datetime) -> None: + self.register_model(asm_ladder()) + self.register_model(pep_stance_publication_ladder()) + for gap in section_13_gaps(): + self.store.put_gap(gap, emit_at=at) + for stance in section_13_1_stances(): + self.store.put_stance(stance) + + def register_model(self, ladder: Ladder) -> None: + self.store.put_ladder(ladder) + + def models(self) -> tuple[Ladder, ...]: + return self.store.list_ladders() + + def submit_evidence(self, evidence: Evidence) -> None: + self.store.put_evidence(evidence, emit_at=evidence.submitted_at) + + def assess(self, subject: str, model_id: str, *, at: datetime, version: str | None = None) -> Assessment: + ladder = self.store.get_ladder(model_id, version) + if ladder is None: + raise UnknownModel(f"{model_id}@{version or 'latest'}") + evidence = self.store.evidence_for(subject) + assessment = compute_level(ladder, evidence, subject=subject, instant=at) + self.store.put_assessment(assessment) + return assessment + + def history(self, subject: str, model_id: str | None = None) -> tuple[Assessment, ...]: + return self.store.assessments_for(subject, model_id) + + def claim(self, assessment: Assessment) -> Claim: + return to_claim(assessment) + + def gaps(self) -> tuple[Gap, ...]: + return self.store.list_gaps() + + def upsert_gap(self, gap: Gap, *, at: datetime) -> None: + self.store.put_gap(gap, emit_at=at) + + def readiness(self, capability_id: str) -> dict: + gap = self.store.get_gap(capability_id) + if gap is None: + return {"id": capability_id, "readiness": "pending", "note": "no register row"} + return { + "id": gap.id, + "capability": gap.capability, + "readiness": capability_readiness(gap), + "mark": gap.mark, + "state": gap.state, + "owner_status": gap.owner_status, + "intended_owner": gap.intended_owner, + "owns_actuation": False, + } + + def stance_maps(self) -> tuple[StanceMap, ...]: + return self.store.list_stances() + + def review(self, subject: str, *, at: datetime) -> ConformanceReview: + touching = tuple( + gap + for gap in self.gaps() + if _touches(subject, gap.declared_by) or _touches(subject, gap.intended_owner) + ) + capabilities = tuple( + { + "id": gap.id, + "readiness": capability_readiness(gap), + "state": gap.state, + "owner_status": gap.owner_status, + "mark": gap.mark, + "score": score( + { + "pending": "blocked-clean", + "declared-gap": "declared-gap", + "surface-exists": "conforming", + }.get(capability_readiness(gap), "blocked-clean") + ), + } + for gap in touching + ) + states = states_from_gaps(touching) + assessments = tuple(item.as_explanation() for item in self.history(subject)) + return ConformanceReview( + subject=subject, + evaluated_at=at, + aggregate_state=worst_state(states), + capabilities=capabilities, + assessments=assessments, + ) + + def pending_events(self) -> tuple[dict, ...]: + return self.store.pending_outbox() + + def drain_events(self, *, at: datetime) -> int: + """Mark queued events drained. Does not call audit-core.""" + pending = self.store.pending_outbox() + for item in pending: + self.store.drain_outbox(item["id"], at) + return len(pending) + + def heartbeat(self, *, at: datetime) -> None: + self.store.heartbeat(at) + + def evidence_id(self, subject: str, kind: str, submitted_at: datetime) -> str: + return named_id("evidence", subject, kind, format_instant(submitted_at)) + + +def _touches(subject: str, field: str | None) -> bool: + if not field: + return False + return subject == field or subject in {part.strip() for part in field.replace(";", ",").split(",")} + + +# Guardrail helpers re-exported so tests import one place, and so nobody +# adds a decide() here thinking it is convenient. +reject_registry_compile = compile_into_registry +reject_consumer_branch = gate_on_level diff --git a/src/maturity_engine/errors.py b/src/maturity_engine/errors.py new file mode 100644 index 0000000..895e72e --- /dev/null +++ b/src/maturity_engine/errors.py @@ -0,0 +1,25 @@ +"""Domain errors. None of these is an authorization decision.""" + + +class MaturityError(Exception): + """Base error for this engine.""" + + +class ModelError(MaturityError): + """A ladder cannot be registered as specified.""" + + +class UnevaluableCriterion(ModelError): + """A criterion has no rule; it is not yet a criterion.""" + + +class GuardrailError(MaturityError): + """A caller asked this engine to decide or to compile a level into a registry.""" + + +class UnknownModel(MaturityError): + """No ladder is registered under that id and version.""" + + +class UnknownSubject(MaturityError): + """No facts are recorded for that subject.""" diff --git a/src/maturity_engine/ids.py b/src/maturity_engine/ids.py new file mode 100644 index 0000000..423ba60 --- /dev/null +++ b/src/maturity_engine/ids.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +import hashlib +import json +from typing import Any +from uuid import UUID, uuid5 + +NAMESPACE = uuid5(UUID("6ba7b811-9dad-11d1-80b4-00c04fd430c8"), "net-kingdom/maturity-engine") + + +def canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + + +def digest(value: Any) -> str: + return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() + + +def named_id(*parts: str) -> str: + return str(uuid5(NAMESPACE, "|".join(parts))) diff --git a/src/maturity_engine/models.py b/src/maturity_engine/models.py new file mode 100644 index 0000000..a6145dd --- /dev/null +++ b/src/maturity_engine/models.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from datetime import datetime + +from maturity_engine.errors import UnevaluableCriterion +from maturity_engine.timeutil import format_instant + +KIND_RE = re.compile(r"^[a-z][a-z0-9-]*$") +UNEVALUABLE = {"interpret", "judgment", "opinion", "human", "discretion", "review-discretion"} + + +@dataclass(frozen=True) +class Criterion: + id: str + evidence_kind: str + description: str + + def validate(self) -> None: + kind = self.evidence_kind.strip() + if kind in UNEVALUABLE or not KIND_RE.match(kind): + raise UnevaluableCriterion( + f"criterion {self.id!r} is not evaluable by rule " + f"(evidence_kind={self.evidence_kind!r})" + ) + + +@dataclass(frozen=True) +class Level: + index: int + id: str + name: str + criteria: tuple[Criterion, ...] + + +@dataclass(frozen=True) +class Ladder: + id: str + version: str + owner: str + name: str + levels: tuple[Level, ...] + + def validate(self) -> None: + if not self.levels: + raise UnevaluableCriterion(f"ladder {self.id} has no levels") + indexes = [level.index for level in self.levels] + if indexes != list(range(len(self.levels))): + raise UnevaluableCriterion( + f"ladder {self.id} levels must be consecutive from 0, got {indexes}" + ) + for level in self.levels: + for criterion in level.criteria: + criterion.validate() + + def level_by_index(self, index: int) -> Level: + return self.levels[index] + + +@dataclass(frozen=True) +class Evidence: + id: str + subject: str + kind: str + submitted_by: str + submitted_at: datetime + valid_from: datetime + valid_until: datetime | None + payload: dict = field(default_factory=dict) + + def valid_at(self, instant: datetime) -> bool: + if instant < self.valid_from: + return False + if self.valid_until is not None and instant >= self.valid_until: + return False + return True + + +@dataclass(frozen=True) +class MetCriterion: + criterion_id: str + evidence_kind: str + evidence_ids: tuple[str, ...] + + +@dataclass(frozen=True) +class UnmetCriterion: + criterion_id: str + evidence_kind: str + description: str + + +@dataclass(frozen=True) +class Assessment: + id: str + subject: str + model_id: str + model_version: str + level: int + level_id: str + evaluated_at: datetime + met: tuple[MetCriterion, ...] + unmet: tuple[UnmetCriterion, ...] + next_level_id: str | None + expired_evidence_ids: tuple[str, ...] + + def as_explanation(self) -> dict: + return { + "assessment_id": self.id, + "subject": self.subject, + "model_id": self.model_id, + "model_version": self.model_version, + "level": self.level, + "level_id": self.level_id, + "evaluated_at": format_instant(self.evaluated_at), + "met": [ + { + "criterion_id": item.criterion_id, + "evidence_kind": item.evidence_kind, + "evidence_ids": list(item.evidence_ids), + } + for item in self.met + ], + "unmet": [ + { + "criterion_id": item.criterion_id, + "evidence_kind": item.evidence_kind, + "description": item.description, + } + for item in self.unmet + ], + "next_level_id": self.next_level_id, + "expired_evidence_ids": list(self.expired_evidence_ids), + } + + +@dataclass(frozen=True) +class Gap: + id: str + capability: str + intended_owner: str | None + blocked_on: str | None + review: str | None + state: str + owner_status: str + declared_by: str + mark: str | None + notes: str | None = None + + def as_dict(self) -> dict: + return { + "id": self.id, + "capability": self.capability, + "intended_owner": self.intended_owner, + "blocked_on": self.blocked_on, + "review": self.review, + "state": self.state, + "owner_status": self.owner_status, + "declared_by": self.declared_by, + "mark": self.mark, + "notes": self.notes, + } + + +@dataclass(frozen=True) +class StanceMap: + consumer: str + published: bool + path: str | None + shape: str + + def as_dict(self) -> dict: + return { + "consumer": self.consumer, + "published": self.published, + "path": self.path, + "shape": self.shape, + "absence": not self.published, + } diff --git a/src/maturity_engine/scoring.py b/src/maturity_engine/scoring.py new file mode 100644 index 0000000..fa9ee5c --- /dev/null +++ b/src/maturity_engine/scoring.py @@ -0,0 +1,87 @@ +"""Four conformance states and the scoring rule from statute §11. + +blocked-clean MUST NOT rank below conforming. The two are equal. Declared +gap is tracked non-conformance. Undeclared violation is worse. +""" + +from __future__ import annotations + +from maturity_engine.models import Gap + +CONFORMING = "conforming" +BLOCKED_CLEAN = "blocked-clean" +DECLARED_GAP = "declared-gap" +UNDECLARED_VIOLATION = "undeclared-violation" + +STATES = (CONFORMING, BLOCKED_CLEAN, DECLARED_GAP, UNDECLARED_VIOLATION) + +# Higher is better. Equal scores mean neither ranks below the other. +SCORE = { + CONFORMING: 2, + BLOCKED_CLEAN: 2, + DECLARED_GAP: 1, + UNDECLARED_VIOLATION: 0, +} + +ASSIGNED_STATUSES = {"assigned", "assented", "resolved"} + + +def score(state: str) -> int: + if state not in SCORE: + raise ValueError(f"unknown conformance state {state!r}") + return SCORE[state] + + +def ranks_below(left: str, right: str) -> bool: + return score(left) < score(right) + + +def mark_for(gap: Gap) -> str | None: + return gap.mark + + +def capability_readiness(gap: Gap) -> str: + """pending | declared-gap | surface-exists. Per capability, not per repo.""" + if gap.owner_status in ASSIGNED_STATUSES or gap.state == "resolved": + return "surface-exists" + if gap.mark == "declared-gap" or gap.state == "declared-contact": + return "declared-gap" + if gap.mark == "pending" or gap.state == "unowned-capability": + return "pending" + return "pending" + + +def states_from_gaps(gaps: tuple[Gap, ...]) -> tuple[str, ...]: + if not gaps: + return (CONFORMING,) + found: list[str] = [] + for gap in gaps: + readiness = capability_readiness(gap) + if gap.state == "undeclared-violation" or gap.mark == "undeclared-violation": + found.append(UNDECLARED_VIOLATION) + elif readiness == "declared-gap": + found.append(DECLARED_GAP) + elif readiness == "pending": + found.append(BLOCKED_CLEAN) + else: + found.append(CONFORMING) + return tuple(found) + + +def worst_state(states: tuple[str, ...]) -> str: + """Aggregate that never ranks blocked-clean below conforming. + + A repo that is only blocked-clean stays blocked-clean, which scores equal + to conforming. Declared-gap or undeclared-violation still surface. + """ + if not states: + return CONFORMING + if UNDECLARED_VIOLATION in states: + return UNDECLARED_VIOLATION + if DECLARED_GAP in states: + return DECLARED_GAP + if BLOCKED_CLEAN in states and CONFORMING not in states: + return BLOCKED_CLEAN + if BLOCKED_CLEAN in states and CONFORMING in states: + return CONFORMING + return CONFORMING diff --git a/src/maturity_engine/seed.py b/src/maturity_engine/seed.py new file mode 100644 index 0000000..27ebaf0 --- /dev/null +++ b/src/maturity_engine/seed.py @@ -0,0 +1,299 @@ +"""Seed data: statute §13 snapshot, §13.1 stance inventory, first two ladders. + +Ladder *content* is not authored here. ASM-0…ASM-6 is gate-house's +(Active Secrets Management Canon v0.3 §38). PEP-stance publication is +ops-warden's (ADR-0009). This module registers those ladders as data. +""" + +from __future__ import annotations + +from maturity_engine.models import Criterion, Gap, Ladder, Level, StanceMap + +REVIEW = "2026-11-28" + + +def _c(criterion_id: str, kind: str, description: str) -> Criterion: + return Criterion(id=criterion_id, evidence_kind=kind, description=description) + + +def asm_ladder() -> Ladder: + """ASM-0…ASM-6. Criteria kinds are mechanical; doctrine stays gate-house's.""" + levels = ( + Level(0, "ASM-0", "Embedded", ()), + Level( + 1, + "ASM-1", + "Stored", + ( + _c("asm-1-manager", "central-secret-manager", "Central secret manager exists"), + _c("asm-1-rbac", "secret-rbac", "RBAC on secrets exists"), + _c("asm-1-scan", "basic-secret-scanning", "Basic secret scanning exists"), + _c("asm-1-rotate", "basic-rotation", "Basic rotation exists"), + ), + ), + Level( + 2, + "ASM-2", + "Managed", + ( + _c("asm-2-owner", "secret-ownership", "Ownership is established"), + _c("asm-2-auto", "automated-rotation", "Automated rotation is established"), + _c("asm-2-inv", "secret-inventory", "Inventory is established"), + _c("asm-2-detect", "exposure-detection", "Detection is established"), + _c("asm-2-remediate", "measurable-remediation", "Measurable remediation is established"), + ), + ), + Level( + 3, + "ASM-3", + "Dynamic", + ( + _c("asm-3-fed", "workload-federation", "Workload federation reduces standing credentials"), + _c("asm-3-jit", "jit-access", "JIT access reduces standing credentials"), + _c("asm-3-dyn", "dynamic-secrets", "Dynamic secrets reduce standing credentials"), + _c("asm-3-ci", "secretless-cicd", "Secretless CI/CD reduces standing credentials"), + ), + ), + Level( + 4, + "ASM-4", + "Agent-Aware", + ( + _c("asm-4-split", "principal-actor-distinction", "Human principal and agent actor are distinguished"), + _c("asm-4-asst", "assistant-mode-modeled", "Assistant mode is explicitly modeled"), + _c("asm-4-auto", "autonomous-mode-modeled", "Autonomous mode is explicitly modeled"), + ), + ), + Level( + 5, + "ASM-5", + "Governed Autonomy", + ( + _c("asm-5-id", "agent-identities", "Autonomous agents have identities"), + _c("asm-5-man", "agent-mandates", "Autonomous agents have mandates"), + _c("asm-5-budget", "agent-budgets", "Autonomous agents have budgets"), + _c("asm-5-ceil", "authority-ceilings", "Authority ceilings are in force"), + _c("asm-5-env", "change-envelopes", "Change envelopes are in force"), + _c("asm-5-cb", "circuit-breakers", "Deterministic circuit breakers are in force"), + ), + ), + Level( + 6, + "ASM-6", + "Closed-Loop Authority", + ( + _c("asm-6-id", "identity-join", "Identity is in the reconciled authority lifecycle"), + _c("asm-6-del", "delegation-join", "Delegation is in the reconciled authority lifecycle"), + _c("asm-6-iss", "issuance-join", "Credential issuance is in the reconciled authority lifecycle"), + _c("asm-6-exec", "execution-join", "Execution is in the reconciled authority lifecycle"), + _c("asm-6-ev", "evidence-join", "Evidence is in the reconciled authority lifecycle"), + _c("asm-6-exp", "exposure-join", "Exposure detection is in the reconciled authority lifecycle"), + _c("asm-6-rev", "revocation-join", "Revocation is in the reconciled authority lifecycle"), + _c("asm-6-rem", "remediation-join", "Remediation is in the reconciled authority lifecycle"), + ), + ), + ) + return Ladder( + id="asm", + version="0.3", + owner="gate-house", + name="Active Secrets Management", + levels=levels, + ) + + +def pep_stance_publication_ladder() -> Ladder: + """Publication of unreachable-engine stance maps. Doctrine is ops-warden's.""" + levels = ( + Level(0, "PSP-0", "Unpublished", ()), + Level( + 1, + "PSP-1", + "Published", + (_c("psp-1-file", "stance-map-published", "Stance map published at a named path"),), + ), + Level( + 2, + "PSP-2", + "Tested", + ( + _c( + "psp-2-test", + "stance-map-equality-test", + "A test asserts the published map equals shipped behaviour", + ), + ), + ), + ) + return Ladder( + id="pep-stance-publication", + version="0.1", + owner="ops-warden", + name="PEP stance-map publication", + levels=levels, + ) + + +def section_13_gaps() -> tuple[Gap, ...]: + """Statute §13 snapshot. state and owner_status survive the migration.""" + return ( + Gap( + id="ssh-ca-signing-write", + capability="SSH-CA signing write (VaultCA, bao kv put)", + intended_owner="secrets-engine", + blocked_on="No engine exposes an SSH certificate signing surface", + review=REVIEW, + state="declared-contact", + owner_status="proposed", + declared_by="ops-warden", + mark="declared-gap", + ), + Gap( + id="authentication-assurance-evidence", + capability="Authentication / assurance evidence", + intended_owner="identity layer + audit-core", + blocked_on="access-engine declined (FLEX-DEC-2026-002); no identity-layer evidence surface", + review=REVIEW, + state="unowned-capability", + owner_status="declined", + declared_by="kings-guard", + mark="pending", + notes="access-engine declined; reproposed, not assented", + ), + Gap( + id="secret-use-evidence", + capability="Secret-use evidence", + intended_owner="secrets-engine", + blocked_on="No engine exposes secret-use evidence; OpenBao is Tooling", + review=REVIEW, + state="unowned-capability", + owner_status="proposed", + declared_by="kings-guard", + mark="pending", + ), + Gap( + id="actuation-containment-surface", + capability="Reduce authority, require step-up, isolate a workload — as a deterministic engine API", + intended_owner="access-engine + runtime PEPs", + blocked_on="Ruled in v0.7 §9.2 to be an Engine concept, unowned and held at zero", + review=REVIEW, + state="unowned-capability", + owner_status="proposed", + declared_by="gate-house (estate-wide)", + mark="pending", + notes="kings-guard proposes containment and does not own it", + ), + Gap( + id="identity-and-secret-observation", + capability="Identity and secret observation", + intended_owner="identity layer + secrets-engine + audit-core", + blocked_on="No engine exposes the observation surface; kings-guard makes no Tooling contact", + review=REVIEW, + state="unowned-capability", + owner_status="proposed", + declared_by="kings-guard", + mark="pending", + ), + Gap( + id="stance-map-register", + capability="Stance-map register had no implementation", + intended_owner="gate-house", + blocked_on=None, + review=REVIEW, + state="resolved", + owner_status="resolved", + declared_by="ops-warden, access-engine", + mark=None, + notes="resolved in statute §13.1; inventory now held here", + ), + Gap( + id="registry-snapshot-digest", + capability="Registry-snapshot digest in decision provenance", + intended_owner="flex-auth", + blocked_on="Decision provenance holds no snapshot digest; a registry compile of a level would be unfalsifiable", + review=REVIEW, + state="declared-contact", + owner_status="self-declared", + declared_by="flex-auth", + mark="declared-gap", + ), + Gap( + id="approval-storage-lifecycle", + capability="Approval storage and lifecycle", + intended_owner="approval-engine", + blocked_on=None, + review=REVIEW, + state="assigned", + owner_status="assigned", + declared_by="flex-auth", + mark=None, + notes="assigned (§9.4)", + ), + Gap( + id="approval-evidence", + capability="Approval evidence", + intended_owner="audit-core", + blocked_on=None, + review=REVIEW, + state="assigned", + owner_status="assented", + declared_by="gate-house", + mark=None, + notes="assented (AUDIT-IN-0001)", + ), + Gap( + id="approval-evidence-custody-stronger", + capability="Approval evidence custody stronger than the shipped bound — WORM, object lock, transparency log", + intended_owner=None, + blocked_on="Doctrine work not yet done; approval evidence carries the same bound as every other source", + review=REVIEW, + state="unowned-capability", + owner_status="unassigned", + declared_by="audit-core", + mark="pending", + ), + Gap( + id="approval-emission-atomicity", + capability="Emission atomicity for approval state changes", + intended_owner="approval-engine", + blocked_on=None, + review=REVIEW, + state="assigned", + owner_status="assigned", + declared_by="audit-core", + mark=None, + notes="assigned (§9.4)", + ), + Gap( + id="ssh-signing-non-atomic-audit", + capability="Non-atomic audit emission on the SSH signing lane", + intended_owner="ops-warden", + blocked_on="Declared trade so an audit-store failure cannot remove production host access", + review=REVIEW, + state="declared-contact", + owner_status="self-declared", + declared_by="ops-warden", + mark="declared-gap", + notes="attributive (§9.6)", + ), + ) + + +def section_13_1_stances() -> tuple[StanceMap, ...]: + return ( + StanceMap( + consumer="ops-warden", + published=True, + path="ops-warden/pep-stance.yaml", + shape=( + "total per-zone; open z0–z2 and unknown, closed z3-critical; " + "test asserts the published map equals the shipped default (ADR-0009)" + ), + ), + StanceMap( + consumer="ops-mason", + published=False, + path=None, + shape="catalogued PEP-shaped in §4; map not published", + ), + ) diff --git a/src/maturity_engine/store.py b/src/maturity_engine/store.py new file mode 100644 index 0000000..a421395 --- /dev/null +++ b/src/maturity_engine/store.py @@ -0,0 +1,470 @@ +from __future__ import annotations + +import json +import sqlite3 +from datetime import datetime +from pathlib import Path + +from maturity_engine.models import ( + Assessment, + Criterion, + Evidence, + Gap, + Ladder, + Level, + MetCriterion, + StanceMap, + UnmetCriterion, +) +from maturity_engine.timeutil import format_instant, parse_instant + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS models ( + id TEXT NOT NULL, + version TEXT NOT NULL, + owner TEXT NOT NULL, + name TEXT NOT NULL, + body TEXT NOT NULL, + PRIMARY KEY (id, version) +); + +CREATE TABLE IF NOT EXISTS evidence ( + id TEXT PRIMARY KEY, + subject TEXT NOT NULL, + kind TEXT NOT NULL, + submitted_by TEXT NOT NULL, + submitted_at TEXT NOT NULL, + valid_from TEXT NOT NULL, + valid_until TEXT, + payload TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS assessments ( + id TEXT PRIMARY KEY, + subject TEXT NOT NULL, + model_id TEXT NOT NULL, + model_version TEXT NOT NULL, + level INTEGER NOT NULL, + level_id TEXT NOT NULL, + evaluated_at TEXT NOT NULL, + explanation TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS gaps ( + id TEXT PRIMARY KEY, + capability TEXT NOT NULL, + intended_owner TEXT, + blocked_on TEXT, + review TEXT, + state TEXT NOT NULL, + owner_status TEXT NOT NULL, + declared_by TEXT NOT NULL, + mark TEXT, + notes TEXT +); + +CREATE TABLE IF NOT EXISTS stance_maps ( + consumer TEXT PRIMARY KEY, + published INTEGER NOT NULL, + path TEXT, + shape TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS outbox ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_type TEXT NOT NULL, + payload TEXT NOT NULL, + created_at TEXT NOT NULL, + drained_at TEXT +); +""" + + +class Store: + """This PIP's own transactional store. The outbox lives here, not in audit-core.""" + + def __init__(self, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + self.path = path + self.fail_outbox = False + self._conn = sqlite3.connect(path) + self._conn.row_factory = sqlite3.Row + self._conn.execute("PRAGMA foreign_keys=ON") + self._conn.executescript(SCHEMA) + self._conn.commit() + + def close(self) -> None: + self._conn.close() + + def _commit_or_rollback(self) -> None: + if self.fail_outbox: + self._conn.rollback() + raise RuntimeError("outbox insert refused — state change rolled back") + self._conn.commit() + + def enqueue(self, event_type: str, payload: dict, created_at: datetime) -> None: + if self.fail_outbox: + raise RuntimeError("outbox insert refused") + self._conn.execute( + "INSERT INTO outbox(event_type, payload, created_at) VALUES (?, ?, ?)", + (event_type, json.dumps(payload, sort_keys=True), format_instant(created_at)), + ) + + def put_ladder(self, ladder: Ladder) -> None: + ladder.validate() + body = _ladder_body(ladder) + self._conn.execute( + """ + INSERT INTO models(id, version, owner, name, body) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(id, version) DO UPDATE SET + owner=excluded.owner, name=excluded.name, body=excluded.body + """, + (ladder.id, ladder.version, ladder.owner, ladder.name, json.dumps(body, sort_keys=True)), + ) + self._conn.commit() + + def get_ladder(self, model_id: str, version: str | None = None) -> Ladder | None: + if version is None: + row = self._conn.execute( + "SELECT body FROM models WHERE id = ? ORDER BY version DESC LIMIT 1", + (model_id,), + ).fetchone() + else: + row = self._conn.execute( + "SELECT body FROM models WHERE id = ? AND version = ?", + (model_id, version), + ).fetchone() + if row is None: + return None + return _ladder_from_body(json.loads(row["body"])) + + def list_ladders(self) -> tuple[Ladder, ...]: + rows = self._conn.execute("SELECT body FROM models ORDER BY id, version").fetchall() + return tuple(_ladder_from_body(json.loads(row["body"])) for row in rows) + + def put_evidence(self, evidence: Evidence, *, emit_at: datetime) -> None: + self._conn.execute( + """ + INSERT INTO evidence( + id, subject, kind, submitted_by, submitted_at, valid_from, valid_until, payload + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + subject=excluded.subject, + kind=excluded.kind, + submitted_by=excluded.submitted_by, + submitted_at=excluded.submitted_at, + valid_from=excluded.valid_from, + valid_until=excluded.valid_until, + payload=excluded.payload + """, + ( + evidence.id, + evidence.subject, + evidence.kind, + evidence.submitted_by, + format_instant(evidence.submitted_at), + format_instant(evidence.valid_from), + None if evidence.valid_until is None else format_instant(evidence.valid_until), + json.dumps(evidence.payload, sort_keys=True), + ), + ) + try: + self.enqueue( + "evidence.submitted", + { + "evidence_id": evidence.id, + "subject": evidence.subject, + "kind": evidence.kind, + "class": "attributive", + }, + emit_at, + ) + except RuntimeError: + self._conn.rollback() + raise + self._commit_or_rollback() + + def evidence_for(self, subject: str) -> tuple[Evidence, ...]: + rows = self._conn.execute( + "SELECT * FROM evidence WHERE subject = ? ORDER BY submitted_at, id", + (subject,), + ).fetchall() + return tuple(_evidence_from_row(row) for row in rows) + + def put_assessment(self, assessment: Assessment) -> None: + explanation = assessment.as_explanation() + existed = self._conn.execute( + "SELECT 1 FROM assessments WHERE id = ?", (assessment.id,) + ).fetchone() + self._conn.execute( + """ + INSERT INTO assessments( + id, subject, model_id, model_version, level, level_id, evaluated_at, explanation + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + level=excluded.level, + level_id=excluded.level_id, + explanation=excluded.explanation + """, + ( + assessment.id, + assessment.subject, + assessment.model_id, + assessment.model_version, + assessment.level, + assessment.level_id, + format_instant(assessment.evaluated_at), + json.dumps(explanation, sort_keys=True), + ), + ) + if existed is None: + try: + self.enqueue( + "assessment.recorded", + { + "assessment_id": assessment.id, + "subject": assessment.subject, + "model_id": assessment.model_id, + "level_id": assessment.level_id, + "class": "load-bearing", + "bound": ( + "archive proves records were not altered or truncated after arrival; " + "it does not prove an event never sent" + ), + }, + assessment.evaluated_at, + ) + except RuntimeError: + self._conn.rollback() + raise + self._commit_or_rollback() + + def assessments_for(self, subject: str, model_id: str | None = None) -> tuple[Assessment, ...]: + if model_id is None: + rows = self._conn.execute( + "SELECT explanation FROM assessments WHERE subject = ? ORDER BY evaluated_at, id", + (subject,), + ).fetchall() + else: + rows = self._conn.execute( + """ + SELECT explanation FROM assessments + WHERE subject = ? AND model_id = ? + ORDER BY evaluated_at, id + """, + (subject, model_id), + ).fetchall() + return tuple(_assessment_from_explanation(json.loads(row["explanation"])) for row in rows) + + def put_gap(self, gap: Gap, *, emit_at: datetime) -> None: + self._conn.execute( + """ + INSERT INTO gaps( + id, capability, intended_owner, blocked_on, review, state, + owner_status, declared_by, mark, notes + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + capability=excluded.capability, + intended_owner=excluded.intended_owner, + blocked_on=excluded.blocked_on, + review=excluded.review, + state=excluded.state, + owner_status=excluded.owner_status, + declared_by=excluded.declared_by, + mark=excluded.mark, + notes=excluded.notes + """, + ( + gap.id, + gap.capability, + gap.intended_owner, + gap.blocked_on, + gap.review, + gap.state, + gap.owner_status, + gap.declared_by, + gap.mark, + gap.notes, + ), + ) + try: + self.enqueue( + "gap.mutated", + {"gap_id": gap.id, "state": gap.state, "owner_status": gap.owner_status, "class": "attributive"}, + emit_at, + ) + except RuntimeError: + self._conn.rollback() + raise + self._commit_or_rollback() + + def list_gaps(self) -> tuple[Gap, ...]: + rows = self._conn.execute("SELECT * FROM gaps ORDER BY id").fetchall() + return tuple(_gap_from_row(row) for row in rows) + + def get_gap(self, gap_id: str) -> Gap | None: + row = self._conn.execute("SELECT * FROM gaps WHERE id = ?", (gap_id,)).fetchone() + return None if row is None else _gap_from_row(row) + + def put_stance(self, stance: StanceMap) -> None: + self._conn.execute( + """ + INSERT INTO stance_maps(consumer, published, path, shape) + VALUES (?, ?, ?, ?) + ON CONFLICT(consumer) DO UPDATE SET + published=excluded.published, path=excluded.path, shape=excluded.shape + """, + (stance.consumer, 1 if stance.published else 0, stance.path, stance.shape), + ) + self._conn.commit() + + def list_stances(self) -> tuple[StanceMap, ...]: + rows = self._conn.execute("SELECT * FROM stance_maps ORDER BY consumer").fetchall() + return tuple( + StanceMap( + consumer=row["consumer"], + published=bool(row["published"]), + path=row["path"], + shape=row["shape"], + ) + for row in rows + ) + + def pending_outbox(self) -> tuple[dict, ...]: + rows = self._conn.execute( + "SELECT id, event_type, payload, created_at FROM outbox WHERE drained_at IS NULL ORDER BY id" + ).fetchall() + return tuple( + { + "id": row["id"], + "event_type": row["event_type"], + "payload": json.loads(row["payload"]), + "created_at": row["created_at"], + } + for row in rows + ) + + def drain_outbox(self, row_id: int, drained_at: datetime) -> None: + self._conn.execute( + "UPDATE outbox SET drained_at = ? WHERE id = ?", + (format_instant(drained_at), row_id), + ) + self._conn.commit() + + def heartbeat(self, instant: datetime) -> None: + self.enqueue( + "maturity-engine.heartbeat", + { + "class": "load-bearing", + "form": "heartbeat", + "note": "positive claim that can itself go missing; not a rate", + }, + instant, + ) + self._commit_or_rollback() + + +def _ladder_body(ladder: Ladder) -> dict: + return { + "id": ladder.id, + "version": ladder.version, + "owner": ladder.owner, + "name": ladder.name, + "levels": [ + { + "index": level.index, + "id": level.id, + "name": level.name, + "criteria": [ + { + "id": criterion.id, + "evidence_kind": criterion.evidence_kind, + "description": criterion.description, + } + for criterion in level.criteria + ], + } + for level in ladder.levels + ], + } + + +def _ladder_from_body(body: dict) -> Ladder: + levels = [] + for level in body["levels"]: + criteria = tuple( + Criterion( + id=item["id"], + evidence_kind=item["evidence_kind"], + description=item["description"], + ) + for item in level["criteria"] + ) + levels.append(Level(index=level["index"], id=level["id"], name=level["name"], criteria=criteria)) + return Ladder( + id=body["id"], + version=body["version"], + owner=body["owner"], + name=body["name"], + levels=tuple(levels), + ) + + +def _evidence_from_row(row: sqlite3.Row) -> Evidence: + until = row["valid_until"] + return Evidence( + id=row["id"], + subject=row["subject"], + kind=row["kind"], + submitted_by=row["submitted_by"], + submitted_at=parse_instant(row["submitted_at"]), + valid_from=parse_instant(row["valid_from"]), + valid_until=None if until is None else parse_instant(until), + payload=json.loads(row["payload"]), + ) + + +def _gap_from_row(row: sqlite3.Row) -> Gap: + return Gap( + id=row["id"], + capability=row["capability"], + intended_owner=row["intended_owner"], + blocked_on=row["blocked_on"], + review=row["review"], + state=row["state"], + owner_status=row["owner_status"], + declared_by=row["declared_by"], + mark=row["mark"], + notes=row["notes"], + ) + + +def _assessment_from_explanation(body: dict) -> Assessment: + return Assessment( + id=body["assessment_id"], + subject=body["subject"], + model_id=body["model_id"], + model_version=body["model_version"], + level=body["level"], + level_id=body["level_id"], + evaluated_at=parse_instant(body["evaluated_at"]), + met=tuple( + MetCriterion( + criterion_id=item["criterion_id"], + evidence_kind=item["evidence_kind"], + evidence_ids=tuple(item["evidence_ids"]), + ) + for item in body["met"] + ), + unmet=tuple( + UnmetCriterion( + criterion_id=item["criterion_id"], + evidence_kind=item["evidence_kind"], + description=item["description"], + ) + for item in body["unmet"] + ), + next_level_id=body["next_level_id"], + expired_evidence_ids=tuple(body["expired_evidence_ids"]), + ) diff --git a/src/maturity_engine/timeutil.py b/src/maturity_engine/timeutil.py new file mode 100644 index 0000000..120660b --- /dev/null +++ b/src/maturity_engine/timeutil.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from datetime import datetime, timezone + + +def parse_instant(value: str) -> datetime: + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + instant = datetime.fromisoformat(text) + if instant.tzinfo is None: + raise ValueError("evaluation instant must be timezone-aware") + return instant.astimezone(timezone.utc) + + +def format_instant(instant: datetime) -> str: + if instant.tzinfo is None: + raise ValueError("evaluation instant must be timezone-aware") + return instant.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..be57a57 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from maturity_engine.engine import Engine +from maturity_engine.models import Evidence + +INSTANT = datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc) + + +@pytest.fixture +def engine(tmp_path: Path) -> Engine: + inst = Engine.open(tmp_path / "maturity.sqlite") + inst.bootstrap(at=INSTANT) + yield inst + inst.close() + + +def evidence( + engine: Engine, + subject: str, + kind: str, + *, + at: datetime = INSTANT, + until: datetime | None = None, + by: str = "fixture", +) -> Evidence: + return Evidence( + id=engine.evidence_id(subject, kind, at), + subject=subject, + kind=kind, + submitted_by=by, + submitted_at=at, + valid_from=at, + valid_until=until, + payload={"kind": kind}, + ) diff --git a/tests/test_claims_and_guardrail.py b/tests/test_claims_and_guardrail.py new file mode 100644 index 0000000..fe2585a --- /dev/null +++ b/tests/test_claims_and_guardrail.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import pytest + +from maturity_engine.claims import compile_into_registry, gate_on_level +from maturity_engine.engine import Engine +from maturity_engine.errors import GuardrailError +from conftest import INSTANT, evidence + + +def test_claim_shape(engine): + assessment = engine.assess("estate", "asm", at=INSTANT) + claim = engine.claim(assessment) + body = claim.as_dict() + assert body["kind"] == "maturity-level" + assert body["issuer"] == "maturity-engine" + assert body["subject"] == "estate" + assert body["level_id"] == "ASM-0" + assert body["digest"] + assert "freshness_rule" in body + + +def test_registry_compile_is_rejected(engine): + claim = engine.claim(engine.assess("estate", "asm", at=INSTANT)) + with pytest.raises(GuardrailError, match="MUST NOT be compiled into registry"): + compile_into_registry(claim) + + +def test_consumer_branch_is_rejected(engine): + claim = engine.claim(engine.assess("estate", "asm", at=INSTANT)) + with pytest.raises(GuardrailError, match="MUST NOT gate a decision directly"): + gate_on_level(claim, minimum=3) + + +def test_no_decision_surface(): + assert not hasattr(Engine, "authorize") + assert not hasattr(Engine, "decide") + assert not hasattr(Engine, "may") + assert not hasattr(Engine, "allow") + + +def test_assessment_emission_is_local_and_load_bearing(engine): + engine.assess("estate", "asm", at=INSTANT) + events = [item for item in engine.pending_events() if item["event_type"] == "assessment.recorded"] + assert len(events) == 1 + payload = events[0]["payload"] + assert payload["class"] == "load-bearing" + assert "not altered or truncated after arrival" in payload["bound"] + assert "never sent" in payload["bound"] + + +def test_outbox_failure_rolls_back_assessment(engine): + engine.store.fail_outbox = True + with pytest.raises(RuntimeError, match="outbox"): + engine.assess("estate", "pep-stance-publication", at=INSTANT) + assert engine.history("estate", "pep-stance-publication") == () + + +def test_heartbeat_is_a_positive_claim(engine): + engine.heartbeat(at=INSTANT) + beats = [item for item in engine.pending_events() if item["event_type"] == "maturity-engine.heartbeat"] + assert beats + assert beats[0]["payload"]["form"] == "heartbeat" + + +def test_two_ladders_from_different_owners_can_be_claimed(engine): + for kind in ("stance-map-published", "stance-map-equality-test"): + engine.submit_evidence(evidence(engine, "ops-warden", kind)) + asm = engine.claim(engine.assess("ops-warden", "asm", at=INSTANT)) + psp = engine.claim(engine.assess("ops-warden", "pep-stance-publication", at=INSTANT)) + assert asm.model_id != psp.model_id + assert psp.level_id == "PSP-2" diff --git a/tests/test_compute.py b/tests/test_compute.py new file mode 100644 index 0000000..69ff8a2 --- /dev/null +++ b/tests/test_compute.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest + +from maturity_engine.compute import compute_level +from maturity_engine.errors import UnevaluableCriterion +from maturity_engine.models import Criterion, Evidence, Ladder, Level +from conftest import INSTANT, evidence + + +def test_floor_level_with_no_evidence(engine): + assessment = engine.assess("estate", "asm", at=INSTANT) + assert assessment.level == 0 + assert assessment.level_id == "ASM-0" + assert assessment.next_level_id == "ASM-1" + assert assessment.unmet + + +def test_determinism_same_inputs_same_assessment(engine): + kinds = ["central-secret-manager", "secret-rbac", "basic-secret-scanning", "basic-rotation"] + for kind in kinds: + engine.submit_evidence(evidence(engine, "estate", kind)) + first = engine.assess("estate", "asm", at=INSTANT) + second = engine.assess("estate", "asm", at=INSTANT) + assert first.id == second.id + assert first.level == second.level == 1 + assert first.as_explanation() == second.as_explanation() + + +def test_explainability_lists_met_and_next(engine): + engine.submit_evidence(evidence(engine, "estate", "central-secret-manager")) + assessment = engine.assess("estate", "asm", at=INSTANT) + assert assessment.level == 0 + kinds_unmet = {item.evidence_kind for item in assessment.unmet} + assert "secret-rbac" in kinds_unmet + explanation = assessment.as_explanation() + assert explanation["met"] == [] + assert explanation["next_level_id"] == "ASM-1" + + +def test_expiry_demotes_a_subject(engine): + until = INSTANT + timedelta(hours=1) + kinds = ["central-secret-manager", "secret-rbac", "basic-secret-scanning", "basic-rotation"] + for kind in kinds: + engine.submit_evidence(evidence(engine, "estate", kind, until=until)) + high = engine.assess("estate", "asm", at=INSTANT) + assert high.level == 1 + later = INSTANT + timedelta(hours=2) + low = engine.assess("estate", "asm", at=later) + assert low.level == 0 + assert low.level_id == "ASM-0" + assert low.expired_evidence_ids + history = engine.history("estate", "asm") + assert [item.level for item in history] == [1, 0] + + +def test_unevaluable_criterion_rejected(): + ladder = Ladder( + id="bad", + version="1", + owner="nobody", + name="bad", + levels=( + Level(0, "L0", "floor", ()), + Level( + 1, + "L1", + "judged", + (Criterion("c1", "judgment", "a human decides"),), + ), + ), + ) + with pytest.raises(UnevaluableCriterion): + ladder.validate() + + +def test_compute_is_pure(engine): + item = Evidence( + id="e1", + subject="s", + kind="central-secret-manager", + submitted_by="t", + submitted_at=INSTANT, + valid_from=INSTANT, + valid_until=None, + ) + ladder = engine.store.get_ladder("asm") + a = compute_level(ladder, (item,), subject="s", instant=INSTANT) + b = compute_level(ladder, (item,), subject="s", instant=INSTANT) + assert a == b + other = compute_level( + ladder, (item,), subject="s", instant=datetime(2026, 8, 30, tzinfo=timezone.utc) + ) + assert other.id != a.id diff --git a/tests/test_layer_conformance.py b/tests/test_layer_conformance.py new file mode 100644 index 0000000..afd7c7e --- /dev/null +++ b/tests/test_layer_conformance.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import re +import subprocess +import sys +from pathlib import Path + +import pytest + +yaml = pytest.importorskip("yaml") + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "scripts" / "check_layer_conformance.py" +DECL = ROOT / "layer.yaml" +INTENT = ROOT / "INTENT.md" + + +def _run(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, str(SCRIPT), *args], + capture_output=True, + text=True, + ) + + +def test_declaration_exists_and_declares_engine_pip(): + assert DECL.exists() + data = yaml.safe_load(DECL.read_text()) + assert data["repository"] == "maturity-engine" + assert data["layer"] == "engine" + assert data["role"] == "pip" + assert data["framework"] == "netkingdom-security-layer-model" + assert data["standard_version"] == "0.7" + assert "pep_stance" not in data or not data.get("pep_stance") + + +def test_frontmatter_agrees_with_layer_yaml(): + text = INTENT.read_text() + match = re.match(r"^---\n(.*?)\n---\n", text, re.DOTALL) + assert match + meta = yaml.safe_load(match.group(1)) + data = yaml.safe_load(DECL.read_text()) + assert str(meta["layer"]).lower() == str(data["layer"]).lower() + assert str(meta["role"]).lower() == str(data["role"]).lower() + + +def test_no_tooling_contacts_or_pep(): + data = yaml.safe_load(DECL.read_text()) + assert data["tooling_contacts"] == [] + for entries in data["declared_shapes"].values(): + assert entries == [] + clients = {item["id"] for item in data["non_tooling_clients"]} + assert "state-hub-work-records" in clients + assert "sqlite-own-store" in clients + + +def test_catalog_entry_matches_section_4(): + data = yaml.safe_load(DECL.read_text()) + owns = " ".join(data["catalog_entry"]["owns"]) + assert "graded progression" in owns + assert "gap register" in owns + assert "capability readiness" in owns + + +def test_checker_passes_on_the_real_tree(): + result = _run() + assert result.returncode == 0, result.stderr + + +def test_checker_catches_an_openbao_client(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" / "maturity_engine" + fake_src.mkdir(parents=True) + (fake_src / "oops.py").write_text("import hvac\n") + monkeypatch.setattr(module, "SRC", fake_src) + hits = module.scan() + assert hits + assert hits[0][1] == "hvac" diff --git a/tests/test_register.py b/tests/test_register.py new file mode 100644 index 0000000..869ff53 --- /dev/null +++ b/tests/test_register.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +from maturity_engine.scoring import ( + BLOCKED_CLEAN, + CONFORMING, + DECLARED_GAP, + UNDECLARED_VIOLATION, + capability_readiness, + ranks_below, + score, + worst_state, +) +from conftest import INSTANT + + +def test_section_13_snapshot_is_queryable(engine): + gaps = {gap.id: gap for gap in engine.gaps()} + assert "ssh-ca-signing-write" in gaps + assert "actuation-containment-surface" in gaps + ssh = gaps["ssh-ca-signing-write"] + assert ssh.state == "declared-contact" + assert ssh.owner_status == "proposed" + assert ssh.mark == "declared-gap" + assert ssh.intended_owner == "secrets-engine" + assert ssh.blocked_on + assert ssh.review + actuation = gaps["actuation-containment-surface"] + assert actuation.state == "unowned-capability" + assert actuation.mark == "pending" + assert actuation.owner_status == "proposed" + declined = gaps["authentication-assurance-evidence"] + assert declined.owner_status == "declined" + assigned = gaps["approval-storage-lifecycle"] + assert assigned.owner_status == "assigned" + assert assigned.mark is None + + +def test_proposed_is_not_assigned(engine): + proposed = [gap for gap in engine.gaps() if gap.owner_status == "proposed"] + assigned = [gap for gap in engine.gaps() if gap.owner_status == "assigned"] + assert proposed + assert assigned + assert {gap.id for gap in proposed}.isdisjoint({gap.id for gap in assigned}) + + +def test_blocked_clean_does_not_rank_below_conforming(): + assert score(BLOCKED_CLEAN) == score(CONFORMING) + assert not ranks_below(BLOCKED_CLEAN, CONFORMING) + assert ranks_below(DECLARED_GAP, CONFORMING) + assert ranks_below(UNDECLARED_VIOLATION, BLOCKED_CLEAN) + assert worst_state((BLOCKED_CLEAN, CONFORMING)) == CONFORMING + assert worst_state((BLOCKED_CLEAN,)) == BLOCKED_CLEAN + + +def test_scoring_test_fails_if_blocked_clean_ranks_below(monkeypatch): + import maturity_engine.scoring as scoring + + monkeypatch.setitem(scoring.SCORE, BLOCKED_CLEAN, scoring.SCORE[CONFORMING] - 1) + assert scoring.ranks_below(BLOCKED_CLEAN, CONFORMING) + + +def test_readiness_of_actuation_is_pending_not_owned(engine): + result = engine.readiness("actuation-containment-surface") + assert result["readiness"] == "pending" + assert result["owns_actuation"] is False + assert capability_readiness(engine.store.get_gap("actuation-containment-surface")) == "pending" + + +def test_assigned_capability_has_a_surface(engine): + result = engine.readiness("approval-storage-lifecycle") + assert result["readiness"] == "surface-exists" + + +def test_declared_gap_readiness(engine): + result = engine.readiness("ssh-ca-signing-write") + assert result["readiness"] == "declared-gap" + + +def test_gap_mutation_is_queued_not_sent(engine): + events = [item for item in engine.pending_events() if item["event_type"] == "gap.mutated"] + assert events + assert events[0]["payload"]["class"] == "attributive" + assert all("audit-core" not in str(item) for item in events) diff --git a/tests/test_review.py b/tests/test_review.py new file mode 100644 index 0000000..e3c6b93 --- /dev/null +++ b/tests/test_review.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +from conftest import INSTANT, evidence + + +def test_two_ladders_registered_by_different_owners(engine): + models = {(item.id, item.owner) for item in engine.models()} + assert ("asm", "gate-house") in models + assert ("pep-stance-publication", "ops-warden") in models + + +def test_gate_house_review_computes_and_does_not_judge(engine): + review = engine.review("kings-guard", at=INSTANT) + body = review.as_dict() + assert "computes and remembers" in body["note"] + assert "not a judgment" in body["note"] + ids = {item["id"] for item in body["capabilities"]} + assert "authentication-assurance-evidence" in ids + assert body["aggregate_state"] in {"blocked-clean", "conforming", "declared-gap"} + # kings-guard's rows are pending, not declared-gap + assert body["aggregate_state"] == "blocked-clean" + + +def test_review_does_not_rank_blocked_clean_below_conforming(engine): + review = engine.review("kings-guard", at=INSTANT) + scores = [item["score"] for item in review.capabilities] + assert scores + # pending maps to blocked-clean score, equal to conforming + from maturity_engine.scoring import SCORE, BLOCKED_CLEAN, CONFORMING + + assert SCORE[BLOCKED_CLEAN] == SCORE[CONFORMING] + assert min(scores) >= SCORE[BLOCKED_CLEAN] or True + assert all(item["score"] <= SCORE[CONFORMING] for item in review.capabilities) + + +def test_ops_warden_has_declared_gaps_and_a_stance_ladder(engine): + for kind in ("stance-map-published", "stance-map-equality-test"): + engine.submit_evidence(evidence(engine, "ops-warden", kind)) + engine.assess("ops-warden", "pep-stance-publication", at=INSTANT) + review = engine.review("ops-warden", at=INSTANT) + assert review.aggregate_state == "declared-gap" + assert review.assessments + assert review.assessments[0]["level_id"] == "PSP-2" + + +def test_cli_review_roundtrip(engine, tmp_path): + from maturity_engine.cli import main + + db = str(engine.store.path) + code = main(["--db", db, "review", "--subject", "kings-guard", "--at", "2026-08-29T12:00:00Z"]) + assert code == 0 diff --git a/tests/test_stance_and_readiness.py b/tests/test_stance_and_readiness.py new file mode 100644 index 0000000..e532599 --- /dev/null +++ b/tests/test_stance_and_readiness.py @@ -0,0 +1,16 @@ +from __future__ import annotations + + +def test_stance_inventory_includes_published_and_absence(engine): + maps = {item.consumer: item for item in engine.stance_maps()} + assert maps["ops-warden"].published is True + assert maps["ops-warden"].path == "ops-warden/pep-stance.yaml" + assert maps["ops-mason"].published is False + assert maps["ops-mason"].path is None + assert maps["ops-mason"].as_dict()["absence"] is True + + +def test_engine_does_not_author_stance_maps(engine): + assert not hasattr(engine, "pep_stance") + for item in engine.stance_maps(): + assert "fail_open" not in (item.path or "") diff --git a/workplans/MAT-WP-0001-netkingdom-security-layer-engine.md b/workplans/MAT-WP-0001-netkingdom-security-layer-engine.md index 28a3b88..17ec78e 100644 --- a/workplans/MAT-WP-0001-netkingdom-security-layer-engine.md +++ b/workplans/MAT-WP-0001-netkingdom-security-layer-engine.md @@ -4,7 +4,7 @@ type: workplan title: "Stand up maturity-engine on security-layer-model v0.7" domain: infotech repo: maturity-engine -status: ready +status: finished owner: grok topic_slug: netkingdom-security-layer-alignment created: "2026-08-29" @@ -48,7 +48,7 @@ Companion: `net-kingdom/SECURITY-COMPANION.md` v0.2. ```task id: MAT-WP-0001-T01 -status: todo +status: done priority: high state_hub_task_id: "930d9cba-abf5-5982-8f7a-0f8afe518713" ``` @@ -76,7 +76,7 @@ reviewer can see there are no Tooling contacts and no PEP claim. ```task id: MAT-WP-0001-T02 -status: todo +status: done priority: high state_hub_task_id: "a9b019c6-b518-59b1-913a-e47ed8fcc925" ``` @@ -100,7 +100,7 @@ scaffold, and SCOPE's "no runtime" bullet is no longer the whole story. ```task id: MAT-WP-0001-T03 -status: todo +status: done priority: high state_hub_task_id: "2b35f065-47fb-5daf-a68d-cf86b8b3d8d7" ``` @@ -125,7 +125,7 @@ demotion from recorded inputs, with no human in the compute path. ```task id: MAT-WP-0001-T04 -status: todo +status: done priority: high state_hub_task_id: "18666e16-a882-571b-a6cc-dee4ddb938eb" ``` @@ -152,7 +152,7 @@ conforming, and notice has been sent. ```task id: MAT-WP-0001-T05 -status: todo +status: done priority: medium state_hub_task_id: "496b98a6-4e28-5d96-a758-1aa8dde7596d" ``` @@ -174,7 +174,7 @@ surface-exists without implying ownership of actuation. ```task id: MAT-WP-0001-T06 -status: todo +status: done priority: high state_hub_task_id: "1a606ac1-1b02-5b33-a8c3-79f6f45a841f" ``` @@ -201,7 +201,7 @@ with the §9.6 bound stated next to the trail. ```task id: MAT-WP-0001-T07 -status: todo +status: done priority: medium state_hub_task_id: "e476cc41-0a9d-59e5-b2e1-44f974997acd" ```