diff --git a/Makefile b/Makefile index d862c20..fa9bc65 100644 --- a/Makefile +++ b/Makefile @@ -1,8 +1,8 @@ -.PHONY: test test-unit test-scenarios test-integration test-conformance +.PHONY: test test-unit test-scenarios test-integration test-conformance check-layer PYTHON ?= python3 -test: test-unit +test: test-unit check-layer test-unit: PYTHONPATH=src $(PYTHON) -m unittest discover -s tests -p 'test_*.py' @@ -11,4 +11,7 @@ test-scenarios: test-unit test-integration: test-unit -test-conformance: test-unit +test-conformance: test-unit check-layer + +check-layer: + PYTHONPATH=src $(PYTHON) scripts/check_layer_conformance.py --report diff --git a/SCOPE.md b/SCOPE.md index 29cd8bd..7abe43c 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -127,10 +127,11 @@ self-service and admin portal, public-registration orchestration, and flex-auth caller identity (live A2 on `flex-auth-user-engine`) are in the repo and, where applicable, on Railiance. -`USER-WP-0024` is the active workplan: layer declaration is in `INTENT.md`; -machine-readable conformance, the published PEP stance map, decision -lifetime, evidence cadence, and the access-control-fact claim contract -are not yet shipped. +`USER-WP-0024` ships the layer-conformance follow-through: `layer.yaml` +and `pep-stance.yaml`, fail-closed stance recording without a minted +decision id, request-bound allow lifetime, local-authorization +confinement, evidence classification and heartbeat, and the +access-control-fact claim contract. Still operator-owned, not remaining product scope: @@ -142,18 +143,19 @@ Still operator-owned, not remaining product scope: - `policy.enabled` and tenant-engine caller `enforce` belong to flex-auth / tenant-engine. -Layer-model residue that is remaining product scope, tracked in -`USER-WP-0024` and assessed in +Layer-model residue that `USER-WP-0024` closed, assessed in `history/2026-08-29-security-layer-scope-intent-assessment.md`: -- no `layer.yaml` and no conformance check; -- unreachable-engine stance is fail-closed in - `FlexAuthHTTPAdapter` but unpublished, untested against a map, and - recorded as a locally minted `decision_id`; -- `AuthorizationDecision` carries no lifetime; -- denials have no declared emission cadence or heartbeat; -- `LocalAuthorizationCheckPort` defaults to allow and is a second - decision point if a production runtime ever constructed it. +- `layer.yaml` plus `scripts/check_layer_conformance.py`; +- published `pep-stance.yaml`, fail-closed total, tested equal to + `FlexAuthHTTPAdapter`; engine-unavailable DENY records stance + application and carries no `decision_id`; +- `AuthorizationDecision` allows are request-bound with a 30s lifetime; +- denials and revocations are load-bearing; cadence is + `record_evidence_heartbeat()`; +- `LocalAuthorizationCheckPort` cannot be constructed when + `USER_ENGINE_FLEX_AUTH_TOKEN_FILE` is set, and `runtime.py` does not + import it. ## Against INTENT.md @@ -169,12 +171,12 @@ and PEP obligations. | Users, links, memberships, catalogs, projections, events | Met. | | NetKingdom identity-domain integration layer | Met for the owned slice. Consumes KeyCape, flex-auth, tenant-engine, identity-provisioner, audit-core, email-connect. | | Applications answer who / which scopes / what to project | Met via `/me`, identity context, catalogs, and projections. | -| Engine / PIP declaration in own voice | Met in `INTENT.md` as of 2026-08-29. Closes the declaration half of `USER-IN-0001`. | -| Subject context is a claim, never a decision | Held in production path (`FlexAuthHTTPAdapter`). Not yet proven for hats/access-control facts under §6.1, and the local double can still decide. | -| PEP-shaped: no side effect without a decision or recorded stance | Partial. Production asks flex-auth and fail-closes on errors. Stance is unpublished; unavailable DENY mints a local `decision_id`. | -| Published unreachable-engine stance map, tested equal to shipped behaviour | Missing. | -| Every allow has a lifetime | Missing on `AuthorizationDecision`. | -| Evidence bound: no completeness claim; cadence for load-bearing events | Partial. Mutations enqueue audit+outbox in the same store transaction. Completeness is not claimed in code; docs do not classify load-bearing vs attributive or declare a cadence. | +| Engine / PIP declaration in own voice | Met in `INTENT.md` and `layer.yaml`. Closes `USER-IN-0001`. | +| Subject context is a claim, never a decision | Held. Production asks flex-auth. Hats and access-control facts have no effect field; tests forbid compiling them into allow/deny. The local double is confined to tests/standalone. | +| PEP-shaped: no side effect without a decision or recorded stance | Met. Unavailable DENY records `stance_applied=fail_closed` and no `decision_id`. | +| Published unreachable-engine stance map, tested equal to shipped behaviour | Met (`pep-stance.yaml`). Gate-house still owes the §13.1 inventory row. | +| Every allow has a lifetime | Met. Production allows are request-bound, 30s. Standing grants are denied. | +| Evidence bound: no completeness claim; cadence for load-bearing events | Met. Classification and heartbeat in `docs/evidence-classification.md` and `UserEngineService.record_evidence_heartbeat()`. | | Not an IdP, PDP, secret store, directory, or org authority | Held. | | Optional UI, not UI-driven | Held, with a narrower reading: an optional portal now lives *in this repo* over the same APIs. INTENT's "not a UI application" still applies to product identity. | | Canon-aligned mappings without taking IAM as SoT | Met (`USER-WP-0007`, interface card). Access-review, policy, and control remain references, not owned records. | @@ -191,6 +193,5 @@ Still aspirational, and deliberately not started here: - observing production, or actuating containment — estate-wide zeros, not this repo's gaps to close. -Those remain INTENT, not a hole in SCOPE, except the layer-conformance -rows above, which are now SCOPE because INTENT has adopted the accepted -standard. +Those remain INTENT, not a hole in SCOPE. Layer-conformance rows above +are met by `USER-WP-0024`; the §13.1 inventory row is gate-house's. diff --git a/docs/evidence-classification.md b/docs/evidence-classification.md new file mode 100644 index 0000000..a064e2f --- /dev/null +++ b/docs/evidence-classification.md @@ -0,0 +1,46 @@ +# Local evidence classification + +Status: implemented +Updated: 2026-08-29 +Standard: `net-kingdom/canon/standards/security-layer-model_v0.7.md` §9.6 +Companion: `net-kingdom/SECURITY-COMPANION.md` §6 + +Local audit records and the durable outbox prove the records they hold +were **not altered or truncated after arrival**. They do not prove that +an event never sent did not happen. + +Sound sentences: + +- the archive proves the records it holds were not altered or truncated after arrival +- absence of a record is not evidence of non-occurrence + +Unsound sentences, not used here: + +- the audit record proves it happened +- there is no record, so it did not happen +- the trail is complete + +## Classification + +| Kind | Test | Local form | +| --- | --- | --- | +| **Load-bearing** | a control's soundness depends on the event being present | authorization denials; `account.status_changed`; `tenant_account.status_changed`; `prepared_account.revoked`; `family_invitation.revoked` | +| **Attributive** | forensic reconstruction; no control branches on presence | user/profile/membership-added/catalog/registration/hat events | +| **Heartbeat** | positive claim that can itself go missing | `user_engine.evidence.heartbeat` | + +Successful mutations enqueue local audit and outbox in the same store +transaction. That prevents accidental omission (crash between save and +emit) of the *local* record. Drain to `audit-core` is after commit. + +## Cadence + +Rate monitoring cannot detect suppression of rare denials and +revocations. The required form is a **heartbeat**: +`UserEngineService.record_evidence_heartbeat()` emits a positive claim +with counts of load-bearing classes already on the local trail. The +payload sets `completeness_claimed: false`. Expected interval: one hour +when a drain loop is running. The heartbeat is the claim that can go +missing; it is not a completeness proof. + +Adversarial omission at a compromised source is a known residual. +Nothing in this model prevents it. diff --git a/docs/evidence-gap-examples.md b/docs/evidence-gap-examples.md index 4f2450a..7e1abd5 100644 --- a/docs/evidence-gap-examples.md +++ b/docs/evidence-gap-examples.md @@ -1,12 +1,17 @@ # Evidence Gap Examples Status: candidate -Updated: 2026-06-05 +Updated: 2026-08-29 `user-engine` should not pretend missing review or governance material exists. When identity-domain context lacks evidence, policy, control, review, or task references, the gap must be explicit and handoff-ready. +Local audit and the durable outbox prove the records they hold were not +altered or truncated after arrival. Absence of a record is not evidence +of non-occurrence. Classification and the denial/revocation heartbeat +are in `docs/evidence-classification.md`. Completeness is not claimed. + ## Gap Shape ```yaml diff --git a/docs/hats-realms-services-assets-access-profiles.md b/docs/hats-realms-services-assets-access-profiles.md index 392934f..816e07a 100644 --- a/docs/hats-realms-services-assets-access-profiles.md +++ b/docs/hats-realms-services-assets-access-profiles.md @@ -87,8 +87,14 @@ contexts are omitted from projections for other applications. ## Export Boundary `export_access_control_facts` returns an adapter-neutral manifest plus facts. -External authorization engines or ACL systems can consume these facts, but -they remain responsible for final policy decisions and runtime enforcement. +Hats, access profiles, and `AccessControlFact` values are PIP **claims**. +They are not an authorization decision. Compiling them into a local +allow or deny before `access-engine` runs is an early decision +(security-layer-model §6.1) and is forbidden. + +`select_active_hat` and `export_access_control_facts` never return an +effect. `access-engine` remains the only decision point; consumers that +need an allow or deny must ask it. ## Redaction And Diagnostics diff --git a/docs/operability.md b/docs/operability.md index 405d6ad..f1a67f4 100644 --- a/docs/operability.md +++ b/docs/operability.md @@ -4,9 +4,12 @@ Use `readiness()` for dependency checks and `operability_snapshot()` for runtime counters and invariant checks. The snapshot currently reports store -readiness, audit correlation completeness, outbox diagnostic availability, and -counts for users, accounts, tenant accounts, memberships, applications, -catalogs, profile values, audit records, and pending outbox events. +readiness, whether written audit records carry a correlation id, outbox +diagnostic availability, and counts for users, accounts, tenant accounts, +memberships, applications, catalogs, profile values, audit records, and +pending outbox events. Those checks describe the records in hand. They +do not prove that every event that should have been emitted was emitted. +See `docs/evidence-classification.md`. ## Structured Logs diff --git a/layer.yaml b/layer.yaml new file mode 100644 index 0000000..531f305 --- /dev/null +++ b/layer.yaml @@ -0,0 +1,79 @@ +# user-engine — NetKingdom security layer declaration +# +# Framework: net-kingdom/canon/standards/security-layer-model_v0.7.md +# Companion: net-kingdom/SECURITY-COMPANION.md v0.2 +# Declare: INTENT.md (own voice) + this file (§11 machine-readable form) +# Validate: python3 scripts/check_layer_conformance.py +# +# Engine/PIP: same authoritative user-domain inputs yield the same result. +# Catalogued Tooling is key-cape and OpenBao. This repository holds no client +# for either. PostgreSQL is the modeled-concept store, not a §4 Tooling row. + +schema_version: "0.1" +framework: netkingdom-security-layer-model +standard_version: "0.7" +repository: user-engine +layer: engine +role: pip +declared_by: INTENT.md +declared_at: "2026-08-29" + +pep_shape: true +pep_stance: pep-stance.yaml + +catalog_entry: + owns: + - users + - accounts + - memberships + role: PIP + +# Empty is a claim. scripts/check_layer_conformance.py fails the commit that +# introduces an OpenBao, Vault, LDAP, or cluster client. +tooling_contacts: [] + +own_store: + - id: postgres-modeled-concept + module: src/user_engine/adapters/postgres.py + import_root: psycopg + symbol: PostgresUserEngineStore + store: PostgreSQL + write: true + note: Persistence of the modeled concept a PIP owns. Not catalogued Tooling. + +non_tooling_clients: + - id: flex-auth-engine-api + target: access-engine + layer: engine + module: src/user_engine/adapters/flex_auth.py + operation: "HTTP POST /v1/check" + write: false + note: Engine API. The shape §5 prescribes for a PEP-shaped consumer. + + - id: oidc-claim-input + target: key-cape + layer: tooling-as-claim-input + module: src/user_engine/oidc.py + operation: "OIDC discovery and JWKS for verified IAM Profile claims" + write: false + note: Consumes identity claims as PIP input. Not a key-cape admin client. + + - id: env-injected-secrets + target: Railiance secret injection + layer: not-catalogued + module: src/user_engine/runtime.py + operation: "read environment and projected token files" + write: false + note: No OpenBao client. SecretProvider remains an unused port. + + - 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: Agents writing this repository, not the service runtime. + +declared_shapes: + "5.1": [] + "5.2": [] + "5.3": [] diff --git a/pep-stance.yaml b/pep-stance.yaml new file mode 100644 index 0000000..9b2c4c7 --- /dev/null +++ b/pep-stance.yaml @@ -0,0 +1,53 @@ +# user-engine — PEP unreachable-engine stance map +# +# Framework: net-kingdom/canon/standards/security-layer-model_v0.7.md §6.4, §9.3 +# Companion: net-kingdom/SECURITY-COMPANION.md §5 +# Validate: tests/test_layer_conformance.py +# +# user-engine is a PEP-shaped Engine (PIP). Protected mutations proceed only +# with a decision from access-engine, or under this published stance whose +# application is recorded in place of a decision. +# +# Total by construction: every zone in security-zones_v0.1, plus unknown and +# not-applicable. No implicit default. Production is fail-closed everywhere, +# including unknown. + +schema_version: "0.1" +framework: netkingdom-security-layer-model +standard_version: "0.7" +repository: user-engine +pep_shape: true +declared_by: INTENT.md + +protected_action: "user-engine owned mutations (account, membership, registration, hat, catalog, profile)" +decision_engine: access-engine +scope: security-zone + +stance: + z0-experimental: fail_closed + z1-operational: fail_closed + z2-protected: fail_closed + z2-continuity: fail_closed + z3-critical: fail_closed + unknown: fail_closed + not-applicable: fail_closed + +allow_binding: request +allow_lifetime_seconds: 30 + +on_apply: + recorded_fields: + - stance_applied + - stance_scope + - outcome + - decision_id + decision_id: present only where access-engine rendered a decision + written_to: + - "local audit_records (UserEngineService._authorize)" + never_recorded: "caller tokens, secret material, profile payloads" + +verdict_caching: none + +reconstructability: + bound: "§9.6 — local audit and outbox prove the records they hold" + completeness_claimed: false diff --git a/scripts/check_layer_conformance.py b/scripts/check_layer_conformance.py new file mode 100644 index 0000000..f7750db --- /dev/null +++ b/scripts/check_layer_conformance.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""Check user-engine against the NetKingdom security layer model (§5, §11). + +Read-only. user-engine is Engine/PIP and holds no catalogued Tooling client. +PostgreSQL is the modeled-concept store, declared under own_store. + +The failure this exists to catch is a convenience: an OpenBao, Vault, LDAP, +or cluster client arriving as one import. That is an undeclared violation. + +Exit 0 clean, 1 undeclared contact found, 2 declaration malformed. +""" +from __future__ import annotations + +import argparse +import ast +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" / "user_engine" +DECL = ROOT / "layer.yaml" +sys.path.insert(0, str(ROOT / "src")) + +from user_engine.layer_yaml import load_mapping # noqa: E402 + +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)", + "docker": "container runtime client", + "redis": "direct datastore connection", +} + +OWN_STORE_IMPORTS = { + "psycopg": "PostgreSQL modeled-concept store", + "psycopg2": "PostgreSQL modeled-concept store", + "asyncpg": "PostgreSQL modeled-concept store", + "sqlalchemy": "database client", + "pymysql": "database client", +} + + +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 = load_mapping(DECL) + except (ValueError, OSError) 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 data["layer"] != "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["tooling_contacts"] not in ([], None): + print( + "FAIL: tooling_contacts must be empty; catalogued Tooling clients " + "are undeclared violations for this Engine", + file=sys.stderr, + ) + raise SystemExit(2) + return data + + +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 scan_own_store() -> 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 OWN_STORE_IMPORTS: + hits.append((path, module, OWN_STORE_IMPORTS[module])) + return hits + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--report", action="store_true", help="print the declaration summary") + args = parser.parse_args() + + decl = load_declaration() + hits = scan() + own_store_hits = scan_own_store() + own_store = decl.get("own_store") or [] + + if args.report: + print( + f"user-engine — layer {decl['layer']}/{decl['role']}, " + f"standard v{decl['standard_version']}" + ) + print(" tooling contacts declared: 0") + print(f" own-store declarations: {len(own_store)}") + print(f" own-store imports: {len(own_store_hits)}") + print(f" pep stance: {decl.get('pep_stance')}") + + if hits: + print("", file=sys.stderr) + print( + "FAIL: undeclared Tooling-layer client (§11 undeclared violation)", + 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 own_store_hits and not own_store: + print( + "FAIL: modeled-concept store import with no own_store declaration", + file=sys.stderr, + ) + for path, module, what in own_store_hits: + print(f" {path.relative_to(ROOT)}: imports {module!r} — {what}", file=sys.stderr) + return 1 + + if not args.report: + print( + f"OK: no catalogued Tooling client in {SRC.relative_to(ROOT)} " + "(Engine/PIP, §11)" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/user_engine/adapters/flex_auth.py b/src/user_engine/adapters/flex_auth.py index fef320b..d3602b8 100644 --- a/src/user_engine/adapters/flex_auth.py +++ b/src/user_engine/adapters/flex_auth.py @@ -11,6 +11,13 @@ from user_engine.domain import ( AuthorizationDecision, AuthorizationEffect, AuthorizationRequest, + utc_now, +) +from user_engine.pep_stance import ( + ALLOW_BINDING, + ALLOW_LIFETIME, + DEFAULT_STANCE_SCOPE, + UNREACHABLE_STANCE, ) @@ -86,16 +93,28 @@ class FlexAuthHTTPAdapter: for item in body.get("obligations", ()) if isinstance(item, dict) and item.get("type") ) + lifetime = ALLOW_LIFETIME if effect in { + AuthorizationEffect.ALLOW, + AuthorizationEffect.AUDIT_ONLY, + } else None + issued_at = utc_now() if lifetime is not None else None return AuthorizationDecision( effect=effect, decision_id=decision_id, reason=reason, obligations=obligations, + binding=ALLOW_BINDING, + lifetime=lifetime, + issued_at=issued_at, ) except (HTTPError, URLError, TimeoutError, OSError, ValueError, KeyError, TypeError): return AuthorizationDecision( effect=AuthorizationEffect.DENY, + decision_id=None, reason="authorization service unavailable", + stance_applied=UNREACHABLE_STANCE, + stance_scope=DEFAULT_STANCE_SCOPE, + binding=ALLOW_BINDING, ) def batch_check( diff --git a/src/user_engine/adapters/local.py b/src/user_engine/adapters/local.py index 33428a4..9b584ae 100644 --- a/src/user_engine/adapters/local.py +++ b/src/user_engine/adapters/local.py @@ -3,6 +3,7 @@ from __future__ import annotations import copy +import os from contextlib import contextmanager from dataclasses import dataclass, field from typing import Iterable, Iterator, Mapping, cast @@ -426,18 +427,25 @@ class InMemoryUserEngineStore: class LocalAuthorizationCheckPort: - """Deterministic local authorization adapter. + """Test and standalone authorization double. Not a production PDP. - Rules are action-specific. The default is allow so isolated tests and local - demos can focus on user-engine behavior while still exercising the port. + Construction fails when a production flex-auth caller token is configured, + so ``create_application()`` cannot assemble this adapter by mistake. """ + standalone_double = True + def __init__( self, *, default_effect: AuthorizationEffect = AuthorizationEffect.ALLOW, action_effects: dict[str, AuthorizationEffect] | None = None, ) -> None: + if os.environ.get("USER_ENGINE_FLEX_AUTH_TOKEN_FILE"): + raise RuntimeError( + "LocalAuthorizationCheckPort cannot be constructed in a " + "production runtime (USER_ENGINE_FLEX_AUTH_TOKEN_FILE is set)" + ) self.default_effect = default_effect self.action_effects = action_effects or {} self.requests: list[AuthorizationRequest] = [] @@ -445,7 +453,7 @@ class LocalAuthorizationCheckPort: def check(self, request: AuthorizationRequest) -> AuthorizationDecision: self.requests.append(request) effect = self.action_effects.get(request.action, self.default_effect) - return AuthorizationDecision(effect=effect, reason="local") + return AuthorizationDecision.for_standalone(effect, reason="local") def batch_check( self, requests: Iterable[AuthorizationRequest] diff --git a/src/user_engine/domain/models.py b/src/user_engine/domain/models.py index fe5ae60..803fcd4 100644 --- a/src/user_engine/domain/models.py +++ b/src/user_engine/domain/models.py @@ -8,7 +8,7 @@ these shapes instead of putting domain rules in infrastructure code. from __future__ import annotations from dataclasses import dataclass, field -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from enum import StrEnum from typing import Any, Mapping from uuid import uuid4 @@ -656,9 +656,14 @@ class AuthorizationRequest: @dataclass(frozen=True) class AuthorizationDecision: effect: AuthorizationEffect - decision_id: str = field(default_factory=lambda: new_id("dec")) + decision_id: str | None = None reason: str | None = None obligations: tuple[str, ...] = () + binding: str = "request" + lifetime: timedelta | None = None + issued_at: datetime | None = None + stance_applied: str | None = None + stance_scope: str | None = None @property def allowed(self) -> bool: @@ -667,6 +672,33 @@ class AuthorizationDecision: AuthorizationEffect.AUDIT_ONLY, } + def expired(self, now: datetime | None = None) -> bool: + if self.lifetime is None or self.issued_at is None: + return False + return (now or utc_now()) > self.issued_at + self.lifetime + + @classmethod + def for_standalone( + cls, + effect: AuthorizationEffect, + *, + reason: str, + obligations: tuple[str, ...] = (), + ) -> AuthorizationDecision: + allowed = effect in { + AuthorizationEffect.ALLOW, + AuthorizationEffect.AUDIT_ONLY, + } + issued = utc_now() if allowed else None + return cls( + effect=effect, + reason=reason, + obligations=obligations, + binding="standalone", + lifetime=timedelta(hours=1) if allowed else None, + issued_at=issued, + ) + @dataclass(frozen=True) class AuditRecord: diff --git a/src/user_engine/evidence.py b/src/user_engine/evidence.py new file mode 100644 index 0000000..48a4f7e --- /dev/null +++ b/src/user_engine/evidence.py @@ -0,0 +1,88 @@ +"""Local evidence classification under security-layer-model §9.6. + +An append-only local audit and outbox prove the records they hold were +not altered after arrival. They do not prove that an event never sent +did not happen. Completeness is not claimed. +""" + +from __future__ import annotations + +from datetime import timedelta +from typing import Iterable + +from user_engine.domain import AuditRecord, OutboxEvent + +HEARTBEAT_EVENT_TYPE = "user_engine.evidence.heartbeat" +HEARTBEAT_INTERVAL = timedelta(hours=1) + +# Low-volume classes whose absence can be mistaken for "it was allowed" +# or "the membership still holds". Cadence is a heartbeat, not a rate. +LOAD_BEARING_OUTBOX_TYPES = frozenset( + { + "account.status_changed", + "tenant_account.status_changed", + "prepared_account.revoked", + "family_invitation.revoked", + } +) + +LOAD_BEARING_AUDIT_SUMMARIES = frozenset( + { + "authorization denied", + "authorization denied (stance fail_closed)", + } +) + +ATTRIBUTIVE_OUTBOX_TYPES = frozenset( + { + "user.created", + "user.self_service_profile_updated", + "identity.linked", + "membership.added", + "profile.value_set", + "application.registered", + "application.bound", + "catalog.published", + "registration.started", + "registration.factor_verified", + "registration.completed", + "access_profile.registered", + "active_access_context.selected", + } +) + + +def classify_outbox(event_type: str) -> str: + if event_type == HEARTBEAT_EVENT_TYPE: + return "heartbeat" + if event_type in LOAD_BEARING_OUTBOX_TYPES: + return "load-bearing" + return "attributive" + + +def classify_audit(summary: str | None) -> str: + if summary in LOAD_BEARING_AUDIT_SUMMARIES or ( + summary is not None and summary.startswith("authorization denied") + ): + return "load-bearing" + return "attributive" + + +def load_bearing_counts( + audit_records: Iterable[AuditRecord], + outbox_events: Iterable[OutboxEvent], +) -> dict[str, int]: + counts: dict[str, int] = { + "authorization_denied": 0, + "account.status_changed": 0, + "tenant_account.status_changed": 0, + "prepared_account.revoked": 0, + "family_invitation.revoked": 0, + } + for record in audit_records: + if classify_audit(record.summary) == "load-bearing": + counts["authorization_denied"] += 1 + for event in outbox_events: + if event.event_type in counts: + counts[event.event_type] += 1 + return counts diff --git a/src/user_engine/layer_yaml.py b/src/user_engine/layer_yaml.py new file mode 100644 index 0000000..c85c0fc --- /dev/null +++ b/src/user_engine/layer_yaml.py @@ -0,0 +1,134 @@ +"""Minimal YAML mapping loader for layer and PEP-stance declarations. + +Stdlib only. Handles the subset this repository actually writes: nested +maps, lists of scalars, lists of maps, quoted strings, booleans, null, +integers, and empty lists. Not a general YAML implementation. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + + +def load_mapping(path: Path) -> dict[str, Any]: + data = load_mapping_text(path.read_text()) + if not isinstance(data, dict): + raise ValueError(f"{path} did not parse as a mapping") + return data + + +def load_mapping_text(text: str) -> dict[str, Any]: + data = _parse(text) + if not isinstance(data, dict): + raise ValueError("YAML text did not parse as a mapping") + return data + + +def _parse(text: str) -> Any: + lines: list[tuple[int, str]] = [] + for raw in text.splitlines(): + stripped = raw.split("#", 1)[0].rstrip() + if not stripped: + continue + indent = len(raw) - len(raw.lstrip(" ")) + lines.append((indent, stripped.lstrip(" "))) + value, _ = _parse_block(lines, 0, 0) + return value + + +def _parse_block( + lines: list[tuple[int, str]], index: int, indent: int +) -> tuple[Any, int]: + if index >= len(lines): + return {}, index + current_indent, content = lines[index] + if current_indent < indent: + return {}, index + if content.startswith("- "): + return _parse_list(lines, index, current_indent) + return _parse_map(lines, index, current_indent) + + +def _parse_map( + lines: list[tuple[int, str]], index: int, indent: int +) -> tuple[dict[str, Any], int]: + result: dict[str, Any] = {} + while index < len(lines): + current_indent, content = lines[index] + if current_indent < indent: + break + if current_indent > indent: + raise ValueError(f"unexpected indent at {content!r}") + if content.startswith("- "): + break + key, separator, remainder = content.partition(":") + if not separator: + raise ValueError(f"expected key: value, got {content!r}") + key = _parse_scalar(key.strip()) + if not isinstance(key, str): + key = str(key) + remainder = remainder.strip() + index += 1 + if remainder in ("", "|", ">"): + if index < len(lines) and lines[index][0] > indent: + value, index = _parse_block(lines, index, lines[index][0]) + else: + value = None + else: + value = _parse_scalar(remainder) + result[key] = value + return result, index + + +def _parse_list( + lines: list[tuple[int, str]], index: int, indent: int +) -> tuple[list[Any], int]: + result: list[Any] = [] + while index < len(lines): + current_indent, content = lines[index] + if current_indent < indent: + break + if current_indent > indent: + raise ValueError(f"unexpected indent at {content!r}") + if not content.startswith("- "): + break + item = content[2:].strip() + index += 1 + if not item: + if index < len(lines) and lines[index][0] > indent: + value, index = _parse_block(lines, index, lines[index][0]) + else: + value = None + result.append(value) + continue + if ":" in item and not item.startswith(("'", '"')): + key, _, remainder = item.partition(":") + mapping: dict[str, Any] = {key.strip(): _parse_scalar(remainder.strip())} + if index < len(lines) and lines[index][0] > indent: + nested, index = _parse_block(lines, index, lines[index][0]) + if isinstance(nested, dict): + mapping.update(nested) + result.append(mapping) + else: + result.append(_parse_scalar(item)) + return result, index + + +def _parse_scalar(text: str) -> Any: + if text in ("", "~", "null", "Null", "NULL"): + return None + if text in ("true", "True"): + return True + if text in ("false", "False"): + return False + if text == "[]": + return [] + if text == "{}": + return {} + if len(text) >= 2 and text[0] == text[-1] and text[0] in "'\"": + return text[1:-1] + try: + return int(text) + except ValueError: + return text diff --git a/src/user_engine/pep_stance.py b/src/user_engine/pep_stance.py new file mode 100644 index 0000000..aadf47a --- /dev/null +++ b/src/user_engine/pep_stance.py @@ -0,0 +1,22 @@ +"""Published unreachable-engine stance. Must equal pep-stance.yaml.""" + +from __future__ import annotations + +from datetime import timedelta + +UNREACHABLE_STANCE = "fail_closed" +ALLOW_BINDING = "request" +ALLOW_LIFETIME = timedelta(seconds=30) +VERDICT_CACHING = "none" + +STANCE = { + "z0-experimental": UNREACHABLE_STANCE, + "z1-operational": UNREACHABLE_STANCE, + "z2-protected": UNREACHABLE_STANCE, + "z2-continuity": UNREACHABLE_STANCE, + "z3-critical": UNREACHABLE_STANCE, + "unknown": UNREACHABLE_STANCE, + "not-applicable": UNREACHABLE_STANCE, +} + +DEFAULT_STANCE_SCOPE = "unknown" diff --git a/src/user_engine/service.py b/src/user_engine/service.py index 0feef6f..01e6379 100644 --- a/src/user_engine/service.py +++ b/src/user_engine/service.py @@ -19,6 +19,7 @@ from user_engine.domain import ( ApplicationBinding, AttributeDefinition, AuditRecord, + AuthorizationDecision, AuthorizationRequest, CanonEntityReference, CanonRelationshipReference, @@ -68,6 +69,10 @@ from user_engine.errors import ( NotFoundError, ValidationError, ) +from user_engine.evidence import ( + HEARTBEAT_EVENT_TYPE, + load_bearing_counts, +) from user_engine.ports import ( AuthorizationCheckPort, FactorVerificationAdapter, @@ -2894,6 +2899,37 @@ class UserEngineService: issues=issues, ) + def record_evidence_heartbeat( + self, *, correlation_id: str | None = None + ) -> OutboxEvent: + """Emit a positive evidence claim that can itself go missing. + + Counts load-bearing denials and revocations already on the local + trail. Does not claim the trail is complete. + """ + correlation_id = correlation_id or new_id("corr") + counts = load_bearing_counts( + self.store.audit_log(), self.store.pending_outbox() + ) + event = OutboxEvent( + event_id=new_id("evt"), + event_type=HEARTBEAT_EVENT_TYPE, + aggregate_id="user-engine:evidence", + payload={ + "completeness_claimed": False, + "form": "heartbeat", + "bound": ( + "the archive proves the records it holds were not " + "altered or truncated after arrival" + ), + "load_bearing": counts, + }, + tenant=PLATFORM_TENANT, + correlation_id=correlation_id, + ) + self.store.append_outbox(event) + return event + def structured_log_context( self, *, @@ -4242,22 +4278,75 @@ class UserEngineService: ), ) decision = self.authorization.check(request) - if not decision.allowed: - self.store.append_audit( - AuditRecord( - audit_id=new_id("aud"), - actor=actor, + if decision.allowed: + if decision.lifetime is None: + self._record_authorization_denial( + actor, action=action, - subject=resource_id, + resource_id=resource_id, tenant=tenant, correlation_id=correlation_id, - decision_id=decision.decision_id, application_id=application_id, - summary="authorization denied", + decision=AuthorizationDecision( + effect=decision.effect, + reason="authorization allow has no lifetime", + ), + summary="authorization denied (allow has no lifetime)", ) + raise AuthorizationDenied("authorization allow has no lifetime") + if decision.expired(): + self._record_authorization_denial( + actor, + action=action, + resource_id=resource_id, + tenant=tenant, + correlation_id=correlation_id, + application_id=application_id, + decision=decision, + summary="authorization denied (allow expired)", + ) + raise AuthorizationDenied("authorization allow expired") + return decision + summary = "authorization denied" + if decision.stance_applied: + summary = f"authorization denied (stance {decision.stance_applied})" + self._record_authorization_denial( + actor, + action=action, + resource_id=resource_id, + tenant=tenant, + correlation_id=correlation_id, + application_id=application_id, + decision=decision, + summary=summary, + ) + raise AuthorizationDenied(decision.reason or "authorization denied") + + def _record_authorization_denial( + self, + actor: Actor, + *, + action: str, + resource_id: str, + tenant: str, + correlation_id: str, + application_id: str | None, + decision: AuthorizationDecision, + summary: str, + ) -> None: + self.store.append_audit( + AuditRecord( + audit_id=new_id("aud"), + actor=actor, + action=action, + subject=resource_id, + tenant=tenant, + correlation_id=correlation_id, + decision_id=decision.decision_id, + application_id=application_id, + summary=summary, ) - raise AuthorizationDenied(decision.reason or "authorization denied") - return decision + ) def _record_mutation( self, diff --git a/src/user_engine/testing/fixtures.py b/src/user_engine/testing/fixtures.py index 3c70f65..195868a 100644 --- a/src/user_engine/testing/fixtures.py +++ b/src/user_engine/testing/fixtures.py @@ -61,7 +61,7 @@ class StaticAuthorizationCheckPort: def check(self, request: AuthorizationRequest) -> AuthorizationDecision: self.requests.append(request) - return AuthorizationDecision(effect=self.effect, reason="fixture") + return AuthorizationDecision.for_standalone(self.effect, reason="fixture") def batch_check( self, requests: Iterable[AuthorizationRequest] diff --git a/src/user_engine/testing/scenarios.py b/src/user_engine/testing/scenarios.py index 8f9234d..588a53d 100644 --- a/src/user_engine/testing/scenarios.py +++ b/src/user_engine/testing/scenarios.py @@ -125,8 +125,8 @@ class ScenarioAuthorizationHarness: effect = self.action_effects.get(request.action, self.default_effect) if _cross_tenant_denied(request) or _assurance_denied(request): effect = AuthorizationEffect.DENY - return AuthorizationDecision( - effect=effect, + return AuthorizationDecision.for_standalone( + effect, reason="scenario", obligations=self.action_obligations.get(request.action, ()), ) diff --git a/tests/test_evidence.py b/tests/test_evidence.py new file mode 100644 index 0000000..fca063d --- /dev/null +++ b/tests/test_evidence.py @@ -0,0 +1,230 @@ +import unittest +from dataclasses import fields +from datetime import timedelta + +from user_engine.adapters.local import InMemoryUserEngineStore, LocalAuthorizationCheckPort +from user_engine.domain import ( + AccessControlFact, + AccessMembershipRequirement, + AccessProfile, + ActiveAccessContext, + AccountStatus, + AuthorizationDecision, + AuthorizationEffect, +) +from user_engine.evidence import ( + HEARTBEAT_EVENT_TYPE, + classify_audit, + classify_outbox, + load_bearing_counts, +) +from user_engine.service import AccessControlFactExport, AccessProfileSelection, UserEngineService +from user_engine.testing.fixtures import ( + FixtureIdentityClaimsAdapter, + human_actor_claims, + sample_application, + sample_application_binding, + sample_catalog, +) + + +_FORBIDDEN_DECISION_FIELDS = frozenset( + {"effect", "allowed", "decision", "decision_id", "deny", "allow"} +) + + +class EvidenceClassificationTests(unittest.TestCase): + def test_denials_and_revocations_are_load_bearing(self): + self.assertEqual(classify_audit("authorization denied"), "load-bearing") + self.assertEqual( + classify_audit("authorization denied (stance fail_closed)"), + "load-bearing", + ) + self.assertEqual(classify_outbox("account.status_changed"), "load-bearing") + self.assertEqual(classify_outbox("prepared_account.revoked"), "load-bearing") + self.assertEqual(classify_outbox("user.created"), "attributive") + self.assertEqual(classify_outbox(HEARTBEAT_EVENT_TYPE), "heartbeat") + + def test_heartbeat_counts_load_bearing_events_and_does_not_claim_completeness(self): + service, _, _ = _service() + session = service.me(_claims(), correlation_id="corr-me") + service.set_account_status( + session.actor, + session.user.user_id, + AccountStatus.DISABLED, + correlation_id="corr-disable", + ) + deny = LocalAuthorizationCheckPort(default_effect=AuthorizationEffect.DENY) + denied = UserEngineService( + store=service.store, + identity_adapter=FixtureIdentityClaimsAdapter(), + authorization=deny, + ) + from user_engine.errors import AuthorizationDenied + + with self.assertRaises(AuthorizationDenied): + denied.create_user( + session.actor, + display_name="x", + primary_email=None, + correlation_id="corr-denied", + ) + + event = service.record_evidence_heartbeat(correlation_id="corr-heartbeat") + self.assertEqual(event.event_type, HEARTBEAT_EVENT_TYPE) + self.assertFalse(event.payload["completeness_claimed"]) + self.assertEqual(event.payload["form"], "heartbeat") + self.assertGreaterEqual(event.payload["load_bearing"]["account.status_changed"], 1) + self.assertGreaterEqual(event.payload["load_bearing"]["authorization_denied"], 1) + self.assertIn("were not altered or truncated after arrival", event.payload["bound"]) + self.assertNotIn("complete", event.payload["bound"]) + + counts = load_bearing_counts(service.store.audit_log(), service.store.pending_outbox()) + self.assertEqual(counts["account.status_changed"], 1) + + def test_allow_without_lifetime_is_denied(self): + from user_engine.errors import AuthorizationDenied + + class _StandingGrant: + def check(self, request): + return AuthorizationDecision(effect=AuthorizationEffect.ALLOW, reason="standing") + + store = InMemoryUserEngineStore() + service = UserEngineService( + store=store, + identity_adapter=FixtureIdentityClaimsAdapter(), + authorization=_StandingGrant(), + ) + with self.assertRaisesRegex(AuthorizationDenied, "no lifetime"): + service.me(_claims(), correlation_id="corr-standing") + self.assertEqual(store.audit_log()[-1].summary, "authorization denied (allow has no lifetime)") + self.assertIsNone(store.audit_log()[-1].decision_id) + + def test_expired_allow_is_denied(self): + from user_engine.errors import AuthorizationDenied + + class _StaleAllow: + def check(self, request): + decision = AuthorizationDecision.for_standalone( + AuthorizationEffect.ALLOW, reason="stale" + ) + return AuthorizationDecision( + effect=decision.effect, + reason=decision.reason, + binding=decision.binding, + lifetime=timedelta(seconds=1), + issued_at=decision.issued_at - timedelta(seconds=5), + ) + + store = InMemoryUserEngineStore() + service = UserEngineService( + store=store, + identity_adapter=FixtureIdentityClaimsAdapter(), + authorization=_StaleAllow(), + ) + with self.assertRaisesRegex(AuthorizationDenied, "expired"): + service.me(_claims(), correlation_id="corr-stale") + + +class AccessControlFactsAreClaimsTests(unittest.TestCase): + def test_claim_shapes_have_no_authorization_effect(self): + for model in (AccessControlFact, AccessProfile, ActiveAccessContext): + names = {item.name for item in fields(model)} + overlap = names & _FORBIDDEN_DECISION_FIELDS + self.assertFalse( + overlap, + f"{model.__name__} carries decision fields {overlap}", + ) + + def test_hat_selection_and_export_never_return_allow_or_deny(self): + service, _, _ = _service() + session = _bootstrap(service) + service.add_membership( + session.actor, + session.user.user_id, + tenant="tenant:coulomb", + scope_type="realm", + scope_id="realm:citadel", + kind="operator", + correlation_id="corr-realm-membership", + ) + profile = service.register_access_profile( + session.actor, + AccessProfile( + tenant="tenant:coulomb", + display_name="Operator", + hat="operator", + realm_id="realm:citadel", + membership_requirements=( + AccessMembershipRequirement( + scope_type="realm", + scope_id="realm:citadel", + kind="operator", + ), + ), + ), + correlation_id="corr-profile-register", + ) + selection = service.select_active_hat( + session.actor, + session.user.user_id, + profile.access_profile_id, + correlation_id="corr-select-hat", + ) + export = service.export_access_control_facts( + session.actor, + tenant="tenant:coulomb", + user_id=session.user.user_id, + correlation_id="corr-export-facts", + ) + self.assertIsInstance(selection, AccessProfileSelection) + self.assertIsInstance(export, AccessControlFactExport) + self._assert_no_decision_payload(selection) + self._assert_no_decision_payload(export) + for fact in export.facts: + self._assert_no_decision_payload(fact) + + def _assert_no_decision_payload(self, value) -> None: + blob = repr(value).lower() + self.assertNotIn("authorizationeffect", blob) + self.assertNotIn("effect=allow", blob) + self.assertNotIn("effect=deny", blob) + if hasattr(value, "__dict__") or hasattr(value, "__dataclass_fields__"): + names = {item.name for item in fields(type(value))} + self.assertFalse(names & _FORBIDDEN_DECISION_FIELDS) + + +def _service(): + store = InMemoryUserEngineStore() + service = UserEngineService( + store=store, + identity_adapter=FixtureIdentityClaimsAdapter(), + authorization=LocalAuthorizationCheckPort(), + ) + return service, store, None + + +def _bootstrap(service: UserEngineService): + session = service.me(_claims(), correlation_id="corr-me") + service.register_application( + session.actor, + sample_application(), + binding=sample_application_binding(), + correlation_id="corr-app", + ) + service.publish_catalog( + session.actor, + sample_catalog(), + correlation_id="corr-catalog", + ) + return session + + +def _claims(): + claims = human_actor_claims(subject="ada", tenant="tenant:coulomb") + claims["roles"] = ["tenant-admin"] + return claims + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_layer_conformance.py b/tests/test_layer_conformance.py new file mode 100644 index 0000000..266066f --- /dev/null +++ b/tests/test_layer_conformance.py @@ -0,0 +1,187 @@ +import ast +import importlib.util +import os +import subprocess +import sys +import tempfile +import unittest +from datetime import timedelta +from pathlib import Path +from unittest.mock import patch + +from user_engine.adapters.flex_auth import FlexAuthHTTPAdapter +from user_engine.adapters.local import LocalAuthorizationCheckPort +from user_engine.domain import AuthorizationEffect, utc_now +from user_engine.layer_yaml import load_mapping, load_mapping_text +from user_engine.pep_stance import ( + ALLOW_BINDING, + ALLOW_LIFETIME, + STANCE, + UNREACHABLE_STANCE, + VERDICT_CACHING, +) + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "scripts" / "check_layer_conformance.py" +LAYER = ROOT / "layer.yaml" +STANCE_FILE = ROOT / "pep-stance.yaml" +RUNTIME = ROOT / "src" / "user_engine" / "runtime.py" + + +class LayerDeclarationTests(unittest.TestCase): + def test_declares_engine_pip_in_own_voice(self): + data = load_mapping(LAYER) + self.assertEqual(data["repository"], "user-engine") + self.assertEqual(data["layer"], "engine") + self.assertEqual(data["role"], "pip") + self.assertEqual(data["tooling_contacts"], []) + self.assertEqual(data["pep_stance"], "pep-stance.yaml") + self.assertTrue(data["pep_shape"]) + self.assertEqual(data["declared_by"], "INTENT.md") + + def test_intent_frontmatter_matches_declaration(self): + text = (ROOT / "INTENT.md").read_text() + front = text.split("---", 2)[1] + intent = load_mapping_text(front) + decl = load_mapping(LAYER) + self.assertEqual(intent["layer"].lower(), decl["layer"]) + self.assertEqual(intent["role"].lower(), decl["role"]) + + def test_own_store_is_declared(self): + data = load_mapping(LAYER) + self.assertTrue(data["own_store"]) + self.assertEqual(data["own_store"][0]["import_root"], "psycopg") + + def test_checker_passes_on_the_real_tree(self): + result = subprocess.run( + [sys.executable, str(SCRIPT)], + capture_output=True, + text=True, + cwd=ROOT, + ) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_checker_catches_an_undeclared_tooling_client(self): + spec = importlib.util.spec_from_file_location( + "check_layer_conformance", SCRIPT + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + with tempfile.TemporaryDirectory() as directory: + fake_src = Path(directory) / "src" / "user_engine" + fake_src.mkdir(parents=True) + (fake_src / "secrets.py").write_text( + "import hvac\n\n\ndef leak():\n hvac.Client().read('secret')\n" + ) + original_src = module.SRC + try: + module.SRC = fake_src + hits = module.scan() + finally: + module.SRC = original_src + self.assertTrue(hits, "a direct OpenBao client was not detected") + self.assertEqual(hits[0][1], "hvac") + + +class PepStanceTests(unittest.TestCase): + def test_published_map_equals_shipped_behaviour(self): + published = load_mapping(STANCE_FILE) + self.assertEqual(published["stance"], STANCE) + self.assertEqual(published["allow_binding"], ALLOW_BINDING) + self.assertEqual( + timedelta(seconds=published["allow_lifetime_seconds"]), ALLOW_LIFETIME + ) + self.assertEqual(published["verdict_caching"], VERDICT_CACHING) + self.assertTrue( + all(value == UNREACHABLE_STANCE for value in published["stance"].values()) + ) + + def test_stance_is_total_over_the_zone_model(self): + required = { + "z0-experimental", + "z1-operational", + "z2-protected", + "z2-continuity", + "z3-critical", + "unknown", + "not-applicable", + } + self.assertEqual(required, set(STANCE)) + + def test_flex_auth_unavailable_applies_fail_closed_without_decision_id(self): + from urllib.error import URLError + + adapter = FlexAuthHTTPAdapter(base_url="http://flex-auth") + with patch( + "user_engine.adapters.flex_auth.urlopen", side_effect=URLError("down") + ): + decision = adapter.check(_request()) + self.assertEqual(decision.effect, AuthorizationEffect.DENY) + self.assertIsNone(decision.decision_id) + self.assertEqual(decision.stance_applied, UNREACHABLE_STANCE) + self.assertEqual(decision.reason, "authorization service unavailable") + + def test_flex_auth_allow_carries_published_lifetime(self): + import io + import json + + class _Response(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *_args): + self.close() + + body = _Response(json.dumps({"id": "decision:1", "effect": "allow"}).encode()) + adapter = FlexAuthHTTPAdapter(base_url="http://flex-auth") + with patch("user_engine.adapters.flex_auth.urlopen", return_value=body): + decision = adapter.check(_request()) + self.assertEqual(decision.effect, AuthorizationEffect.ALLOW) + self.assertEqual(decision.decision_id, "decision:1") + self.assertEqual(decision.binding, ALLOW_BINDING) + self.assertEqual(decision.lifetime, ALLOW_LIFETIME) + self.assertIsNotNone(decision.issued_at) + self.assertFalse(decision.expired(now=utc_now())) + + def test_production_runtime_does_not_import_local_authorization(self): + tree = ast.parse(RUNTIME.read_text()) + imported = set() + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + imported.update(alias.name for alias in node.names) + elif isinstance(node, ast.Import): + imported.update(alias.name for alias in node.names) + self.assertIn("FlexAuthHTTPAdapter", imported) + self.assertNotIn("LocalAuthorizationCheckPort", imported) + self.assertNotIn("StaticAuthorizationCheckPort", imported) + + def test_local_authorization_port_cannot_construct_with_production_token(self): + with patch.dict( + os.environ, {"USER_ENGINE_FLEX_AUTH_TOKEN_FILE": "/var/run/token"} + ): + with self.assertRaisesRegex(RuntimeError, "cannot be constructed"): + LocalAuthorizationCheckPort() + + +def _request(): + from user_engine.domain import Actor, AuthorizationRequest, PrincipalType + + actor = Actor( + issuer="https://issuer", + subject="subject-1", + tenant="tenant-a", + principal_type=PrincipalType.HUMAN, + audience=("user-engine",), + ) + return AuthorizationRequest( + actor=actor, + resource_type="user-engine:user", + resource_id="user-1", + action="user.update", + tenant="tenant-a", + correlation_id="corr-1", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_platform_adapters.py b/tests/test_platform_adapters.py index acde039..0103533 100644 --- a/tests/test_platform_adapters.py +++ b/tests/test_platform_adapters.py @@ -44,6 +44,8 @@ class PlatformAdapterTests(unittest.TestCase): decision = FlexAuthHTTPAdapter(base_url="http://flex-auth").check(_request()) self.assertEqual(decision.effect, AuthorizationEffect.DENY) self.assertEqual(decision.reason, "authorization service unavailable") + self.assertIsNone(decision.decision_id) + self.assertEqual(decision.stance_applied, "fail_closed") def test_flex_auth_reads_rotating_caller_token_for_each_decision(self): with tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8") as token_file: diff --git a/workplans/USER-WP-0024-security-layer-conformance.md b/workplans/USER-WP-0024-security-layer-conformance.md index 3e14982..7228d1f 100644 --- a/workplans/USER-WP-0024-security-layer-conformance.md +++ b/workplans/USER-WP-0024-security-layer-conformance.md @@ -4,7 +4,7 @@ type: workplan title: "Security layer model conformance for Engine/PIP" domain: communication repo: user-engine -status: active +status: finished owner: grok topic_slug: netkingdom created: "2026-08-29" @@ -52,7 +52,7 @@ by itself make the runtime conforming. ```task id: USER-WP-0024-T02 -status: todo +status: done priority: high state_hub_task_id: "2874fd05-5b54-5812-9e89-2832408e488b" ``` @@ -65,46 +65,33 @@ claim input), and the path of the PEP stance map. Adapt undeclared OpenBao, Vault, LDAP, or cluster client fails the commit that introduces it. -Do not add a Tooling client to have something to declare. - -**Done when:** `layer.yaml` is the machine-readable declaration, the -checker exits 0 on the current tree, and a test covers the empty -Tooling-contact claim. +Done 2026-08-29. `layer.yaml`, `scripts/check_layer_conformance.py`, and +`tests/test_layer_conformance.py`. Checker exits 0; a synthetic `hvac` +import fails the scan. ## T03 — PEP obligations on the production path ```task id: USER-WP-0024-T03 -status: todo +status: done priority: high state_hub_task_id: "caf7ef66-d221-5548-947b-17b5ff1865ee" ``` -Ship the four §6.4 obligations for this PEP-shaped Engine: +Ship the four §6.4 obligations for this PEP-shaped Engine. -1. Publish `pep-stance.yaml` (total, no implicit default). Production - remains fail-closed, including unknown. A test asserts the file equals - `FlexAuthHTTPAdapter` behaviour. -2. On engine-unavailable DENY, record stance application (scope, failure - mode, outcome). `decision_id` is present only when flex-auth rendered - a decision. Stop minting a local id via `AuthorizationDecision`'s - default factory on that path. -3. Give `AuthorizationDecision` an explicit lifetime (TTL or - session/obligation binding). Do not add a verdict cache. -4. Confine `LocalAuthorizationCheckPort` to tests and standalone - fixtures. Production `create_application()` must be unable to - construct a default-allow adapter. - -**Done when:** the published map matches shipped fail-closed behaviour, -unavailable DENY audits carry no engine decision id, allows carry a -lifetime, and the local port cannot enter the production runtime -assembly. +Done 2026-08-29. `pep-stance.yaml` is total fail-closed including unknown +and equals `user_engine.pep_stance`. Unavailable DENY sets +`stance_applied` and `decision_id=None`. Allows are request-bound with a +30s lifetime. `LocalAuthorizationCheckPort` raises when +`USER_ENGINE_FLEX_AUTH_TOKEN_FILE` is set; `runtime.py` does not import +it. ## T04 — Evidence bound and cadence ```task id: USER-WP-0024-T04 -status: todo +status: done priority: medium state_hub_task_id: "b4cebd66-21d0-5a6d-b718-971af70bebb6" ``` @@ -113,30 +100,30 @@ Classify local audit and outbox events as load-bearing or attributive. Keep mutation+audit+outbox atomic in the store transaction (already true). For denials and membership/account revocations — low-volume load-bearing classes — declare a heartbeat or reconciliation, not a -rate. Update `docs/evidence-gap-examples.md` and operability copy so the -trail is never described as complete, and so absence of a record is not -read as non-occurrence. +rate. -**Done when:** the classification is in-repo, the heartbeat or -reconciliation form is named, and docs use the sound evidence sentences. +Done 2026-08-29. `src/user_engine/evidence.py`, +`UserEngineService.record_evidence_heartbeat()`, +`docs/evidence-classification.md`. Operability copy no longer says +audit correlation completeness. ## T05 — Access-control facts remain claims ```task id: USER-WP-0024-T05 -status: todo +status: done priority: medium state_hub_task_id: "f1ac9244-0639-5e97-8981-babdbe8991e5" ``` Prove under statute §6.1 that hats, access profiles, and `AccessControlFact` exports cannot be mistaken for an authorization -decision. Add a test that selection and export never return allow/deny. -Document the consumer contract: compiling those facts into a local -decision is an early decision and is forbidden. +decision. -**Done when:** a test and the hats/boundary docs state the claim -contract, and no user-engine API on that path exposes an effect. +Done 2026-08-29. `tests/test_evidence.py` forbids effect fields on claim +shapes and asserts hat selection and export never return allow/deny. +`docs/hats-realms-services-assets-access-profiles.md` states the +consumer contract. ## T06 — Close the intake and request the stance-map row