"""Source-only admission primitive; never authorizes or dispatches native work. Trusted owner mediation must validate claim/PDP/consume before transitions. One approval admits one operation. A persisted reservation cannot be resumed. The process lock is held across the whole owner invocation, including cleanup. """ from __future__ import annotations import fcntl import os import re import sqlite3 import stat import uuid from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path class Refused(Exception): pass @dataclass(frozen=True) class Binding: approval_id: str request_digest: str operation_id: str plan_digest: str def validate(self): if any(type(v) is not str for v in (self.approval_id, self.request_digest, self.operation_id, self.plan_digest)): raise Refused("binding fields must be strings") if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}", self.approval_id): raise Refused("invalid approval identifier") # Preserve the exact served PDP/Approval value; never silently strip it. if not re.fullmatch(r"sha256:[0-9a-f]{64}", self.request_digest): raise Refused("invalid canonical request digest") if not re.fullmatch(r"[0-9a-f]{64}", self.plan_digest): raise Refused("invalid owner plan content digest") try: if str(uuid.UUID(self.operation_id)) != self.operation_id: raise ValueError() except (ValueError, AttributeError): raise Refused("invalid operation identifier") from None _NEXT = { "reserved": {"consume-confirmed", "failed", "unknown"}, "consume-confirmed": {"authenticating", "failed", "unknown"}, "authenticating": {"grant-issuing", "failed", "unknown"}, "grant-issuing": {"grant-issued", "failed", "unknown"}, "grant-issued": {"applying", "failed", "unknown"}, "applying": {"readback", "failed", "unknown"}, "readback": {"verified", "failed", "unknown"}, "verified": {"cleanup-confirmed", "unknown"}, } @dataclass(frozen=True) class Cleanup: """Owner-derived exact call commitment, never a handle or authority token. The owner mediator must include these commitments in its approved plan before reservation, then compare the actual native cleanup call immediately before dispatch. This journal does not authenticate that plan or native adapter. """ kind: str call_digest: str def validate(self): if type(self.kind) is not str or self.kind not in { "task-token-revoke", "identity-token-revoke", "grant-policy-delete" }: raise Refused("unsupported cleanup obligation") if type(self.call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", self.call_digest): raise Refused("invalid cleanup call digest") class Journal: def __init__(self, directory: Path): # Require an existing owner-private directory: deployment creates it. if directory.is_symlink() or directory.resolve() != directory.absolute(): raise Refused("journal path is indirect") self.directory = directory self._private(directory.stat(), directory=True) self._lock_fd = self._open("execution.lock") try: fd = self._open("journal.sqlite") except BaseException: os.close(self._lock_fd) del self._lock_fd raise original = os.fstat(fd) os.close(fd) self.db = sqlite3.connect(directory / "journal.sqlite", isolation_level=None) current = (directory / "journal.sqlite").stat() if (original.st_dev, original.st_ino) != (current.st_dev, current.st_ino): self.close() raise Refused("journal changed during open") self.db.execute("PRAGMA synchronous=FULL") self.db.execute("PRAGMA journal_mode=DELETE") self.db.execute("CREATE TABLE IF NOT EXISTS operations (approval TEXT PRIMARY KEY, request TEXT NOT NULL, operation TEXT UNIQUE NOT NULL, plan TEXT NOT NULL, state TEXT NOT NULL, evidence TEXT)") self.db.execute("CREATE TABLE IF NOT EXISTS calls (operation TEXT NOT NULL, ordinal INTEGER NOT NULL, call_digest TEXT NOT NULL, outcome TEXT NOT NULL, evidence TEXT, PRIMARY KEY(operation,ordinal))") self.db.execute("CREATE TABLE IF NOT EXISTS cleanup (operation TEXT NOT NULL, ordinal INTEGER NOT NULL, kind TEXT NOT NULL, call_digest TEXT NOT NULL, outcome TEXT NOT NULL, evidence TEXT, PRIMARY KEY(operation,ordinal), UNIQUE(operation,kind))") self.db.execute("CREATE TABLE IF NOT EXISTS cleanup_calls (operation TEXT NOT NULL, obligation INTEGER NOT NULL, step INTEGER NOT NULL, call_digest TEXT NOT NULL, outcome TEXT NOT NULL, evidence TEXT, PRIMARY KEY(operation,obligation,step))") self.db.execute("CREATE TABLE IF NOT EXISTS recovery_attempts (attempt TEXT PRIMARY KEY, operation TEXT NOT NULL, original TEXT NOT NULL)") self.db.execute("CREATE TABLE IF NOT EXISTS recovery_calls (attempt TEXT NOT NULL, obligation INTEGER NOT NULL, step INTEGER NOT NULL, digest TEXT NOT NULL, outcome TEXT NOT NULL, evidence TEXT, PRIMARY KEY(attempt,obligation,step))") self.db.execute("CREATE TABLE IF NOT EXISTS recovery_results (attempt TEXT NOT NULL, obligation INTEGER NOT NULL, confirmed INTEGER NOT NULL, evidence TEXT NOT NULL, PRIMARY KEY(attempt,obligation))") self._held = False self._admission = None self._consumed_admission = None self._sync_directory() @staticmethod def _private(s, directory=False): kind = stat.S_ISDIR(s.st_mode) if directory else stat.S_ISREG(s.st_mode) if not kind or s.st_uid != os.getuid() or s.st_mode & 0o077: raise Refused("journal is not owner private") if not directory and s.st_nlink != 1: raise Refused("journal file has multiple links") def _open(self, name): fd = os.open(self.directory / name, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) try: self._private(os.fstat(fd)) except BaseException: os.close(fd) raise return fd def _sync_directory(self): fd = os.open(self.directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: os.fsync(fd) finally: os.close(fd) @contextmanager def locked(self): if self._held: raise Refused("nested journal admission") try: fcntl.flock(self._lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise Refused("owner invocation already active") from None self._held = True try: yield self finally: self._admission = None self._consumed_admission = None self._held = False fcntl.flock(self._lock_fd, fcntl.LOCK_UN) def reserve(self, binding: Binding, cleanup: tuple[Cleanup, ...] = ()): binding.validate() if type(cleanup) is not tuple or len(cleanup) > 3: raise Refused("closed cleanup plan required") for item in cleanup: if type(item) is not Cleanup: raise Refused("typed cleanup obligation required") item.validate() if len({item.kind for item in cleanup}) != len(cleanup): raise Refused("duplicate cleanup obligation") if not self._held or self._admission is not None: raise Refused("reservation requires fresh exclusive invocation") try: self.db.execute("BEGIN IMMEDIATE") self.db.execute("INSERT INTO operations VALUES (?,?,?,?,?,NULL)", (binding.approval_id, binding.request_digest, binding.operation_id, binding.plan_digest, "reserved")) self.db.executemany("INSERT INTO cleanup VALUES (?,?,?,?,'owed',NULL)", [(binding.operation_id, i, item.kind, item.call_digest) for i, item in enumerate(cleanup)]) self.db.execute("COMMIT") self._sync_directory() except sqlite3.IntegrityError: self.db.execute("ROLLBACK") raise Refused("approval or operation already admitted") from None except BaseException: if self.db.in_transaction: self.db.execute("ROLLBACK") raise self._admission = binding def transition(self, binding: Binding, expected: str, target: str, evidence_digest=None): if not self._held or self._admission != binding: raise Refused("no live admission") if target not in _NEXT.get(expected, set()): raise Refused("invalid state transition") if target == "cleanup-confirmed" and evidence_digest is None: raise Refused("cleanup requires evidence") if evidence_digest is not None and not re.fullmatch(r"[0-9a-f]{64}", evidence_digest): raise Refused("invalid evidence digest") try: self.db.execute("BEGIN IMMEDIATE") pending=self.db.execute("SELECT COUNT(*) FROM calls WHERE operation=? AND outcome='pending'",(binding.operation_id,)).fetchone()[0] started = self.db.execute("SELECT COUNT(*) FROM cleanup WHERE operation=? AND outcome!='owed'", (binding.operation_id,)).fetchone()[0] if started and target not in {'unknown', 'cleanup-confirmed'}: raise Refused("cleanup has closed business execution") if target == 'cleanup-confirmed': total, owed = self.db.execute("SELECT COUNT(*),COALESCE(SUM(outcome!='confirmed'),0) FROM cleanup WHERE operation=?", (binding.operation_id,)).fetchone() if not total or owed: raise Refused("explicit cleanup obligations are unconfirmed") if pending and target != 'unknown': raise Refused("unconfirmed external call requires unknown") changed = self.db.execute( "UPDATE operations SET state=?, evidence=? WHERE approval=? AND request=? AND operation=? AND plan=? AND state=?", (target, evidence_digest, binding.approval_id, binding.request_digest, binding.operation_id, binding.plan_digest, expected)).rowcount if changed != 1: raise Refused("journal state changed") self.db.execute("COMMIT") self._sync_directory() if target == "consume-confirmed": self._consumed_admission = binding except BaseException: if self.db.in_transaction: self.db.execute("ROLLBACK") raise def observe(self, binding: Binding): binding.validate() row = self.db.execute("SELECT request,operation,plan,state,evidence FROM operations WHERE approval=?", (binding.approval_id,)).fetchone() if row is None: return None if row[:3] != (binding.request_digest, binding.operation_id, binding.plan_digest): raise Refused("journal binding mismatch") # Readback cannot create a live admission, even for an interrupted state. pending = self.db.execute("SELECT COUNT(*) FROM calls WHERE operation=? AND outcome='pending'", (binding.operation_id,)).fetchone()[0] pending += self.db.execute("SELECT COUNT(*) FROM cleanup WHERE operation=? AND outcome='pending'", (binding.operation_id,)).fetchone()[0] pending += self.db.execute("SELECT COUNT(*) FROM cleanup_calls WHERE operation=? AND outcome='pending'", (binding.operation_id,)).fetchone()[0] pending += self.db.execute("SELECT COUNT(*) FROM recovery_calls AS c JOIN recovery_attempts AS a ON c.attempt=a.attempt WHERE a.operation=? AND c.outcome='pending'", (binding.operation_id,)).fetchone()[0] return {"state": row[3], "evidence_digest": row[4], "dispatch_allowed": False, "external_outcome_unknown": bool(pending)} def next_call_ordinal(self, binding: Binding): """Continue the same admitted invocation after owner auth/consume calls. This never grants dispatch or resumes a persisted admission. The actual before_call transaction remains the final serialization/intent guard. """ if not self._held or self._admission != binding: raise Refused("next call requires same live admission") count,pending=self.db.execute("SELECT COUNT(*),COALESCE(SUM(outcome='pending'),0) FROM calls WHERE operation=?",(binding.operation_id,)).fetchone() cleanup=self.db.execute("SELECT COUNT(*) FROM cleanup WHERE operation=? AND outcome!='owed'",(binding.operation_id,)).fetchone()[0] state=self.db.execute("SELECT state FROM operations WHERE operation=?",(binding.operation_id,)).fetchone()[0] if pending or cleanup or state in {'failed','unknown','cleanup-confirmed'} or count>63: raise Refused("call sequence is uncertain, closed or exhausted") return count def before_call(self, binding: Binding, ordinal: int, call_digest: str): """Persist one exact plan-call intent before the mediator issues it. Caller must derive ordinal/digest from the independently pinned owner plan. This primitive doesn't validate method/body or decide whether a call is allowed. """ if not self._held or self._admission != binding: raise Refused("no live admission") if type(ordinal) is not int or ordinal < 0 or ordinal > 63: raise Refused("invalid fixed-plan ordinal") if type(call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", call_digest): raise Refused("invalid call digest") try: self.db.execute("BEGIN IMMEDIATE") count,pending = self.db.execute("SELECT COUNT(*), COALESCE(SUM(outcome='pending'),0) FROM calls WHERE operation=?",(binding.operation_id,)).fetchone() state=self.db.execute("SELECT state FROM operations WHERE operation=?",(binding.operation_id,)).fetchone()[0] cleanup_started = self.db.execute("SELECT COUNT(*) FROM cleanup WHERE operation=? AND outcome!='owed'", (binding.operation_id,)).fetchone()[0] if cleanup_started or ordinal != count or pending or state in {'failed','unknown','cleanup-confirmed'}: raise Refused("call is replayed, uncertain or terminal") self.db.execute("INSERT INTO calls VALUES (?,?,?,'pending',NULL)",(binding.operation_id,ordinal,call_digest)) self.db.execute("COMMIT") self._sync_directory() except BaseException: if self.db.in_transaction: self.db.execute("ROLLBACK") raise def after_call(self, binding: Binding, ordinal: int, call_digest: str, evidence_digest: str): if not self._held or self._admission != binding: raise Refused("no live admission") if type(ordinal) is not int or ordinal < 0 or ordinal > 63: raise Refused("invalid fixed-plan ordinal") if type(call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", call_digest): raise Refused("invalid call digest") if type(evidence_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", evidence_digest): raise Refused("invalid evidence digest") try: self.db.execute("BEGIN IMMEDIATE") n=self.db.execute("UPDATE calls SET outcome='complete', evidence=? WHERE operation=? AND ordinal=? AND call_digest=? AND outcome='pending'",(evidence_digest,binding.operation_id,ordinal,call_digest)).rowcount if n != 1: raise Refused("call confirmation mismatch") self.db.execute("COMMIT") self._sync_directory() except BaseException: if self.db.in_transaction: self.db.execute("ROLLBACK") raise def cleanup_status(self, binding: Binding): # Observation never restores live execution or cleanup authority. if self.observe(binding) is None: raise Refused("operation is not admitted") rows = self.db.execute("SELECT ordinal,kind,call_digest,outcome,evidence FROM cleanup WHERE operation=? ORDER BY ordinal", (binding.operation_id,)).fetchall() return {"confirmed": bool(rows) and all(row[3] == "confirmed" for row in rows), "obligations": [{"ordinal": row[0], "kind": row[1], "call_digest": row[2], "outcome": row[3], "evidence_digest": row[4]} for row in rows]} def _cleanup_admission(self, binding, ordinal, obligation): if not self._held or self._admission != binding or self._consumed_admission != binding: raise Refused("cleanup requires same live consumed admission") if type(ordinal) is not int or not 0 <= ordinal < 3 or type(obligation) is not Cleanup: raise Refused("exact cleanup obligation required") obligation.validate() def before_cleanup(self, binding: Binding, ordinal: int, obligation: Cleanup): """Commit intent for one pre-admitted cleanup call, even after unknown. Never resumes an interrupted invocation. Independent owner recovery after a crash still requires the owner's protected recovery channel/approval. Cleanup has no business deadline: its authority is the original separate explicit obligation, and cannot authorize a read, repair or business write. """ self._cleanup_admission(binding, ordinal, obligation) try: self.db.execute("BEGIN IMMEDIATE") changed = self.db.execute("UPDATE cleanup SET outcome='pending' WHERE operation=? AND ordinal=? AND kind=? AND call_digest=? AND outcome='owed'", (binding.operation_id, ordinal, obligation.kind, obligation.call_digest)).rowcount if changed != 1: raise Refused("cleanup is unadmitted, replayed or uncertain") self.db.execute("COMMIT") self._sync_directory() except BaseException: if self.db.in_transaction: self.db.execute("ROLLBACK") raise def after_cleanup(self, binding: Binding, ordinal: int, obligation: Cleanup, *, confirmed: bool, evidence_digest: str): self._cleanup_admission(binding, ordinal, obligation) if type(confirmed) is not bool or type(evidence_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", evidence_digest): raise Refused("closed cleanup evidence required") try: self.db.execute("BEGIN IMMEDIATE") if confirmed and self.db.execute("SELECT COUNT(*) FROM cleanup_calls WHERE operation=? AND obligation=? AND outcome='pending'", (binding.operation_id,ordinal)).fetchone()[0]: raise Refused("pending cleanup native call cannot confirm") changed = self.db.execute("UPDATE cleanup SET outcome=?, evidence=? WHERE operation=? AND ordinal=? AND kind=? AND call_digest=? AND outcome='pending'", ('confirmed' if confirmed else 'unconfirmed', evidence_digest, binding.operation_id, ordinal, obligation.kind, obligation.call_digest)).rowcount if changed != 1: raise Refused("cleanup confirmation mismatch") self.db.execute("COMMIT") self._sync_directory() except BaseException: if self.db.in_transaction: self.db.execute("ROLLBACK") raise def before_cleanup_call(self, binding, ordinal, obligation, step, call_digest): """Journal each native revoke/delete AND its independent verification.""" self._cleanup_admission(binding,ordinal,obligation) if type(step) is not int or not 0<=step<2 or type(call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}",call_digest): raise Refused("exact bounded cleanup native call required") try: self.db.execute("BEGIN IMMEDIATE") parent=self.db.execute("SELECT kind,call_digest,outcome FROM cleanup WHERE operation=? AND ordinal=?",(binding.operation_id,ordinal)).fetchone() count,pending=self.db.execute("SELECT COUNT(*),COALESCE(SUM(outcome='pending'),0) FROM cleanup_calls WHERE operation=? AND obligation=?",(binding.operation_id,ordinal)).fetchone() if parent!=(obligation.kind,obligation.call_digest,'pending') or step!=count or pending: raise Refused("cleanup call unadmitted, replayed or uncertain") self.db.execute("INSERT INTO cleanup_calls VALUES (?,?,?,?,'pending',NULL)",(binding.operation_id,ordinal,step,call_digest)) self.db.execute("COMMIT");self._sync_directory() except BaseException: if self.db.in_transaction:self.db.execute("ROLLBACK") raise def after_cleanup_call(self,binding,ordinal,obligation,step,call_digest,evidence_digest): self._cleanup_admission(binding,ordinal,obligation) if (type(step) is not int or not 0<=step<2 or type(call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}",call_digest) or type(evidence_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}",evidence_digest)): raise Refused("exact cleanup native evidence required") try: self.db.execute("BEGIN IMMEDIATE") parent=self.db.execute("SELECT kind,call_digest,outcome FROM cleanup WHERE operation=? AND ordinal=?",(binding.operation_id,ordinal)).fetchone() if parent!=(obligation.kind,obligation.call_digest,'pending'): raise Refused("cleanup parent is no longer pending") changed=self.db.execute("UPDATE cleanup_calls SET outcome='complete',evidence=? WHERE operation=? AND obligation=? AND step=? AND call_digest=? AND outcome='pending'", (evidence_digest,binding.operation_id,ordinal,step,call_digest)).rowcount if changed!=1:raise Refused("cleanup native confirmation mismatch") self.db.execute("COMMIT");self._sync_directory() except BaseException: if self.db.in_transaction:self.db.execute("ROLLBACK") raise def close(self): if hasattr(self, "db"): self.db.close() if hasattr(self, "_lock_fd"): os.close(self._lock_fd)