email-connect/scripts/sender_transition.py

"""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