email-connect/scripts/sender_release.py

"""Source-bound node coordinator for the attended sender release."""

from __future__ import annotations

import hashlib
import json
import os
import stat
import urllib.request
from pathlib import Path

from sender_native import FOUNDATION, FOUNDATION_SPEC
from sender_native import Native as BaseNative
from sender_transition import Journal, desired, digest, known, require, run

SCHEMA = "email-connect.sender-release-proposal.v1"
TASK = "da56d175-3126-5eaa-90c5-dd1e9da050d5"
WORKPLAN = "c042789b-4d7f-5c55-8d49-39dd16500f41"
STATE = Path("/var/lib/email-connect-sender-release")
FILES = {"release-sender.py", "sender_release.py", "sender_native.py", "sender_transition.py",
         "capture-ledger.py", "ledger_archive.py", "inspect-sender-ledger.py", "rehearse-ledger.py"}


class Native(BaseNative):
    """Bind projection UIDs observed at sealing, as well as the adapter's specs."""

    def projection(self, proposal):
        pins = super().projection(proposal)
        if "foundation_pins" in proposal:
            require(pins == proposal["foundation_pins"], "reviewed_projection_identity_changed")
        return pins


def source_hashes(sources):
    require(set(sources) == FILES and all(isinstance(v, str) for v in sources.values()),
            "source_closure_changed")
    require(sum(len(v.encode()) for v in sources.values()) <= 2 * 1024**2, "source_capsule_too_large")
    return {name: hashlib.sha256(value.encode()).hexdigest() for name, value in sources.items()}


def owner_return(base):
    # A consumer may observe this handoff, never mark the owner task done itself.
    with urllib.request.urlopen(base + "/tasks/" + TASK, timeout=15) as response:
        value = json.load(response)
    require(value.get("id") == TASK and value.get("workplan_id") == WORKPLAN
            and value.get("status") == "done", "verified_credential_owner_return_required")
    return {key: value[key] for key in ("id", "workplan_id", "status")}


def foundation_receipt(value):
    require(value.get("schema") == "platform.email-connect-smtp-foundation.v1"
            and value.get("proposal_digest") == FOUNDATION
            and value.get("status") == "foundation_admitted_projection_ready"
            and value.get("application_sender_changed") is False
            and isinstance(value.get("service_account_uid"), str)
            and bool(value["service_account_uid"]), "successful_foundation_receipt_required")
    for key in ("native_binding_verified", "native_consumer_auth_verified", "exact_read_verified",
                "mutation_sibling_metadata_denied", "bounded_ttl", "other_service_account_denied",
                "wrong_audience_denied", "workload_token_revocation_verified", "projection_ready"):
        require(value.get(key) is True, "foundation_proof_incomplete")


def validate(packet):
    require(__debug__, "optimized_python_refused")
    plan = packet["proposal"]
    require(plan["schema"] == SCHEMA and packet["proposal_digest"] == "sha256:" + digest(plan),
            "proposal_digest_mismatch")
    require(plan["sources"] == source_hashes(packet["sources"]), "source_digest_mismatch")
    require(plan["after"] == desired(plan["before"]), "unreviewed_configuration_delta")
    require(digest(plan["foundation"]) == FOUNDATION_SPEC, "foundation_contract_changed")
    pins = plan["foundation_pins"]
    require(set(pins) == {"ServiceAccount", "SecretStore", "ExternalSecret", "Secret"}
            and all(isinstance(v, str) and bool(v) for v in pins.values()), "projection_identity_binding_required")
    foundation_receipt(plan["foundation_receipt"])
    require(plan["owner_return"] == {"id": TASK, "workplan_id": WORKPLAN, "status": "done"},
            "verified_credential_owner_return_required")
    return plan


def prepare(sources, foundation, receipt):
    hashes = source_hashes(sources)
    owner = owner_return("http://10.43.68.154:8000")
    require(digest(foundation) == FOUNDATION_SPEC, "foundation_contract_changed")
    foundation_receipt(receipt)
    adapter = Native()
    pins = adapter.projection({"foundation": foundation})
    require(pins["ServiceAccount"] == receipt["service_account_uid"], "admitted_identity_changed")
    native = adapter.prepare()
    require(adapter.projection({"foundation": foundation}) == pins, "projection_changed_during_preparation")
    plan = {"schema": SCHEMA, "before": native["before"], "after": native["after"],
            "sources": hashes, "foundation": foundation, "foundation_receipt": receipt,
            "foundation_pins": pins, "owner_return": owner}
    return {"proposal": plan, "proposal_digest": "sha256:" + digest(plan), "sources": sources}


def private(path, *, directory=False):
    require(not any(p.is_symlink() for p in (path, *path.parents)), "real_private_path_required")
    info = path.stat()
    require(info.st_uid == os.geteuid() and not info.st_mode & 0o077
            and (stat.S_ISDIR(info.st_mode) if directory else
                 stat.S_ISREG(info.st_mode) and info.st_nlink == 1), "private_owned_path_required")


def write(path, value):
    private(path.parent, directory=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
    with os.fdopen(fd, "w") as stream:
        json.dump(value, stream, sort_keys=True)
        stream.write("\n")
        stream.flush()
        os.fsync(stream.fileno())
    sync_directory(path.parent)


def sync_directory(path):
    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def process_identity(pid):
    try:
        # Field 22 follows the parenthesized command; that name may contain spaces.
        fields = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()
        if fields[0] == "Z":
            return None
        return {"pid": pid, "start_ticks": fields[19]}
    except FileNotFoundError:
        return None


def attempt(packet):
    validate(packet)
    return STATE / packet["proposal_digest"].removeprefix("sha256:")


def inspect(packet):
    directory = attempt(packet)
    if not directory.exists():
        return {"status": "attempt_absent", "proposal_digest": packet["proposal_digest"]}
    private(directory, directory=True)
    private(directory / "capsule.json")
    require(json.loads((directory / "capsule.json").read_text()) == packet, "retained_capsule_changed")
    running = False
    for process in [directory / "process.json", *directory.glob("recovery-process-*.json")]:
        if not process.exists():
            continue
        private(process)
        recorded = json.loads(process.read_text())
        running = process_identity(recorded["pid"]) == recorded
        if running:
            break
    # Do not read a partial journal as a success or infer death from a stale file.
    if running:
        return {"status": "attempt_process_running", "proposal_digest": packet["proposal_digest"],
                "pid": recorded["pid"], "retry_automatically": False}
    journal = Journal(directory / "journal.jsonl")
    try:
        require(journal.entries[0]["proposal"] == packet["proposal"], "journal_proposal_changed")
        observed = Native().observe()
        known(observed, packet["proposal"]["before"], packet["proposal"]["after"])
        return {"status": "attempt_inspected", "proposal_digest": packet["proposal_digest"],
                "last_phase": journal.entries[-1]["phase"], "process_running": False,
                "sender": observed["config_data"]["EMAIL_CONNECT_SENDER"],
                "replicas": observed["deployment_spec"]["replicas"], "retry_automatically": False}
    finally:
        journal.close()


def execute(packet, *, recover=False):
    plan = validate(packet)
    directory = attempt(packet)
    if not recover:
        require(owner_return("http://10.43.68.154:8000") == plan["owner_return"], "owner_return_changed")
        # Verify live preimages and projection before creating an attempt.
        adapter = Native()
        require(adapter.observe() == plan["before"], "preflight_changed")
        require(adapter.projection(plan)["ServiceAccount"] == plan["foundation_receipt"]["service_account_uid"],
                "admitted_identity_changed")
        if not STATE.exists():
            STATE.mkdir(mode=0o700)
            sync_directory(STATE.parent)
        private(STATE, directory=True)
        directory.mkdir(mode=0o700)  # Exclusive: no forward replay.
        sync_directory(STATE)
        write(directory / "capsule.json", packet)
        journal = Journal(directory / "journal.jsonl", proposal=plan)
    else:
        private(directory, directory=True)
        private(directory / "capsule.json")
        require(json.loads((directory / "capsule.json").read_text()) == packet, "retained_capsule_changed")
        journal = Journal(directory / "journal.jsonl")
    adapter = None
    try:
        require(journal.entries[0]["proposal"] == plan, "journal_proposal_changed")
        # Journal flock excludes a live attempt even if the launcher lost its reply.
        # Keep the original process record; recovery gets a separately numbered one.
        write(directory / ("recovery-process-" + str(len(journal.entries)) + ".json" if recover
                           else "process.json"), process_identity(os.getpid()))
        adapter = Native(execute=True, journal=journal)
        adapter.acquire()
        result = run(adapter, journal, recover=recover)
        return {"schema": "email-connect.sender-release-result.v1",
                "proposal_digest": packet["proposal_digest"],
                "status": "configuration_recovered_verified" if recover else "sender_deployed_verified",
                "sender": journal.entries[-1]["sender"], "ledger_restored": False,
                "helper_cleanup_verified": True, "continuity": result,
                "journal": str(journal.path), "mail_sent": False}
    finally:
        if adapter is not None:
            adapter.close()
        journal.close()