"""Durable sender-change protocol; adapters must supply native fencing and reads. This module has no CLI or cluster credentials. The native adapter must bind a reviewed proposal, hold the existing capture lock and verify owner admission. No protocol method writes or restores the delivery ledger. """ from __future__ import annotations import copy import fcntl import hashlib import json import math import os import sqlite3 import stat from contextlib import closing from datetime import datetime from pathlib import Path IMAGE = ("forgejo.coulomb.social/coulomb/email-connect@sha256:" "5d79fd1fec15d3a279383820f170d263de3a55f5308df01a0a34732ebaea6d3d") TABLES = {"deliveries", "verifications", "suppressions", "send_attempts", "verification_attempts", "caller_buckets"} SCHEMA = "email-connect.sender-transition.v1" class Refused(RuntimeError): pass def require(condition, reason): if not condition: raise Refused(reason) def digest(value): return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def desired(before): """The only admitted configuration delta; never accepts a free-form patch.""" after = copy.deepcopy(before) spec = after["deployment_spec"] containers = spec["template"]["spec"]["containers"] require(spec["replicas"] == 1 and len(containers) == 1, "unexpected_receiver_shape") container = containers[0] require(container["name"] == "email-connect" and container["image"] == IMAGE, "unexpected_receiver_image") require(container["envFrom"] == [ {"configMapRef": {"name": "email-connect-config"}}, {"secretRef": {"name": "email-connect-runtime", "optional": False}}, ], "unexpected_old_environment") require(container["env"] == [ {"name": "EMAIL_CONNECT_CALLERS_FILE", "value": "/etc/email-connect-callers/callers.json"}, ], "unexpected_environment_override") require(after["config_data"]["EMAIL_CONNECT_SENDER"] == "forgejo@coulomb.social", "unexpected_old_sender") container["envFrom"] = [{"configMapRef": {"name": "email-connect-config"}}] container["env"] += [ {"name": "EMAIL_CONNECT_SMTP_USERNAME", "value": "platform@coulomb.social"}, {"name": "EMAIL_CONNECT_SMTP_PASSWORD", "valueFrom": {"secretKeyRef": { "name": "email-connect-platform-smtp", "key": "password", "optional": False}}}, ] after["config_data"]["EMAIL_CONNECT_SENDER"] = "platform@coulomb.social" return after def with_replicas(spec, replicas): result = copy.deepcopy(spec) result["replicas"] = replicas return result def known(observed, before, after): """Recognize lost-response/mixed phases, but never unrelated live drift.""" for key in ("identity", "protected"): require(observed[key] == before[key] == after[key], "protected_runtime_changed") require(observed["config_data"] in (before["config_data"], after["config_data"]), "configuration_changed_outside_proposal") require(observed["deployment_spec"] in [with_replicas(item["deployment_spec"], replicas) for item in (before, after) for replicas in (0, 1)], "deployment_changed_outside_proposal") def ledger(path): path = Path(path).absolute() require(not any(p.is_symlink() for p in [path, *path.parents]), "symlink_snapshot") info = path.stat() require(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid() and not info.st_mode & 0o077 and 0 < info.st_size <= 128 * 1024**2, "private_bounded_snapshot_required") with closing(sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)) as db: db.execute("PRAGMA trusted_schema=OFF") db.execute("PRAGMA query_only=ON") db.execute("BEGIN") require(db.execute("PRAGMA integrity_check").fetchall() == [("ok",)], "snapshot_integrity_failed") schema = db.execute( "SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name").fetchall() require({r[1] for r in schema if r[0] == "table" and not r[1].startswith("sqlite_")} == TABLES, "unexpected_snapshot_schema") rows = {} for table in sorted(TABLES): columns = db.execute('PRAGMA table_info("' + table + '")').fetchall() keys = [r[1] for r in sorted(columns, key=lambda r: r[5]) if r[5]] require(len(keys) == 1, "unexpected_snapshot_key") names = [r[1] for r in columns] values = [dict(zip(names, r, strict=True)) for r in db.execute('SELECT * FROM "' + table + '"')] rows[table] = {r[keys[0]]: r for r in values} require(len(rows[table]) == len(values) and None not in rows[table], "ambiguous_snapshot_row_identity") return schema, rows def same_except(old, new, fields): return {k: v for k, v in old.items() if k not in fields} == { k: v for k, v in new.items() if k not in fields} def row_preserved(table, old, new): if old == new: return True if table == "verifications": if not same_except(old, new, {"consumed_at", "canceled_at"}): return False if new["consumed_at"] is not None and new["canceled_at"] is not None: return False for key in ("consumed_at", "canceled_at"): if old[key] == new[key]: continue if old[key] is not None or not isinstance(new[key], str): return False try: if datetime.fromisoformat(new[key]).tzinfo is None: return False except ValueError: return False return True if table == "caller_buckets": return (same_except(old, new, {"tokens", "updated_at"}) and all(isinstance(new[k], (int, float)) and math.isfinite(new[k]) for k in ("tokens", "updated_at")) and new["tokens"] >= 0 and new["updated_at"] >= old["updated_at"]) if table in {"send_attempts", "verification_attempts"}: mutable = {"state", "error", "retryable"} if table == "verification_attempts": mutable.add("provider_ref") # Only a definitive retryable failure permits a fresh verification handle. retry = old["state"] == "failed" and old["retryable"] == 1 if retry and table == "verification_attempts": mutable.add("request_id") states = {"in_flight", "failed"} if table == "verification_attempts": states.add("accepted") return (same_except(old, new, mutable) and (retry or old["state"] == "in_flight") and new["state"] in states and (new["state"] != "accepted" or bool(new.get("provider_ref")))) # Accepted deliveries and suppression decisions have no authorized mutation # in this configuration-only operation. New rows remain possible at runtime. return False def continuity(before, after): old_schema, old = ledger(before) new_schema, new = ledger(after) require(old_schema == new_schema, "ledger_schema_changed") changes = {} for table in sorted(TABLES): require(old[table].keys() <= new[table].keys(), "ledger_rows_removed") require(all(row_preserved(table, value, new[table][key]) for key, value in old[table].items()), "ledger_history_changed") changes[table] = { "preserved_rows": len(old[table]), "new_rows": len(new[table]) - len(old[table]), "permitted_updates": sum(value != new[table][key] for key, value in old[table].items()), } return {"all_six_tables_checked": True, "schema_unchanged": True, "tables": changes, "ledger_restored": False, "snapshot_bytes_equal": Path(before).read_bytes() == Path(after).read_bytes()} class Journal: """Private append-only intents; incomplete lines refuse automatic recovery.""" def __init__(self, path, *, proposal=None): self.path = Path(path).absolute() require(not any(p.is_symlink() for p in [self.path, *self.path.parents]), "real_journal_path_required") parent = self.path.parent.stat() require(stat.S_ISDIR(parent.st_mode) and parent.st_uid == os.getuid() and not parent.st_mode & 0o077, "private_journal_parent_required") flags = os.O_RDWR | os.O_APPEND | os.O_NOFOLLOW if proposal is not None: flags |= os.O_CREAT | os.O_EXCL self.fd = os.open(self.path, flags, 0o600) try: info = os.fstat(self.fd) require(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid() and not info.st_mode & 0o077 and info.st_nlink == 1, "private_owned_journal_required") fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB) self.entries = [] if proposal is not None: self.record("prepared", schema=SCHEMA, proposal=proposal, proposal_digest=digest(proposal)) directory = os.open(self.path.parent, os.O_DIRECTORY | os.O_RDONLY) try: os.fsync(directory) finally: os.close(directory) else: require(info.st_size <= 16 * 1024**2, "journal_too_large") data = os.read(self.fd, info.st_size) require(data.endswith(b"\n"), "incomplete_journal_requires_inspection") self.entries = [json.loads(line) for line in data.splitlines()] first = self.entries[0] require(first["schema"] == SCHEMA and first["proposal_digest"] == digest(first["proposal"]), "invalid_proposal_binding") require(all(e["sequence"] == i for i, e in enumerate(self.entries)), "invalid_journal_sequence") except BaseException: os.close(self.fd) raise def record(self, phase, **data): record = dict(data, phase=phase, sequence=len(self.entries)) body = json.dumps(record, sort_keys=True).encode() + b"\n" # Write all bytes before returning; short writes never authorize effects. offset = 0 while offset < len(body): written = os.write(self.fd, body[offset:]) require(written > 0, "journal_write_failed") offset += written os.fsync(self.fd) self.entries.append(record) def close(self): os.close(self.fd) def run(adapter, journal, *, recover=False): """One forward attempt, or explicit recovery to the original configuration. Adapter contract: observe/patch use complete UID/spec CAS; fence proves no other PVC writers; snapshot creates a fresh private consistent copy and binds helper identity through journal intents; cleanup removes only owned helpers; verify proves actual process configuration, health and no-send denials. The native adapter, not this protocol, proves credential admission and fencing. """ first = journal.entries[0] proposal = first["proposal"] before = proposal["before"] after = desired(before) require(proposal["after"] == after, "unreviewed_configuration_delta") require(recover or len(journal.entries) == 1, "forward_replay_refused") if recover: require(journal.entries[-1]["phase"] != "recovered", "recovery_already_completed") target = before if recover else after mode = "recovery" if recover else "forward" def observe(): current = adapter.observe() known(current, before, after) return current try: live = observe() if not recover: require(live == before, "preflight_changed") adapter.admission(proposal) journal.record(mode + "_pause_intent") if live["deployment_spec"]["replicas"] != 0: adapter.patch_deployment(live["deployment_spec"], with_replicas(live["deployment_spec"], 0)) adapter.fence(journal) live = observe() require(live["deployment_spec"]["replicas"] == 0, "receiver_not_paused") journal.record(mode + "_capture_intent") baseline, outcomes = adapter.snapshot(journal, quiesced=True) require(Path(baseline).resolve().is_relative_to(journal.path.parent), "snapshot_outside_private_attempt") ledger(baseline) # Refuse a broken or non-private baseline before switching. baseline_sha256 = hashlib.sha256(Path(baseline).read_bytes()).hexdigest() # Recovery may need to preserve an unknown outcome caused by an earlier # interruption. It does not repair or clear that reservation. if not recover: require(not outcomes["needs_outcome_reconciliation"], "unknown_outcome_before_switch") journal.record(mode + "_captured", outcomes=outcomes, snapshot_name=Path(baseline).relative_to(journal.path.parent).as_posix(), snapshot_sha256=baseline_sha256) live = observe() require(live["deployment_spec"]["replicas"] == 0, "receiver_resumed_before_config") journal.record(mode + "_config_intent") adapter.patch_config(live["config_data"], target["config_data"]) live = observe() require(live["deployment_spec"]["replicas"] == 0, "receiver_resumed_before_deployment") journal.record(mode + "_deployment_intent") adapter.patch_deployment(live["deployment_spec"], with_replicas(target["deployment_spec"], 0)) live = observe() require(live["config_data"] == target["config_data"], "sender_config_not_verified") require(live["deployment_spec"] == with_replicas(target["deployment_spec"], 0), "paused_deployment_not_verified") journal.record(mode + "_resume_intent") adapter.patch_deployment(live["deployment_spec"], target["deployment_spec"]) native = adapter.verify(target, journal) require(observe() == target, "runtime_not_verified") journal.record(mode + "_postflight_intent", native=native) returned, _ = adapter.snapshot(journal, quiesced=False) require(Path(returned).resolve().is_relative_to(journal.path.parent), "snapshot_outside_private_attempt") require(hashlib.sha256(Path(baseline).read_bytes()).hexdigest() == baseline_sha256, "baseline_snapshot_changed") result = continuity(baseline, returned) journal.record(mode + "_cleanup_intent", continuity=result) adapter.cleanup(journal) journal.record("recovered" if recover else "completed", continuity=result, ledger_restored=False, helper_cleanup_verified=True, sender=target["config_data"]["EMAIL_CONNECT_SENDER"]) return result except BaseException: # No guessed rollback or resume in finally. Explicit recovery observes # actual state, captures the CURRENT ledger and retains prior attempts. journal.record("requires_inspection", operation=mode) raise