4e8fa6e831
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
236 lines
11 KiB
Python
236 lines
11 KiB
Python
"""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"},
|
|
}
|
|
|
|
|
|
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._held = False
|
|
self._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._held = False
|
|
fcntl.flock(self._lock_fd, fcntl.LOCK_UN)
|
|
|
|
def reserve(self, binding: Binding):
|
|
binding.validate()
|
|
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.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]
|
|
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()
|
|
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]
|
|
return {"state": row[3], "evidence_digest": row[4], "dispatch_allowed": False,
|
|
"external_outcome_unknown": bool(pending)}
|
|
|
|
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]
|
|
if 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 close(self):
|
|
if hasattr(self, "db"):
|
|
self.db.close()
|
|
if hasattr(self, "_lock_fd"):
|
|
os.close(self._lock_fd)
|