coulomb-social/scripts/backup_install_runtime.py

"""Remote, attended source installation; never captures, sends or starts a unit."""

from __future__ import annotations

import base64
import fcntl
import hashlib
import json
import os
import stat
import subprocess
import types
from contextlib import contextmanager
from datetime import UTC, datetime
from pathlib import Path

ROOT = Path("/")
STATE = Path("/var/lib/coulomb-social-backup")
ATTEMPTS = Path("/var/lib/coulomb-social-backup-installations")
ROUTE_ROOT = Path("/var/lib/rapp-telemetry/backup-route-releases")
CLUSTER_UID = "a553c742-0115-43d4-99a4-a5ca56fe0786"
UNITS = (
    "coulomb-social-paired-backup.service",
    "coulomb-social-paired-backup.timer",
    "coulomb-social-backup-freshness.service",
    "coulomb-social-backup-freshness.timer",
    "coulomb-social-backup-alert@coulomb-social-paired-backup.service.service",
)
PROFILE = dict(
    maximum_new_file_bytes=1024**3,
    state_budget_bytes=8 * 1024**3,
    minimum_free_bytes=8 * 1024**3,
    minimum_free_inodes=1024,
    control_margin_bytes=16 * 1024**2,
)


class Refused(ValueError):
    pass


def require(condition, reason):
    if not condition:
        raise Refused(reason)


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def canonical(value):
    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()


def module(source, name):
    result = types.ModuleType(name)
    exec(compile(source, "<reviewed-" + name + ">", "exec"), result.__dict__)
    return result


def command(args, timeout=30):
    result = subprocess.run(
        args,
        capture_output=True,
        timeout=timeout,
        env=dict(os.environ, KUBECONFIG="/etc/rancher/k3s/k3s.yaml"),
    )
    require(
        result.returncode == 0 and len(result.stdout) < 16 * 1024 * 1024, "bounded_command_failed"
    )
    return result.stdout


def cluster_identity():
    namespace = json.loads(
        command(["k3s", "kubectl", "get", "namespace", "kube-system", "-o", "json"])
    )
    require(namespace["metadata"]["uid"] == CLUSTER_UID, "primary_cluster_mismatch")


def inactive_units():
    rows = {}
    # This host's list-jobs does not implement JSON output; no-legend emits
    # empty output for no jobs, otherwise JOB UNIT TYPE STATE rows.
    jobs = command(["systemctl", "list-jobs", "--no-legend", "--no-pager", "--plain"])
    jobs = [line.split() for line in jobs.decode().splitlines() if line.strip()]
    require(all(len(row) == 4 and row[0].isdecimal() for row in jobs), "job_listing_unrecognized")
    require(
        not any(row[1].startswith("coulomb-social-") for row in jobs),
        "queued_coulomb_unit_job",
    )
    # Also reject an already running alternative instance of the alert template.
    listing = json.loads(
        command(
            ["systemctl", "list-units", "--all", "--output=json", "--no-pager", "coulomb-social-*"]
        )
    )
    require(all(x.get("active") == "inactive" for x in listing), "coulomb_unit_active_or_failed")
    for unit in UNITS:
        raw = command(
            [
                "systemctl",
                "show",
                unit,
                "--property=LoadState,ActiveState,SubState,UnitFileState,FragmentPath",
            ]
        )
        row = dict(line.split("=", 1) for line in raw.decode().splitlines() if "=" in line)
        require(row["ActiveState"] == "inactive" and row["SubState"] == "dead", "unit_not_inactive")
        require(
            row["LoadState"] in ("loaded", "not-found")
            and row.get("UnitFileState", "") in ("", "disabled", "static"),
            "unit_enabled_or_unexpected",
        )
        if row["LoadState"] == "loaded":
            fragment = unit if "@" not in unit else "coulomb-social-backup-alert@.service"
            require(
                row["FragmentPath"] == "/etc/systemd/system/" + fragment, "foreign_unit_fragment"
            )
        rows[unit] = row
    return rows


@contextmanager
def locks():
    handles = []
    try:
        info = STATE.lstat()
        require(
            stat.S_ISDIR(info.st_mode) and info.st_uid == os.geteuid() and not info.st_mode & 0o077,
            "private_backup_state_required",
        )
        for name in ("backup.lock", "snapshot.lock"):
            # Existing lock only; no truncation, replacement or lock-file creation.
            fd = os.open(STATE / name, os.O_RDONLY | os.O_NOFOLLOW)
            handles.append(fd)
            info = os.fstat(fd)
            require(
                stat.S_ISREG(info.st_mode)
                and info.st_uid == os.geteuid()
                and info.st_nlink == 1
                and not info.st_mode & 0o077,
                "unsafe_lock_file",
            )
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        yield
    finally:
        for fd in reversed(handles):
            os.close(fd)


def verified_payload(packet):
    require(__debug__, "optimized_python_refused")
    plan = packet["plan"]
    require(
        packet["proposal_digest"] == "sha256:" + sha(canonical(plan)), "proposal_digest_differs"
    )
    require(
        plan["capacity_profile"] == PROFILE and plan["cluster_uid"] == CLUSTER_UID,
        "operating_scope_differs",
    )
    sources = {}
    for key in ("installer", "route_verifier"):
        raw = base64.b64decode(packet[key], validate=True)
        require(sha(raw) == plan[key + "_sha256"], "source_digest_differs")
        sources[key] = raw
    archive = base64.b64decode(packet["archive"], validate=True)
    installer = module(sources["installer"], "backup_install")
    manifest, contents = installer.read_bundle(
        archive, plan["archive_sha256"], plan["manifest_sha256"]
    )
    require(
        plan["route_packet"]["execution_digest"] == plan["route_execution_digest"],
        "route_digest_differs",
    )
    return installer, manifest, contents, sources


def route_verified(plan, source):
    path = ROUTE_ROOT / plan["route_execution_digest"].split(":")[1] / "receipt.json"
    require(path.is_file() and not path.is_symlink(), "route_receipt_missing")
    info = path.stat()
    require(
        info.st_uid == 0 and not info.st_mode & 0o077 and info.st_size < 65536,
        "route_receipt_not_private",
    )
    receipt = json.loads(path.read_bytes())
    require(
        receipt["status"] == "backup_email_route_deployed_verified"
        and receipt["execution_digest"] == plan["route_execution_digest"],
        "route_not_verified",
    )
    verifier = module(source, "backup_route_runtime")
    revision = verifier.verify_return(plan["route_packet"], after=True)
    require(revision == receipt["helm_revision"], "route_revision_differs")
    return {"helm_revision": revision, "runtime_and_complete_resources_verified": True}


def capacity(contents):
    guard = module(contents["consumer/backup_capacity.py"], "backup_capacity")
    require(
        (
            guard.FILE_LIMIT,
            guard.STATE_BUDGET,
            guard.MINIMUM_FREE,
            guard.MINIMUM_INODES,
            guard.CONTROL_MARGIN,
        )
        == tuple(PROFILE.values()),
        "capacity_profile_differs",
    )
    try:
        report = guard.reserve_plan(STATE, require_file_limit=False)
    except guard.Refused as error:
        raise Refused(str(error)) from None
    require(not report["reasons"], "capacity_headroom_refused")
    return {
        k: report[k]
        for k in (
            "state_bytes",
            "state_reserve_bytes",
            "disk_reserve_bytes",
            "free_bytes",
            "free_inodes",
            "filesystem_blocks_reserved",
        )
    }


def runtime_dependencies():
    code = (
        "import json,boto3,botocore; print(json.dumps(dict(boto3=boto3.__version__,"
        "botocore=botocore.__version__)))"
    )
    actual = json.loads(command(["/opt/railiance-platform/backup-venv/bin/python", "-c", code]))
    require(actual == {"boto3": "1.40.18", "botocore": "1.40.76"}, "sdk_dependency_drift")
    # Check credential references' custody, without reading their contents.
    for name in ("primary.env", "recipient.env"):
        path = Path("/etc/coulomb-social-backup") / name
        info = path.lstat()
        require(
            stat.S_ISREG(info.st_mode)
            and info.st_uid == 0
            and info.st_nlink == 1
            and not info.st_mode & 0o077,
            "credential_reference_custody_differs",
        )
    return actual


def preflight(plan, installer, contents, source):
    cluster_identity()
    units = inactive_units()
    before = installer.observe(ROOT)
    require(before == plan["expected_before"], "native_preimage_changed")
    require(not (ATTEMPTS / plan["attempt_id"]).exists(), "existing_attempt_requires_inspection")
    profile = capacity(contents)
    sdk = runtime_dependencies()
    route = route_verified(plan, source)
    return dict(units=units, capacity=profile, sdk=sdk, route=route)


def run(packet, action):
    require(action in ("preflight", "install", "rollback"), "unknown_action")
    require(os.geteuid() == 0, "root_owner_execution_required")
    installer, manifest, contents, sources = verified_payload(packet)
    plan = packet["plan"]
    directory = ATTEMPTS / plan["attempt_id"]
    require(
        len(plan["attempt_id"]) == 16 and all(c in "0123456789abcdef" for c in plan["attempt_id"]),
        "invalid_attempt_id",
    )
    installer.trusted_parents(ROOT, directory / "receipt.json")
    with locks():
        if action == "rollback":
            cluster_identity()
            inactive_units()
            require(directory.is_dir() and not directory.is_symlink(), "attempt_missing")
            prior = json.loads((directory / "receipt.json").read_text())
            require(
                prior["proposal_digest"] == packet["proposal_digest"], "attempt_proposal_differs"
            )
            record = json.loads((directory / "files/receipt.json").read_text())
            prior.update(status="rollback_in_progress", updated_at=datetime.now(UTC).isoformat())
            installer.journal(directory, prior)
            installer.restore(
                ROOT, directory / "files", record, lambda: command(["systemctl", "daemon-reload"])
            )
            inactive_units()
            prior.update(status="prior_files_restored_units_inactive_verified")
            installer.journal(directory, prior)
            return prior
        result = preflight(plan, installer, contents, sources["route_verifier"])
        receipt = dict(
            schema="coulomb.backup-source-supervised-installation.v1",
            proposal_digest=packet["proposal_digest"],
            status="preflight_verified",
            preflight=result,
            timers_activated=False,
            capture_executed=False,
            email_sent=False,
            retained_data_deleted=False,
        )
        if action == "preflight":
            return receipt
        directory.mkdir(mode=0o700, parents=True, exist_ok=False)
        receipt.update(
            status="installing",
            started_at=datetime.now(UTC).isoformat(),
            capacity_profile=PROFILE,
            founder_confirmed_profile_and_installation=True,
        )
        installer.journal(directory, receipt)
        try:
            files = installer.install(
                manifest,
                contents,
                plan["expected_before"],
                directory / "files",
                root=ROOT,
                reload_manager=lambda: command(["systemctl", "daemon-reload"]),
                verify_inactive=inactive_units,
            )
            # Recheck the external dependency after installation, without a send.
            receipt["route_postflight"] = route_verified(plan, sources["route_verifier"])
            receipt.update(
                status="sources_installed_units_inactive_verified",
                verified_targets=files["verified_targets"],
                units=inactive_units(),
            )
        except Exception as error:
            receipt.update(
                status="installation_requires_inspection", failure_type=type(error).__name__
            )
            raise
        finally:
            receipt["updated_at"] = datetime.now(UTC).isoformat()
            installer.journal(directory, receipt)
        return receipt