diff --git a/arch01_platform.py b/arch01_platform.py new file mode 100644 index 0000000..9d18f5f --- /dev/null +++ b/arch01_platform.py @@ -0,0 +1,892 @@ +"""ARCH-01 Foundation Slice A — atomic platform installation + authority kernel (#822). + +Parents: #820, #821. **First implementation leaf of the ARCH-01 program.** + +This module implements the smallest executable ARCH-01 foundation: + +* a connection-bound authenticated actor context (``cp_actor_*`` / + ``cp_operation_mode`` / ``cp_context_epoch`` SQLite scalar functions that SQL + may *read* but can never *set* — ``[TRUSTED-SERVICE]`` authenticity); +* an immutable authority-dominance lattice with an exact seeded tuple set + (``[SCHEMA]``); +* the principal-equivalence root (a class exists *before* its first principal; + ``principals.current_class_id`` is ``NOT NULL``; ``[SCHEMA]``); +* a single-transaction platform installation that seeds the initial + ``platform.bootstrap`` grant and an immutable ``installed`` marker, validated + by a fail-closed ``install_state`` ``BEFORE INSERT`` trigger (``[SCHEMA]``). + +Everything else in the ARCH-01/02/04 program (evidence stores, repository +bindings, workspaces, PostgreSQL parity, full grant succession, full principal +merge) is out of scope here and tracked in its own issue — see #822 §5/§17. + +**Readiness / production posture.** This subsystem is *disabled by default*. +Nothing in the running MCP server imports or enables it. It becomes a security +boundary only once its readiness checks (the ACs in #822) pass in the target +environment. Instantiating :class:`PlatformKernel` creates an isolated SQLite +database and never touches the operational control-plane store. + +Enforcement classification (per #820 vocabulary): + +* ``[TRUSTED-SERVICE]`` — actor-context authenticity: the scalar functions are + registered by the trusted Python process; SQL cannot define or redefine them. +* ``[SCHEMA]`` — fail-closed aborts, the dominance/immutability/NOT-NULL-class/ + last-active-grant invariants, enforced by CHECK/FK/trigger. +* ``[RUNTIME-ADAPTER]`` — *none* in this slice. + +SQLite-first. ``BEGIN IMMEDIATE`` serializes concurrent installs and concurrent +grant/revoke on the singleton invariant row. PostgreSQL parity is a distinct +issue (#827); this module does **not** claim it. +""" + +from __future__ import annotations + +import os +import sqlite3 +import threading +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Iterator, Optional + +# --------------------------------------------------------------------------- # +# Closed enumerations (#822 §4). +# --------------------------------------------------------------------------- # + +ACTOR_KINDS = ("operator", "supervisor", "service", "installer") +OPERATION_MODES = ("normal", "install", "merge", "internal_service") + +# Exact seeded authority-dominance tuple set (#822 §4). This set is normative: +# the install-state trigger rejects any missing, additional, or malformed tuple. +DOMINANCE_TUPLES = ( + ("platform.bootstrap", "platform.bootstrap"), + ("platform.bootstrap", "project.admin"), + ("platform.bootstrap", "supervisor.root.establish"), + ("supervisor.root", "supervisor.register"), + ("supervisor.root", "supervisor.verify"), + ("supervisor.root", "supervisor.recover"), +) + +# The distinguished operator-key issuer seeded during install. +DISTINGUISHED_ISSUER_KIND = "operator-key" +DISTINGUISHED_ISSUER_ID = "platform.bootstrap.operator-key" + +# Structured result codes (#822 §10). +INSTALLED = "INSTALLED" +ALREADY_INSTALLED = "ALREADY_INSTALLED" +INVALID_ACTOR_CONTEXT = "INVALID_ACTOR_CONTEXT" +INVALID_BOOTSTRAP_STATE = "INVALID_BOOTSTRAP_STATE" +DOMINANCE_SET_MISMATCH = "DOMINANCE_SET_MISMATCH" +AUTHORIZATION_DENIED = "AUTHORIZATION_DENIED" +CONCURRENT_INSTALLATION_LOST = "CONCURRENT_INSTALLATION_LOST" + +# Required audit events (#822 §14). +EVT_PLATFORM_INSTALLED = "platform_installed" +EVT_GRANT_CREATED = "platform_grant_created" +EVT_GRANT_REVOKED = "platform_grant_revoked" +EVT_PRINCIPAL_REGISTERED = "principal_registered" + +SCHEMA_VERSION = 1 + +DB_PATH_ENV = "ARCH01_PLATFORM_DB" + + +class PlatformKernelError(RuntimeError): + """Base class for structured, code-bearing kernel failures.""" + + def __init__(self, code: str, message: str = "") -> None: + super().__init__(message or code) + self.code = code + + +class ActorContextError(PlatformKernelError): + """Raised when a mutation is attempted without a valid actor context.""" + + +# --------------------------------------------------------------------------- # +# Schema (#822 §6). Tables + fail-closed triggers. +# +# Every *mutating* trigger opens with the actor protocol: read the context +# epoch, read the actor fields, and abort unless the context is present, +# non-null, mode/kind well-formed, and epoch-consistent with the active +# transaction. The scalar functions ``cp_*`` are registered from Python only; +# SQL has no statement that can set them, which is the trusted-service boundary. +# --------------------------------------------------------------------------- # + +_ACTOR_KINDS_SQL = ", ".join("'%s'" % k for k in ACTOR_KINDS) +_OP_MODES_SQL = ", ".join("'%s'" % m for m in OPERATION_MODES) + +# Actor-protocol predicate: TRUE when the context is INVALID and the trigger +# must abort. ``cp_actor_context_valid()`` folds "present + non-expired + +# live-epoch == bound-epoch" (the read/re-read epoch equality of #822 §4) into +# one trusted-service answer; the remaining reads assert field well-formedness. +_INVALID_ACTOR = ( + "cp_actor_context_valid() IS NOT 1 " + "OR cp_context_epoch() IS NULL " + "OR cp_actor_principal() IS NULL " + "OR cp_actor_kind() NOT IN (%s) " + "OR cp_operation_mode() NOT IN (%s)" % (_ACTOR_KINDS_SQL, _OP_MODES_SQL) +) + +_ACTOR_GUARD = ( + "SELECT CASE WHEN (%s) " + "THEN RAISE(ABORT, 'INVALID_ACTOR_CONTEXT') END;" % _INVALID_ACTOR +) + +# require_installed: abort a privileged mutation when there is no install +# marker and we are not currently installing (#822 §4). +_REQUIRE_INSTALLED = ( + "SELECT CASE WHEN ((SELECT COUNT(*) FROM install_state) = 0 " + "AND cp_operation_mode() <> 'install') " + "THEN RAISE(ABORT, 'NOT_INSTALLED') END;" +) + +_SCHEMA_SQL = f""" +PRAGMA foreign_keys = ON; + +CREATE TABLE IF NOT EXISTS arch01_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); + +-- Equivalence classes are created BEFORE their first principal (#822 §4). +CREATE TABLE IF NOT EXISTS principal_equivalence_classes ( + class_id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS authoritative_issuers ( + issuer_id INTEGER PRIMARY KEY AUTOINCREMENT, + issuer_kind TEXT NOT NULL, + issuer_ref TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE (issuer_kind, issuer_ref) +); + +-- current_class_id is NOT NULL: a principal cannot exist without a class +-- (#822 AC6). issuer_id is nullable ONLY for the installer during install +-- (#822 AC7), enforced by trg_principals_null_issuer below. +CREATE TABLE IF NOT EXISTS principals ( + principal_id TEXT PRIMARY KEY, + actor_kind TEXT NOT NULL CHECK (actor_kind IN ({_ACTOR_KINDS_SQL})), + current_class_id INTEGER NOT NULL REFERENCES principal_equivalence_classes(class_id), + issuer_id INTEGER REFERENCES authoritative_issuers(issuer_id), + registered_by TEXT REFERENCES principals(principal_id), + created_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS authority_dominance ( + dominant TEXT NOT NULL, + subordinate TEXT NOT NULL, + PRIMARY KEY (dominant, subordinate) +); + +CREATE TABLE IF NOT EXISTS platform_bootstrap_seed ( + seed_id INTEGER PRIMARY KEY CHECK (seed_id = 1), + installer_principal_id TEXT NOT NULL REFERENCES principals(principal_id), + created_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS platform_bootstrap_grants ( + grant_id INTEGER PRIMARY KEY AUTOINCREMENT, + grantee_principal_id TEXT NOT NULL REFERENCES principals(principal_id), + granted_by TEXT REFERENCES principals(principal_id), + active INTEGER NOT NULL DEFAULT 1 CHECK (active IN (0, 1)), + created_at TEXT NOT NULL, + revoked_at TEXT +); + +-- Singleton row; active_count floored at 1 by CHECK so the last active grant +-- can never be revoked (#822 AC11). +CREATE TABLE IF NOT EXISTS platform_active_invariant ( + id INTEGER PRIMARY KEY CHECK (id = 1), + active_count INTEGER NOT NULL CHECK (active_count >= 1) +); + +-- The immutable install marker; inserted LAST in the install transaction. +CREATE TABLE IF NOT EXISTS install_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + marker TEXT NOT NULL CHECK (marker = 'installed'), + installed_at TEXT NOT NULL +); + +-- Append-only (#822 AC14). +CREATE TABLE IF NOT EXISTS audit_records ( + audit_id INTEGER PRIMARY KEY AUTOINCREMENT, + event TEXT NOT NULL, + principal_id TEXT, + detail TEXT, + created_at TEXT NOT NULL +); + +-- ------------------------------------------------------------------------- -- +-- Actor protocol on every mutating trigger (#822 §4, [SCHEMA] fail-closed). +-- ------------------------------------------------------------------------- -- + +CREATE TRIGGER IF NOT EXISTS trg_classes_actor +BEFORE INSERT ON principal_equivalence_classes +BEGIN + {_ACTOR_GUARD} +END; + +CREATE TRIGGER IF NOT EXISTS trg_issuers_actor +BEFORE INSERT ON authoritative_issuers +BEGIN + {_ACTOR_GUARD} +END; + +CREATE TRIGGER IF NOT EXISTS trg_principals_actor +BEFORE INSERT ON principals +BEGIN + {_ACTOR_GUARD} +END; + +CREATE TRIGGER IF NOT EXISTS trg_dominance_actor +BEFORE INSERT ON authority_dominance +BEGIN + {_ACTOR_GUARD} +END; + +CREATE TRIGGER IF NOT EXISTS trg_seed_actor +BEFORE INSERT ON platform_bootstrap_seed +BEGIN + {_ACTOR_GUARD} +END; + +CREATE TRIGGER IF NOT EXISTS trg_grants_actor_insert +BEFORE INSERT ON platform_bootstrap_grants +BEGIN + {_ACTOR_GUARD} + {_REQUIRE_INSTALLED} +END; + +CREATE TRIGGER IF NOT EXISTS trg_grants_actor_update +BEFORE UPDATE ON platform_bootstrap_grants +BEGIN + {_ACTOR_GUARD} +END; + +CREATE TRIGGER IF NOT EXISTS trg_invariant_actor_insert +BEFORE INSERT ON platform_active_invariant +BEGIN + {_ACTOR_GUARD} +END; + +CREATE TRIGGER IF NOT EXISTS trg_invariant_actor_update +BEFORE UPDATE ON platform_active_invariant +BEGIN + {_ACTOR_GUARD} +END; + +CREATE TRIGGER IF NOT EXISTS trg_audit_actor +BEFORE INSERT ON audit_records +BEGIN + {_ACTOR_GUARD} +END; + +-- ------------------------------------------------------------------------- -- +-- NOT-NULL-issuer exception for the installer only (#822 AC7). +-- A NULL issuer_id is accepted solely for an installer principal during +-- install mode, before the marker exists; any other NULL-issuer principal is +-- rejected. install-time issuer linkage (installer -> distinguished issuer) +-- is applied by a later UPDATE, permitted while no marker exists. +-- ------------------------------------------------------------------------- -- + +CREATE TRIGGER IF NOT EXISTS trg_principals_null_issuer +BEFORE INSERT ON principals +WHEN NEW.issuer_id IS NULL +BEGIN + SELECT CASE WHEN NOT ( + NEW.actor_kind = 'installer' + AND cp_operation_mode() = 'install' + AND (SELECT COUNT(*) FROM install_state) = 0 + AND (SELECT COUNT(*) FROM principals WHERE issuer_id IS NULL) = 0 + ) THEN RAISE(ABORT, 'INVALID_BOOTSTRAP_STATE') END; +END; + +-- ------------------------------------------------------------------------- -- +-- Post-install immutability of the authority root (#822 §4, AC9). +-- Registration fields freeze only AFTER the marker exists, so the install +-- transaction's own installer issuer-linkage UPDATE is permitted. +-- ------------------------------------------------------------------------- -- + +CREATE TRIGGER IF NOT EXISTS trg_principals_frozen_update +BEFORE UPDATE ON principals +WHEN (SELECT COUNT(*) FROM install_state) > 0 +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_PRINCIPAL'); +END; + +CREATE TRIGGER IF NOT EXISTS trg_principals_frozen_delete +BEFORE DELETE ON principals +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_PRINCIPAL'); +END; + +-- Distinguished issuer identity is immutable once written. +CREATE TRIGGER IF NOT EXISTS trg_issuers_immutable_update +BEFORE UPDATE ON authoritative_issuers +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_ISSUER'); +END; + +CREATE TRIGGER IF NOT EXISTS trg_issuers_immutable_delete +BEFORE DELETE ON authoritative_issuers +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_ISSUER'); +END; + +-- The dominance lattice is immutable once seeded. +CREATE TRIGGER IF NOT EXISTS trg_dominance_immutable_update +BEFORE UPDATE ON authority_dominance +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_DOMINANCE'); +END; + +CREATE TRIGGER IF NOT EXISTS trg_dominance_immutable_delete +BEFORE DELETE ON authority_dominance +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_DOMINANCE'); +END; + +-- The bootstrap seed is immutable once written. +CREATE TRIGGER IF NOT EXISTS trg_seed_immutable_update +BEFORE UPDATE ON platform_bootstrap_seed +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_SEED'); +END; + +CREATE TRIGGER IF NOT EXISTS trg_seed_immutable_delete +BEFORE DELETE ON platform_bootstrap_seed +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_SEED'); +END; + +-- The install marker is immutable once written. +CREATE TRIGGER IF NOT EXISTS trg_install_state_immutable_update +BEFORE UPDATE ON install_state +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_INSTALL_STATE'); +END; + +CREATE TRIGGER IF NOT EXISTS trg_install_state_immutable_delete +BEFORE DELETE ON install_state +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_INSTALL_STATE'); +END; + +-- Grants: identity is immutable; the ONLY permitted mutation is a single +-- active 1 -> 0 revocation (#822 §4 initial-grant identity immutability + +-- grant/revoke). Reactivation and identity edits are rejected. +CREATE TRIGGER IF NOT EXISTS trg_grants_identity_frozen +BEFORE UPDATE ON platform_bootstrap_grants +WHEN NOT ( + NEW.grant_id = OLD.grant_id + AND NEW.grantee_principal_id = OLD.grantee_principal_id + AND NEW.granted_by IS OLD.granted_by + AND NEW.created_at = OLD.created_at + AND OLD.active = 1 + AND NEW.active = 0 +) +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_GRANT'); +END; + +CREATE TRIGGER IF NOT EXISTS trg_grants_no_delete +BEFORE DELETE ON platform_bootstrap_grants +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_GRANT'); +END; + +-- audit_records is append-only. +CREATE TRIGGER IF NOT EXISTS trg_audit_immutable_update +BEFORE UPDATE ON audit_records +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_AUDIT'); +END; + +CREATE TRIGGER IF NOT EXISTS trg_audit_immutable_delete +BEFORE DELETE ON audit_records +BEGIN + SELECT RAISE(ABORT, 'IMMUTABLE_AUDIT'); +END; + +-- ------------------------------------------------------------------------- -- +-- install_state BEFORE INSERT: validate the whole bootstrap atomically +-- (#822 §4, AC4). Each dominance tuple is checked individually; a missing, +-- additional, or malformed tuple -> DOMINANCE_SET_MISMATCH. The seed<->installer +-- link, the single active NULL-grantor installer grant, the installer's +-- non-NULL issuer, the active invariant, and "no extra principal created under +-- the NULL-issuer exception" -> INVALID_BOOTSTRAP_STATE. +-- ------------------------------------------------------------------------- -- + +CREATE TRIGGER IF NOT EXISTS trg_install_state_validate +BEFORE INSERT ON install_state +BEGIN + SELECT CASE WHEN NOT ( + (SELECT COUNT(*) FROM authority_dominance) = {len(DOMINANCE_TUPLES)} + AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='platform.bootstrap' AND subordinate='platform.bootstrap') + AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='platform.bootstrap' AND subordinate='project.admin') + AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='platform.bootstrap' AND subordinate='supervisor.root.establish') + AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='supervisor.root' AND subordinate='supervisor.register') + AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='supervisor.root' AND subordinate='supervisor.verify') + AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='supervisor.root' AND subordinate='supervisor.recover') + ) THEN RAISE(ABORT, 'DOMINANCE_SET_MISMATCH') END; + + SELECT CASE WHEN NOT ( + (SELECT COUNT(*) FROM platform_bootstrap_seed) = 1 + AND (SELECT COUNT(*) FROM principals) = 1 + AND (SELECT actor_kind FROM principals + WHERE principal_id = (SELECT installer_principal_id FROM platform_bootstrap_seed WHERE seed_id = 1) + ) = 'installer' + AND (SELECT issuer_id FROM principals + WHERE principal_id = (SELECT installer_principal_id FROM platform_bootstrap_seed WHERE seed_id = 1) + ) IS NOT NULL + AND (SELECT COUNT(*) FROM platform_bootstrap_grants + WHERE granted_by IS NULL AND active = 1 + AND grantee_principal_id = (SELECT installer_principal_id FROM platform_bootstrap_seed WHERE seed_id = 1) + ) = 1 + AND (SELECT COUNT(*) FROM platform_bootstrap_grants) = 1 + AND (SELECT active_count FROM platform_active_invariant WHERE id = 1) = 1 + ) THEN RAISE(ABORT, 'INVALID_BOOTSTRAP_STATE') END; +END; +""" + + +def default_db_path() -> str: + return os.environ.get( + DB_PATH_ENV, + os.path.expanduser("~/.cache/gitea-tools/arch01/platform.sqlite3"), + ) + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +@dataclass(frozen=True) +class OperationResult: + """Structured result of a kernel operation (#822 §10).""" + + code: str + detail: str = "" + + @property + def ok(self) -> bool: + return self.code in (INSTALLED, ALREADY_INSTALLED) + + +@dataclass +class _ActorContext: + principal: str + kind: str + mode: str + session: Optional[str] + bound_epoch: int + live_epoch: int + expired: bool = False + + +class PlatformKernel: + """ARCH-01 authority kernel over a single SQLite connection. + + The connection carries the trusted-service actor context: the ``cp_*`` + scalar functions read the context this object holds. Only Python code here + can bind or clear it, so no SQL statement can assert an actor identity — the + trusted-service authenticity boundary of #822 §4. + """ + + def __init__(self, db_path: Optional[str] = None, *, busy_timeout_ms: int = 5000) -> None: + self.db_path = db_path or default_db_path() + if self.db_path != ":memory:": + parent = os.path.dirname(self.db_path) + if parent: + os.makedirs(parent, exist_ok=True) + self._ctx: Optional[_ActorContext] = None + self._epoch_seq = 0 + self._lock = threading.Lock() + # check_same_thread=False is safe: every mutation path is serialized + # by self._lock, so the connection is never used concurrently even when + # callers drive the kernel from different threads (concurrency tests). + self._conn = sqlite3.connect( + self.db_path, isolation_level=None, check_same_thread=False + ) + self._conn.execute("PRAGMA foreign_keys = ON") + self._conn.execute(f"PRAGMA busy_timeout = {int(busy_timeout_ms)}") + self._register_actor_functions() + self._migrate() + + # -- trusted-service actor functions ---------------------------------- # + + def _register_actor_functions(self) -> None: + c = self._conn + c.create_function("cp_actor_principal", 0, lambda: self._ctx.principal if self._ctx else None) + c.create_function("cp_actor_kind", 0, lambda: self._ctx.kind if self._ctx else None) + c.create_function("cp_operation_mode", 0, lambda: self._ctx.mode if self._ctx else None) + c.create_function("cp_service_session", 0, lambda: self._ctx.session if self._ctx else None) + c.create_function("cp_context_epoch", 0, self._fn_context_epoch) + # Trusted-service helper: folds present + non-expired + epoch-consistent + # into the read/re-read epoch equality of #822 §4. + c.create_function("cp_actor_context_valid", 0, self._fn_context_valid) + + def _fn_context_epoch(self) -> Optional[int]: + if self._ctx is None or self._ctx.expired: + return None + return self._ctx.live_epoch + + def _fn_context_valid(self) -> int: + ctx = self._ctx + if ctx is None or ctx.expired: + return 0 + # read/re-read epoch equality: a context whose live epoch has drifted + # from the epoch it was bound to (a stale/replaced connection context) + # is not bound to the active transaction and fails closed. + if ctx.live_epoch != ctx.bound_epoch: + return 0 + if ctx.principal is None: + return 0 + if ctx.kind not in ACTOR_KINDS or ctx.mode not in OPERATION_MODES: + return 0 + return 1 + + # -- context lifecycle ------------------------------------------------ # + + @contextmanager + def actor_context( + self, principal: str, kind: str, mode: str, session: Optional[str] = None + ) -> Iterator[None]: + """Bind a trusted actor context for the duration of the block.""" + prev = self._ctx + self._epoch_seq += 1 + epoch = self._epoch_seq + self._ctx = _ActorContext( + principal=principal, kind=kind, mode=mode, session=session, + bound_epoch=epoch, live_epoch=epoch, + ) + try: + yield + finally: + self._ctx = prev + + def _clear_context(self) -> None: + self._ctx = None + + # -- migration -------------------------------------------------------- # + + def _migrate(self) -> None: + self._conn.executescript(_SCHEMA_SQL) + self._conn.execute( + "INSERT OR IGNORE INTO arch01_meta(key, value) VALUES ('schema_version', ?)", + (str(SCHEMA_VERSION),), + ) + self._conn.execute( + "INSERT OR IGNORE INTO arch01_meta(key, value) VALUES " + "('architecture', 'ARCH-01 Slice A: atomic install + authority kernel (#822); " + "disabled by default until readiness checks pass')" + ) + + # -- introspection ---------------------------------------------------- # + + def is_installed(self) -> bool: + row = self._conn.execute("SELECT COUNT(*) FROM install_state").fetchone() + return bool(row[0]) + + def active_grant_count(self) -> int: + row = self._conn.execute( + "SELECT active_count FROM platform_active_invariant WHERE id = 1" + ).fetchone() + return int(row[0]) if row else 0 + + def audit_events(self) -> list[str]: + return [ + r[0] + for r in self._conn.execute( + "SELECT event FROM audit_records ORDER BY audit_id" + ).fetchall() + ] + + def close(self) -> None: + self._conn.close() + + # -- operations ------------------------------------------------------- # + + def install_platform( + self, + installer_principal_id: str = "platform.installer", + *, + session: Optional[str] = None, + ) -> OperationResult: + """Single atomic install transaction (#822 §4/§7). + + ``BEGIN IMMEDIATE`` serializes concurrent installs; the loser rechecks + the marker and returns ``ALREADY_INSTALLED``, or — if it never acquires + the write lock — ``CONCURRENT_INSTALLATION_LOST``. On any stage failure + the whole transaction rolls back leaving no partial rows (AC3/AC5). + """ + now = _utc_now_iso() + with self._lock: + try: + self._conn.execute("BEGIN IMMEDIATE") + except sqlite3.OperationalError as exc: + if "locked" in str(exc).lower() or "busy" in str(exc).lower(): + return OperationResult(CONCURRENT_INSTALLATION_LOST, str(exc)) + raise + try: + if self.is_installed(): + self._conn.execute("ROLLBACK") + return OperationResult(ALREADY_INSTALLED, "install marker already present") + + with self.actor_context(installer_principal_id, "installer", "install", session): + c = self._conn + # class -> installer principal (temporary NULL issuer) + cur = c.execute( + "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", + (now,), + ) + class_id = cur.lastrowid + c.execute( + "INSERT INTO principals" + "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) " + "VALUES (?, 'installer', ?, NULL, ?, ?)", + (installer_principal_id, class_id, installer_principal_id, now), + ) + # distinguished operator-key issuer + cur = c.execute( + "INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) " + "VALUES (?, ?, ?)", + (DISTINGUISHED_ISSUER_KIND, DISTINGUISHED_ISSUER_ID, now), + ) + issuer_id = cur.lastrowid + # link installer -> issuer (permitted pre-marker) + c.execute( + "UPDATE principals SET issuer_id = ? WHERE principal_id = ?", + (issuer_id, installer_principal_id), + ) + # dominance tuples + c.executemany( + "INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)", + DOMINANCE_TUPLES, + ) + # seed + c.execute( + "INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) " + "VALUES (1, ?, ?)", + (installer_principal_id, now), + ) + # initial grant (granted_by NULL, active) + c.execute( + "INSERT INTO platform_bootstrap_grants" + "(grantee_principal_id, granted_by, active, created_at) " + "VALUES (?, NULL, 1, ?)", + (installer_principal_id, now), + ) + # active invariant + c.execute( + "INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)" + ) + # audit rows for the security-sensitive operation + c.execute( + "INSERT INTO audit_records(event, principal_id, detail, created_at) " + "VALUES (?, ?, ?, ?)", + (EVT_PRINCIPAL_REGISTERED, installer_principal_id, "installer", now), + ) + c.execute( + "INSERT INTO audit_records(event, principal_id, detail, created_at) " + "VALUES (?, ?, ?, ?)", + (EVT_GRANT_CREATED, installer_principal_id, "initial platform.bootstrap grant", now), + ) + # install marker LAST -> fires the whole-bootstrap validator + c.execute( + "INSERT INTO install_state(id, marker, installed_at) VALUES (1, 'installed', ?)", + (now,), + ) + c.execute( + "INSERT INTO audit_records(event, principal_id, detail, created_at) " + "VALUES (?, ?, ?, ?)", + (EVT_PLATFORM_INSTALLED, installer_principal_id, "platform installed", now), + ) + self._conn.execute("COMMIT") + return OperationResult(INSTALLED, "platform installed") + except sqlite3.Error as exc: + self._safe_rollback() + return OperationResult(self._classify(exc), str(exc)) + + def register_principal( + self, + principal_id: str, + actor_kind: str, + issuer_ref: str, + *, + actor_principal: str, + actor_kind_ctx: str = "operator", + session: Optional[str] = None, + ) -> OperationResult: + """Atomically create an equivalence class and its first principal. + + The class is inserted *before* the principal, and ``current_class_id`` + is ``NOT NULL`` (#822 AC6): a principal can never exist classless. + The principal references an existing issuer (non-NULL); the temporary + NULL-issuer exception is reserved for the installer during install + (AC7). + """ + if actor_kind not in ACTOR_KINDS: + return OperationResult(INVALID_BOOTSTRAP_STATE, f"bad actor_kind {actor_kind!r}") + now = _utc_now_iso() + with self._lock: + try: + self._conn.execute("BEGIN IMMEDIATE") + except sqlite3.OperationalError as exc: + return OperationResult(AUTHORIZATION_DENIED, str(exc)) + try: + if not self.is_installed(): + self._conn.execute("ROLLBACK") + return OperationResult(INVALID_BOOTSTRAP_STATE, "platform not installed") + row = self._conn.execute( + "SELECT issuer_id FROM authoritative_issuers WHERE issuer_ref = ?", + (issuer_ref,), + ).fetchone() + if row is None: + self._conn.execute("ROLLBACK") + return OperationResult(INVALID_BOOTSTRAP_STATE, f"unknown issuer {issuer_ref!r}") + issuer_id = row[0] + with self.actor_context(actor_principal, actor_kind_ctx, "normal", session): + cur = self._conn.execute( + "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", + (now,), + ) + class_id = cur.lastrowid + self._conn.execute( + "INSERT INTO principals" + "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (principal_id, actor_kind, class_id, issuer_id, actor_principal, now), + ) + self._conn.execute( + "INSERT INTO audit_records(event, principal_id, detail, created_at) " + "VALUES (?, ?, ?, ?)", + (EVT_PRINCIPAL_REGISTERED, principal_id, actor_kind, now), + ) + self._conn.execute("COMMIT") + return OperationResult(INSTALLED, f"registered {principal_id}") + except sqlite3.Error as exc: + self._safe_rollback() + return OperationResult(self._classify(exc), str(exc)) + + def grant_platform_bootstrap( + self, + grantee_principal_id: str, + granted_by: str, + *, + actor_kind_ctx: str = "operator", + session: Optional[str] = None, + ) -> OperationResult: + """Create an additional active platform.bootstrap grant. + + Serialized on the singleton invariant row via ``BEGIN IMMEDIATE``. + """ + now = _utc_now_iso() + with self._lock: + try: + self._conn.execute("BEGIN IMMEDIATE") + except sqlite3.OperationalError as exc: + return OperationResult(AUTHORIZATION_DENIED, str(exc)) + try: + if not self.is_installed(): + self._conn.execute("ROLLBACK") + return OperationResult(INVALID_BOOTSTRAP_STATE, "platform not installed") + with self.actor_context(granted_by, actor_kind_ctx, "normal", session): + self._conn.execute( + "INSERT INTO platform_bootstrap_grants" + "(grantee_principal_id, granted_by, active, created_at) " + "VALUES (?, ?, 1, ?)", + (grantee_principal_id, granted_by, now), + ) + self._conn.execute( + "UPDATE platform_active_invariant SET active_count = active_count + 1 WHERE id = 1" + ) + self._conn.execute( + "INSERT INTO audit_records(event, principal_id, detail, created_at) " + "VALUES (?, ?, ?, ?)", + (EVT_GRANT_CREATED, grantee_principal_id, f"granted_by={granted_by}", now), + ) + self._conn.execute("COMMIT") + return OperationResult(INSTALLED, f"granted to {grantee_principal_id}") + except sqlite3.Error as exc: + self._safe_rollback() + return OperationResult(self._classify(exc), str(exc)) + + def revoke_platform_bootstrap( + self, + grant_id: int, + *, + actor_principal: str, + actor_kind_ctx: str = "operator", + session: Optional[str] = None, + ) -> OperationResult: + """Revoke an active grant, floored so the last one can never drop. + + The ``active_count >= 1`` CHECK plus ``BEGIN IMMEDIATE`` serialization + make two concurrent revocations unable to remove the final active grant + (#822 AC11): the decrement that would reach zero fails and rolls back. + """ + now = _utc_now_iso() + with self._lock: + try: + self._conn.execute("BEGIN IMMEDIATE") + except sqlite3.OperationalError as exc: + return OperationResult(AUTHORIZATION_DENIED, str(exc)) + try: + if not self.is_installed(): + self._conn.execute("ROLLBACK") + return OperationResult(INVALID_BOOTSTRAP_STATE, "platform not installed") + row = self._conn.execute( + "SELECT active, grantee_principal_id FROM platform_bootstrap_grants WHERE grant_id = ?", + (grant_id,), + ).fetchone() + if row is None or row[0] != 1: + self._conn.execute("ROLLBACK") + return OperationResult(AUTHORIZATION_DENIED, "grant absent or already inactive") + grantee = row[1] + with self.actor_context(actor_principal, actor_kind_ctx, "normal", session): + # Decrement first: the CHECK floor rejects dropping below 1, + # aborting the whole revoke before the grant flips inactive. + self._conn.execute( + "UPDATE platform_active_invariant SET active_count = active_count - 1 WHERE id = 1" + ) + self._conn.execute( + "UPDATE platform_bootstrap_grants SET active = 0, revoked_at = ? WHERE grant_id = ?", + (now, grant_id), + ) + self._conn.execute( + "INSERT INTO audit_records(event, principal_id, detail, created_at) " + "VALUES (?, ?, ?, ?)", + (EVT_GRANT_REVOKED, grantee, f"grant_id={grant_id}", now), + ) + self._conn.execute("COMMIT") + return OperationResult(INSTALLED, f"revoked grant {grant_id}") + except sqlite3.Error as exc: + self._safe_rollback() + return OperationResult(self._classify(exc), str(exc)) + + # -- helpers ---------------------------------------------------------- # + + def _safe_rollback(self) -> None: + try: + self._conn.execute("ROLLBACK") + except sqlite3.Error: + pass + + @staticmethod + def _classify(exc: sqlite3.Error) -> str: + msg = str(exc) + if "INVALID_ACTOR_CONTEXT" in msg: + return INVALID_ACTOR_CONTEXT + if "DOMINANCE_SET_MISMATCH" in msg: + return DOMINANCE_SET_MISMATCH + if "active_count" in msg or "CHECK constraint failed: platform_active_invariant" in msg: + # last-active-grant floor tripped + return AUTHORIZATION_DENIED + if any(tag in msg for tag in ( + "INVALID_BOOTSTRAP_STATE", "IMMUTABLE_", "NOT_INSTALLED", + )): + return INVALID_BOOTSTRAP_STATE + return INVALID_BOOTSTRAP_STATE diff --git a/tests/test_arch01_platform.py b/tests/test_arch01_platform.py new file mode 100644 index 0000000..bb4272c --- /dev/null +++ b/tests/test_arch01_platform.py @@ -0,0 +1,572 @@ +"""Executable acceptance tests for ARCH-01 Slice A (#822). + +Each acceptance criterion (#822 §12) and named test (#822 §13) is exercised +against a real SQLite database. The migration runs on a fresh DB in ``setUp``; +the test-run output is the durable evidence the issue requires (§14). + +Enforcement being proven: + +* ``[TRUSTED-SERVICE]`` — the ``cp_*`` actor functions exist only on the + trusted kernel connection; a raw connection cannot satisfy the triggers. +* ``[SCHEMA]`` — fail-closed aborts, exact dominance set, NOT-NULL class, + immutability, and the last-active-grant floor are enforced by + CHECK/FK/trigger, verified here including raw-write bypass and concurrency. +""" + +from __future__ import annotations + +import os +import sqlite3 +import tempfile +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor + +import arch01_platform as ap +from arch01_platform import ( + ALREADY_INSTALLED, + AUTHORIZATION_DENIED, + CONCURRENT_INSTALLATION_LOST, + DISTINGUISHED_ISSUER_ID, + DOMINANCE_SET_MISMATCH, + DOMINANCE_TUPLES, + INSTALLED, + INVALID_ACTOR_CONTEXT, + INVALID_BOOTSTRAP_STATE, + PlatformKernel, +) + +INSTALLER = "platform.installer" + +_BOOTSTRAP_TABLES = ( + "principal_equivalence_classes", + "principals", + "authoritative_issuers", + "authority_dominance", + "platform_bootstrap_seed", + "platform_bootstrap_grants", + "platform_active_invariant", + "install_state", +) + + +def _count(kernel: PlatformKernel, table: str) -> int: + return kernel._conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] + + +def _count_where(kernel: PlatformKernel, table: str, where: str) -> int: + return kernel._conn.execute(f"SELECT COUNT(*) FROM {table} WHERE {where}").fetchone()[0] + + +def _all_bootstrap_empty(kernel: PlatformKernel) -> bool: + return all(_count(kernel, t) == 0 for t in _BOOTSTRAP_TABLES) + + +class Arch01MemoryTest(unittest.TestCase): + """Single-connection behavior on an in-memory database.""" + + def setUp(self) -> None: + self.kernel = PlatformKernel(":memory:") + + def tearDown(self) -> None: + self.kernel.close() + + # -- AC1 -------------------------------------------------------------- # + def test_install_clean(self) -> None: # t_install_clean(+) + res = self.kernel.install_platform(INSTALLER) + self.assertEqual(res.code, INSTALLED) + self.assertTrue(self.kernel.is_installed()) + self.assertEqual(_count(self.kernel, "install_state"), 1) + self.assertEqual(self.kernel.active_grant_count(), 1) + self.assertIn(ap.EVT_PLATFORM_INSTALLED, self.kernel.audit_events()) + rows = set( + self.kernel._conn.execute( + "SELECT dominant, subordinate FROM authority_dominance" + ).fetchall() + ) + self.assertEqual(rows, set(DOMINANCE_TUPLES)) + issuer_ref = self.kernel._conn.execute( + "SELECT i.issuer_ref FROM principals p JOIN authoritative_issuers i " + "ON p.issuer_id = i.issuer_id WHERE p.principal_id = ?", + (INSTALLER,), + ).fetchone() + self.assertEqual(issuer_ref[0], DISTINGUISHED_ISSUER_ID) + + # -- AC2 -------------------------------------------------------------- # + def test_install_twice(self) -> None: # t_install_twice(-) + self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED) + res2 = self.kernel.install_platform(INSTALLER) + self.assertEqual(res2.code, ALREADY_INSTALLED) + self.assertEqual(_count(self.kernel, "principals"), 1) + self.assertEqual(_count(self.kernel, "platform_bootstrap_grants"), 1) + self.assertEqual(_count(self.kernel, "install_state"), 1) + + # -- AC3 / AC5 -------------------------------------------------------- # + def test_install_stage_rollback(self) -> None: # t_install_stage_rollback + for stop in range(1, 9): + with self.subTest(stages=stop): + k = PlatformKernel(":memory:") + try: + self._partial_bootstrap_then_rollback(k, stop) + self.assertTrue( + _all_bootstrap_empty(k), + f"partial rows survived rollback at stage {stop}", + ) + self.assertFalse(k.is_installed()) + finally: + k.close() + + def test_no_partial_after_rollback(self) -> None: # t_no_partial_after_rollback + k = PlatformKernel(":memory:") + try: + code = self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1]) + self.assertEqual(code, DOMINANCE_SET_MISMATCH) + self.assertTrue(_all_bootstrap_empty(k)) + self.assertFalse(k.is_installed()) + finally: + k.close() + + # -- AC4 -------------------------------------------------------------- # + def test_dominance_missing(self) -> None: # t_dominance_missing(-) + k = PlatformKernel(":memory:") + try: + self.assertEqual( + self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1]), + DOMINANCE_SET_MISMATCH, + ) + self.assertFalse(k.is_installed()) + finally: + k.close() + + def test_dominance_extra(self) -> None: # t_dominance_extra(-) + k = PlatformKernel(":memory:") + try: + extra = DOMINANCE_TUPLES + (("platform.bootstrap", "rogue.extra"),) + self.assertEqual( + self._seed_bootstrap_and_mark(k, dominance=extra), + DOMINANCE_SET_MISMATCH, + ) + self.assertFalse(k.is_installed()) + finally: + k.close() + + def test_dominance_malformed(self) -> None: # t_dominance_malformed(-) + k = PlatformKernel(":memory:") + try: + malformed = DOMINANCE_TUPLES[:-1] + (("supervisor.root", "WRONG.subordinate"),) + self.assertEqual( + self._seed_bootstrap_and_mark(k, dominance=malformed), + DOMINANCE_SET_MISMATCH, + ) + self.assertFalse(k.is_installed()) + finally: + k.close() + + # -- AC6 -------------------------------------------------------------- # + def test_principal_no_class(self) -> None: # t_principal_no_class(-) + with self.kernel.actor_context("op", "operator", "install"): + with self.assertRaises(sqlite3.IntegrityError): + self.kernel._conn.execute( + "INSERT INTO principals" + "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) " + "VALUES ('x', 'operator', NULL, NULL, NULL, '2026-01-01T00:00:00Z')" + ) + + # -- AC7 -------------------------------------------------------------- # + def test_noninstaller_null_issuer(self) -> None: # t_nonobstaller_null_issuer(-) + self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED) + with self.kernel.actor_context("op", "operator", "normal"): + cur = self.kernel._conn.execute( + "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')" + ) + class_id = cur.lastrowid + with self.assertRaises(sqlite3.IntegrityError) as ctx: + self.kernel._conn.execute( + "INSERT INTO principals" + "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) " + "VALUES ('rogue', 'operator', ?, NULL, NULL, '2026-01-01T00:00:00Z')", + (class_id,), + ) + self.assertIn("INVALID_BOOTSTRAP_STATE", str(ctx.exception)) + + def test_installer_null_issuer_only_during_install(self) -> None: + self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED) + with self.kernel.actor_context("i2", "installer", "install"): + cur = self.kernel._conn.execute( + "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')" + ) + class_id = cur.lastrowid + with self.assertRaises(sqlite3.IntegrityError): + self.kernel._conn.execute( + "INSERT INTO principals" + "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) " + "VALUES ('i2', 'installer', ?, NULL, NULL, '2026-01-01T00:00:00Z')", + (class_id,), + ) + + # -- AC8 -------------------------------------------------------------- # + def test_context_missing(self) -> None: # t_context_missing(-) + self.assertIsNone(self.kernel._ctx) + with self.assertRaises(sqlite3.IntegrityError) as ctx: + self.kernel._conn.execute( + "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')" + ) + self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception)) + + def test_context_stale(self) -> None: # t_context_stale(-) + with self.kernel.actor_context("op", "operator", "normal"): + self.kernel._ctx.expired = True + with self.assertRaises(sqlite3.IntegrityError) as ctx: + self.kernel._conn.execute( + "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')" + ) + self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception)) + + def test_context_epoch_shift(self) -> None: # t_context_epoch_shift(-) + with self.kernel.actor_context("op", "operator", "normal"): + self.kernel._ctx.live_epoch = self.kernel._ctx.bound_epoch + 99 + with self.assertRaises(sqlite3.IntegrityError) as ctx: + self.kernel._conn.execute( + "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')" + ) + self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception)) + + def test_bad_actor_kind_or_mode_rejected(self) -> None: + for kind, mode in (("intruder", "normal"), ("operator", "sabotage")): + with self.subTest(kind=kind, mode=mode): + with self.kernel.actor_context("op", kind, mode): + with self.assertRaises(sqlite3.IntegrityError): + self.kernel._conn.execute( + "INSERT INTO principal_equivalence_classes(created_at) " + "VALUES ('2026-01-01T00:00:00Z')" + ) + + # -- AC9 -------------------------------------------------------------- # + def test_bootstrap_immutable_update(self) -> None: # t_bootstrap_immutable_{update} + self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED) + cases = [ + ("UPDATE install_state SET installed_at = 'x' WHERE id = 1", "IMMUTABLE_INSTALL_STATE"), + ("UPDATE platform_bootstrap_seed SET created_at = 'x' WHERE seed_id = 1", "IMMUTABLE_SEED"), + ("UPDATE authority_dominance SET subordinate = 'x' WHERE dominant = 'supervisor.root'", "IMMUTABLE_DOMINANCE"), + (f"UPDATE authoritative_issuers SET issuer_ref = 'x' WHERE issuer_ref = '{DISTINGUISHED_ISSUER_ID}'", "IMMUTABLE_ISSUER"), + (f"UPDATE principals SET actor_kind = 'operator' WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"), + ] + for sql, tag in cases: + with self.subTest(sql=sql): + with self.kernel.actor_context("op", "operator", "normal"): + with self.assertRaises(sqlite3.IntegrityError) as ctx: + self.kernel._conn.execute(sql) + self.assertIn(tag, str(ctx.exception)) + + def test_bootstrap_immutable_delete(self) -> None: # t_bootstrap_immutable_{delete} + self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED) + cases = [ + ("DELETE FROM install_state WHERE id = 1", "IMMUTABLE_INSTALL_STATE"), + ("DELETE FROM platform_bootstrap_seed WHERE seed_id = 1", "IMMUTABLE_SEED"), + ("DELETE FROM authority_dominance", "IMMUTABLE_DOMINANCE"), + ("DELETE FROM authoritative_issuers", "IMMUTABLE_ISSUER"), + (f"DELETE FROM principals WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"), + ("DELETE FROM platform_bootstrap_grants", "IMMUTABLE_GRANT"), + ] + for sql, tag in cases: + with self.subTest(sql=sql): + with self.kernel.actor_context("op", "operator", "normal"): + with self.assertRaises(sqlite3.IntegrityError) as ctx: + self.kernel._conn.execute(sql) + self.assertIn(tag, str(ctx.exception)) + + def test_grant_reactivation_rejected(self) -> None: + self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED) + self.kernel.register_principal( + "op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER + ) + self.assertEqual( + self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED + ) + gid = self.kernel._conn.execute( + "SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'" + ).fetchone()[0] + self.assertEqual( + self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code, + INSTALLED, + ) + with self.kernel.actor_context("op", "operator", "normal"): + with self.assertRaises(sqlite3.IntegrityError) as ctx: + self.kernel._conn.execute( + "UPDATE platform_bootstrap_grants SET active = 1 WHERE grant_id = ?", + (gid,), + ) + self.assertIn("IMMUTABLE_GRANT", str(ctx.exception)) + + # -- AC12 ------------------------------------------------------------- # + def test_raw_write_bypass(self) -> None: # t_raw_write_bypass(raw-bypass) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "p.sqlite3") + k = PlatformKernel(path) + self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED) + k.close() + raw = sqlite3.connect(path) + raw.execute("PRAGMA foreign_keys = ON") + try: + with self.assertRaises(sqlite3.Error): + raw.execute( + "INSERT INTO audit_records(event, created_at) " + "VALUES ('forged', '2026-01-01T00:00:00Z')" + ) + raw.commit() + with self.assertRaises(sqlite3.Error): + raw.execute("UPDATE install_state SET installed_at = 'x' WHERE id = 1") + raw.commit() + with self.assertRaises(sqlite3.Error): + raw.execute("DELETE FROM platform_bootstrap_grants") + raw.commit() + finally: + raw.close() + + # -- AC13 ------------------------------------------------------------- # + def test_audit_created(self) -> None: # t_audit_created(+) + self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED) + self.kernel.register_principal( + "op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER + ) + self.assertEqual( + self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED + ) + gid = self.kernel._conn.execute( + "SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'" + ).fetchone()[0] + self.assertEqual( + self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code, + INSTALLED, + ) + events = self.kernel.audit_events() + for evt in ( + ap.EVT_PLATFORM_INSTALLED, + ap.EVT_GRANT_CREATED, + ap.EVT_GRANT_REVOKED, + ap.EVT_PRINCIPAL_REGISTERED, + ): + self.assertIn(evt, events) + + # -- AC14 ------------------------------------------------------------- # + def test_audit_immutable(self) -> None: # t_audit_immutable(raw-bypass) + self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED) + with self.kernel.actor_context("op", "operator", "normal"): + with self.assertRaises(sqlite3.IntegrityError) as up: + self.kernel._conn.execute("UPDATE audit_records SET event = 'x' WHERE audit_id = 1") + self.assertIn("IMMUTABLE_AUDIT", str(up.exception)) + with self.assertRaises(sqlite3.IntegrityError) as dl: + self.kernel._conn.execute("DELETE FROM audit_records WHERE audit_id = 1") + self.assertIn("IMMUTABLE_AUDIT", str(dl.exception)) + + # -- meta ------------------------------------------------------------- # + def test_schema_meta(self) -> None: + rows = dict(self.kernel._conn.execute("SELECT key, value FROM arch01_meta").fetchall()) + self.assertEqual(rows["schema_version"], str(ap.SCHEMA_VERSION)) + self.assertIn("disabled by default", rows["architecture"]) + + def test_register_principal_creates_class_first(self) -> None: + self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED) + res = self.kernel.register_principal( + "svc1", "service", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER + ) + self.assertEqual(res.code, INSTALLED) + row = self.kernel._conn.execute( + "SELECT current_class_id FROM principals WHERE principal_id = 'svc1'" + ).fetchone() + self.assertIsNotNone(row[0]) + + # -- helpers ---------------------------------------------------------- # + def _partial_bootstrap_then_rollback(self, k: PlatformKernel, stop: int) -> None: + """Execute the first ``stop`` bootstrap statements, then ROLLBACK.""" + now = "2026-01-01T00:00:00Z" + k._conn.execute("BEGIN IMMEDIATE") + class_id = None + issuer_id = None + try: + with k.actor_context(INSTALLER, "installer", "install"): + c = k._conn + if stop >= 1: + class_id = c.execute( + "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,) + ).lastrowid + if stop >= 2: + c.execute( + "INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) " + "VALUES (?, 'installer', ?, NULL, ?, ?)", + (INSTALLER, class_id, INSTALLER, now), + ) + if stop >= 3: + issuer_id = c.execute( + "INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)", + (DISTINGUISHED_ISSUER_ID, now), + ).lastrowid + if stop >= 4: + c.execute( + "UPDATE principals SET issuer_id = ? WHERE principal_id = ?", + (issuer_id, INSTALLER), + ) + if stop >= 5: + c.executemany( + "INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)", + DOMINANCE_TUPLES, + ) + if stop >= 6: + c.execute( + "INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)", + (INSTALLER, now), + ) + if stop >= 7: + c.execute( + "INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)", + (INSTALLER, now), + ) + if stop >= 8: + c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)") + finally: + k._conn.execute("ROLLBACK") + + def _seed_bootstrap_and_mark(self, k: PlatformKernel, dominance) -> str: + """Seed a full bootstrap with a caller-supplied dominance set, then + attempt the marker insert. Returns the classified failure code (or + INSTALLED). Rolls back on failure so no partial rows remain.""" + now = "2026-01-01T00:00:00Z" + k._conn.execute("BEGIN IMMEDIATE") + try: + with k.actor_context(INSTALLER, "installer", "install"): + c = k._conn + class_id = c.execute( + "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,) + ).lastrowid + c.execute( + "INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) " + "VALUES (?, 'installer', ?, NULL, ?, ?)", + (INSTALLER, class_id, INSTALLER, now), + ) + issuer_id = c.execute( + "INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)", + (DISTINGUISHED_ISSUER_ID, now), + ).lastrowid + c.execute( + "UPDATE principals SET issuer_id = ? WHERE principal_id = ?", + (issuer_id, INSTALLER), + ) + c.executemany( + "INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)", + dominance, + ) + c.execute( + "INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)", + (INSTALLER, now), + ) + c.execute( + "INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)", + (INSTALLER, now), + ) + c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)") + c.execute( + "INSERT INTO install_state(id, marker, installed_at) VALUES (1, 'installed', ?)", + (now,), + ) + k._conn.execute("COMMIT") + return INSTALLED + except sqlite3.Error as exc: + k._safe_rollback() + return PlatformKernel._classify(exc) + + +class Arch01ConcurrencyTest(unittest.TestCase): + """Concurrency invariants require file-backed DBs and independent connections.""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.path = os.path.join(self._tmp.name, "p.sqlite3") + + def tearDown(self) -> None: + self._tmp.cleanup() + + # -- AC10 ------------------------------------------------------------- # + def test_concurrent_install(self) -> None: # t_concurrent_install(concurrency) + k1 = PlatformKernel(self.path, busy_timeout_ms=0) + k2 = PlatformKernel(self.path, busy_timeout_ms=0) + barrier = threading.Barrier(2) + results = {} + + def _install(name, kernel): + barrier.wait() + results[name] = kernel.install_platform(INSTALLER).code + + try: + with ThreadPoolExecutor(max_workers=2) as ex: + f1 = ex.submit(_install, "a", k1) + f2 = ex.submit(_install, "b", k2) + f1.result() + f2.result() + codes = sorted(results.values()) + self.assertEqual(codes.count(INSTALLED), 1, f"exactly one install expected: {results}") + other = [c for c in results.values() if c != INSTALLED][0] + self.assertIn(other, (ALREADY_INSTALLED, CONCURRENT_INSTALLATION_LOST)) + self.assertTrue(k1.is_installed()) + self.assertEqual(_count(k1, "install_state"), 1) + self.assertEqual(_count(k1, "principals"), 1) + finally: + k1.close() + k2.close() + + # -- AC11 ------------------------------------------------------------- # + def test_concurrent_last_grant_revoke(self) -> None: # t_concurrent_last_grant_revoke + setup = PlatformKernel(self.path) + self.assertEqual(setup.install_platform(INSTALLER).code, INSTALLED) + setup.register_principal("op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER) + self.assertEqual(setup.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED) + self.assertEqual(setup.active_grant_count(), 2) + gids = [ + r[0] + for r in setup._conn.execute( + "SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1 ORDER BY grant_id" + ).fetchall() + ] + setup.close() + self.assertEqual(len(gids), 2) + + k1 = PlatformKernel(self.path, busy_timeout_ms=3000) + k2 = PlatformKernel(self.path, busy_timeout_ms=3000) + barrier = threading.Barrier(2) + results = {} + + def _revoke(name, kernel, gid): + barrier.wait() + results[name] = kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code + + try: + with ThreadPoolExecutor(max_workers=2) as ex: + f1 = ex.submit(_revoke, "a", k1, gids[0]) + f2 = ex.submit(_revoke, "b", k2, gids[1]) + f1.result() + f2.result() + codes = list(results.values()) + self.assertEqual(codes.count(INSTALLED), 1, f"exactly one revoke should win: {results}") + self.assertEqual(codes.count(AUTHORIZATION_DENIED), 1, f"one revoke must be denied: {results}") + self.assertEqual(k1.active_grant_count(), 1) + self.assertEqual(_count_where(k1, "platform_bootstrap_grants", "active = 1"), 1) + finally: + k1.close() + k2.close() + + def test_revoke_final_grant_denied(self) -> None: + k = PlatformKernel(self.path) + try: + self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED) + gid = k._conn.execute( + "SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1" + ).fetchone()[0] + res = k.revoke_platform_bootstrap(gid, actor_principal=INSTALLER) + self.assertEqual(res.code, AUTHORIZATION_DENIED) + self.assertEqual(k.active_grant_count(), 1) + self.assertEqual(_count_where(k, "platform_bootstrap_grants", "active = 1"), 1) + finally: + k.close() + + +if __name__ == "__main__": + unittest.main()