tegwick
70f408ca23
Some checks failed
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
253 lines
13 KiB
Python
253 lines
13 KiB
Python
"""Injection-only PostgreSQL implementation of the existing owner journal seam.
|
|
|
|
Binding and continuity callbacks are supplied by a trusted owner integration,
|
|
never constructed from an HTTP payload. Neither is an authorization decision.
|
|
No server environment switch, automatic schema creation or recovery admission.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from dataclasses import asdict, dataclass
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from typing import Any, Callable, Protocol
|
|
from uuid import UUID
|
|
|
|
from group_provenance import GroupProvenance, GroupProvenanceUnavailable
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GroupJournalBinding:
|
|
tenant_scope: str
|
|
directory: str
|
|
case_digest: str
|
|
source_digest: str
|
|
principal: str
|
|
epoch: str
|
|
|
|
def __post_init__(self):
|
|
for value in (self.tenant_scope, self.directory, self.principal):
|
|
if not isinstance(value, str) or not 1 <= len(value) <= 256 or any(ord(c) < 32 for c in value):
|
|
raise ValueError("explicit owner journal binding required")
|
|
for value in (self.case_digest, self.source_digest):
|
|
if not isinstance(value, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", value):
|
|
raise ValueError("exact owner source and case digests required")
|
|
if not isinstance(self.epoch, str) or str(UUID(self.epoch)) != self.epoch:
|
|
raise ValueError("explicit canonical journal epoch required")
|
|
|
|
@property
|
|
def digest(self):
|
|
return hashlib.sha256(json.dumps(asdict(self), sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
|
|
|
|
def journal_genesis(epoch: str, directory: str) -> str:
|
|
"""Deterministic initial marker for explicit owner migration/admission only."""
|
|
return hashlib.sha256(json.dumps({"schema": "nk-group-journal-genesis.v1", "epoch": epoch,
|
|
"directory": directory}, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
|
|
|
|
class JournalContinuity(Protocol):
|
|
"""Existing independently retained owner checkpoint, outside restored SQL state.
|
|
|
|
Implementations compare/advance exact epoch+directory+head+commit digest atomically and
|
|
durably. An in-memory implementation is a fixture only. Missing/stale or
|
|
ambiguous checkpoints must refuse; no callback grants mutation authority.
|
|
"""
|
|
def check(self, epoch: str, directory: str, head: int, digest: str) -> bool: ...
|
|
def advance(self, epoch: str, directory: str, previous: int, previous_digest: str, current: int, current_digest: str) -> bool: ...
|
|
|
|
|
|
class PostgresGroupProvenance:
|
|
def __init__(self, *, connect: Callable, binding: GroupJournalBinding,
|
|
continuity: JournalContinuity, capacity: int = 256):
|
|
if not callable(connect) or not isinstance(binding, GroupJournalBinding) or not 1 <= capacity <= 4096:
|
|
raise ValueError("explicit owner connector, binding and bounded capacity required")
|
|
if not callable(getattr(continuity, "check", None)) or not callable(getattr(continuity, "advance", None)):
|
|
raise ValueError("independent owner continuity callbacks required")
|
|
self.connect, self.binding, self.continuity, self.capacity = connect, binding, continuity, capacity
|
|
self._poisoned = False
|
|
|
|
def _continuity(self, method, *args):
|
|
try:
|
|
if getattr(self.continuity, method)(self.binding.epoch, self.binding.directory, *args) is not True:
|
|
raise GroupProvenanceUnavailable()
|
|
except Exception:
|
|
raise GroupProvenanceUnavailable() from None
|
|
|
|
@contextmanager
|
|
def _transaction(self):
|
|
if self._poisoned:
|
|
raise GroupProvenanceUnavailable()
|
|
connection = None
|
|
tx = None
|
|
try:
|
|
connection = self.connect()
|
|
with connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SET LOCAL statement_timeout = '5s'")
|
|
cursor.execute("SET LOCAL idle_in_transaction_session_timeout = '15s'")
|
|
cursor.execute("SELECT epoch::text, directory, head, head_digest, quarantined FROM identity_provisioner.group_journal_meta WHERE singleton = true FOR UPDATE")
|
|
meta = cursor.fetchone()
|
|
if not meta or meta[0] != self.binding.epoch or meta[1] != self.binding.directory or meta[4] is not False or type(meta[2]) is not int or meta[2] < 0 or not isinstance(meta[3], str) or not re.fullmatch(r"[0-9a-f]{64}", meta[3]):
|
|
raise GroupProvenanceUnavailable()
|
|
self._continuity("check", meta[2], meta[3])
|
|
tx = {"cursor": cursor, "head": meta[2], "digest": meta[3], "dirty": False}
|
|
yield tx
|
|
if tx["dirty"]:
|
|
cursor.execute("UPDATE identity_provisioner.group_journal_meta SET head = head + 1, head_digest = %s WHERE singleton = true AND epoch = %s AND directory = %s AND head = %s AND head_digest = %s AND quarantined = false",
|
|
(tx["next_digest"], self.binding.epoch, self.binding.directory, tx["head"], tx["digest"]))
|
|
if cursor.rowcount != 1:
|
|
raise GroupProvenanceUnavailable()
|
|
# SQL commit precedes the independent checkpoint acknowledgment.
|
|
# A lost acknowledgment refuses before the next directory effect.
|
|
if tx and tx["dirty"]:
|
|
self._continuity("advance", tx["head"], tx["digest"], tx["head"] + 1, tx["next_digest"])
|
|
except ValueError:
|
|
raise
|
|
except GroupProvenanceUnavailable:
|
|
# Caller mismatches are domain refusals. Storage/continuity failures
|
|
# poison this object; durable pending records are never rewritten
|
|
# automatically by an exception handler after ambiguous commit.
|
|
self._poisoned = True
|
|
raise
|
|
except Exception:
|
|
self._poisoned = True
|
|
raise GroupProvenanceUnavailable() from None
|
|
finally:
|
|
if connection is not None:
|
|
try:
|
|
connection.close()
|
|
except Exception:
|
|
self._poisoned = True
|
|
raise GroupProvenanceUnavailable() from None
|
|
|
|
def check(self):
|
|
with self._transaction():
|
|
pass
|
|
|
|
def _read(self, cursor, key, request_binding):
|
|
cursor.execute("SELECT binding, version, body, context_json FROM identity_provisioner.group_receipts WHERE context_digest = %s AND key = %s",
|
|
(self.binding.digest, key))
|
|
row = cursor.fetchone()
|
|
if row is None:
|
|
return None
|
|
if row[0] != request_binding:
|
|
raise ValueError("ensure receipt request binding mismatch")
|
|
if row[3] != json.dumps(asdict(self.binding), sort_keys=True):
|
|
raise GroupProvenanceUnavailable()
|
|
receipt = GroupProvenance._load(row[2], key, request_binding)
|
|
if type(row[1]) is not int or row[1] < 1 or type(receipt.get("_pg_version")) is not int or receipt.get("_pg_version") != row[1] or receipt.get("_pg_context") != self.binding.digest:
|
|
raise GroupProvenanceUnavailable()
|
|
return receipt
|
|
|
|
@staticmethod
|
|
def _inputs(key, binding):
|
|
if not isinstance(key, str) or not 1 <= len(key) <= 256 or not isinstance(binding, str) or not re.fullmatch(r"[0-9a-f]{64}", binding):
|
|
raise ValueError("bounded exact journal request binding required")
|
|
|
|
def _active(self, cursor):
|
|
cursor.execute("SELECT 1 FROM identity_provisioner.group_receipts WHERE active = true LIMIT 1")
|
|
if cursor.fetchone():
|
|
raise GroupProvenanceUnavailable()
|
|
|
|
def _transition(self, tx, receipt, operation):
|
|
# Bind the exact transition, not merely a counter. Two restored/forked
|
|
# copies that commit different bodies at the same next head cannot both
|
|
# match an independently retained checkpoint after reopening.
|
|
transition = {"schema": "nk-group-journal-transition.v1", "epoch": self.binding.epoch,
|
|
"directory": self.binding.directory, "prior_head": tx["head"], "prior_digest": tx["digest"],
|
|
"head": tx["head"] + 1, "operation": operation, "context": asdict(self.binding),
|
|
"key": receipt["key"], "request_binding": receipt["binding"],
|
|
"version": receipt["_pg_version"], "body": receipt}
|
|
tx["next_digest"] = hashlib.sha256(json.dumps(transition, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
tx["dirty"] = True
|
|
|
|
def reserve(self, key, binding):
|
|
self._inputs(key, binding)
|
|
with self._transaction() as tx:
|
|
cursor = tx["cursor"]
|
|
receipt = self._read(cursor, key, binding)
|
|
if receipt is not None:
|
|
return receipt, False
|
|
self._active(cursor)
|
|
cursor.execute("SELECT COUNT(*) FROM identity_provisioner.group_receipts")
|
|
if cursor.fetchone()[0] >= self.capacity:
|
|
raise GroupProvenanceUnavailable()
|
|
receipt = {"key": key, "binding": binding, "state": "pending", "intent": None, "groups": {},
|
|
"created_groups": [], "created_memberships": [], "_pg_version": 1, "_pg_context": self.binding.digest}
|
|
cursor.execute("INSERT INTO identity_provisioner.group_receipts (context_digest, context_json, key, binding, version, state, active, body) VALUES (%s, %s, %s, %s, 1, 'pending', true, %s)",
|
|
(self.binding.digest, json.dumps(asdict(self.binding), sort_keys=True), key, binding, json.dumps(receipt, sort_keys=True)))
|
|
if cursor.rowcount != 1:
|
|
raise GroupProvenanceUnavailable()
|
|
self._transition(tx, receipt, "reserve")
|
|
return receipt, True
|
|
|
|
def get(self, key, binding):
|
|
self._inputs(key, binding)
|
|
with self._transaction() as tx:
|
|
receipt = self._read(tx["cursor"], key, binding)
|
|
if receipt is None:
|
|
raise ValueError("unknown ensure receipt or request binding mismatch")
|
|
return receipt
|
|
|
|
def _write(self, tx, receipt, operation="save"):
|
|
self._inputs(receipt.get("key"), receipt.get("binding"))
|
|
if receipt.get("_pg_context") != self.binding.digest or type(receipt.get("_pg_version")) is not int or receipt["_pg_version"] < 1:
|
|
raise GroupProvenanceUnavailable()
|
|
encoded = json.dumps(receipt, sort_keys=True)
|
|
GroupProvenance._load(encoded, receipt["key"], receipt["binding"])
|
|
updated = json.loads(encoded)
|
|
updated["_pg_version"] += 1
|
|
active = True if receipt["state"] in {"pending", "revoking", "uncertain"} else None
|
|
tx["cursor"].execute("UPDATE identity_provisioner.group_receipts SET state = %s, active = %s, body = %s, version = version + 1 WHERE context_digest = %s AND key = %s AND binding = %s AND version = %s",
|
|
(receipt["state"], active, json.dumps(updated, sort_keys=True), self.binding.digest, receipt["key"], receipt["binding"], receipt["_pg_version"]))
|
|
if tx["cursor"].rowcount != 1:
|
|
raise GroupProvenanceUnavailable()
|
|
self._transition(tx, updated, operation)
|
|
return updated
|
|
|
|
def save(self, receipt):
|
|
with self._transaction() as tx:
|
|
updated = self._write(tx, receipt)
|
|
# The provisioner keeps references to nested group pre/poststate across
|
|
# intent persistence. Replace only this backend's acknowledged version;
|
|
# _write serializes without changing the caller's mutation body.
|
|
receipt["_pg_version"] = updated["_pg_version"]
|
|
|
|
def claim_revoke(self, key, binding, delete_if_empty):
|
|
self._inputs(key, binding)
|
|
if not isinstance(delete_if_empty, bool):
|
|
raise ValueError("delete_if_empty must be boolean")
|
|
with self._transaction() as tx:
|
|
receipt = self._read(tx["cursor"], key, binding)
|
|
if receipt is None:
|
|
raise ValueError("unknown ensure receipt or request binding mismatch")
|
|
if receipt["state"] == "revoked":
|
|
if receipt["delete_if_empty"] != delete_if_empty:
|
|
raise ValueError("completed rollback options cannot change")
|
|
return receipt
|
|
if receipt["state"] != "ensured":
|
|
raise GroupProvenanceUnavailable()
|
|
self._active(tx["cursor"])
|
|
receipt.update(state="revoking", delete_if_empty=delete_if_empty)
|
|
updated = self._write(tx, receipt, "claim_revoke")
|
|
return updated
|
|
|
|
|
|
def psycopg_connector(*, connection_info: Callable[[], dict[str, Any]]):
|
|
"""Owner-provided renewable lease values stay in memory; never env/argv/logs.
|
|
|
|
No connection happens here. Lazy import keeps disabled group endpoints free
|
|
of a runtime dependency. The admitted owner supplies TLS/host/db/role facts.
|
|
"""
|
|
if not callable(connection_info):
|
|
raise ValueError("owner connection-info callback required")
|
|
def connect():
|
|
try:
|
|
import psycopg
|
|
values = dict(connection_info())
|
|
return psycopg.connect(**values, connect_timeout=5)
|
|
except Exception:
|
|
raise GroupProvenanceUnavailable() from None
|
|
return connect
|