coulomb-social/scripts/backup_install.py

"""Verified source-bundle installation primitives; no CLI, capture or timer start.

The supervised caller must hold backup/snapshot locks and verify disabled units
and the admitted route/profile. This module confines file writes and preserves
preimages; it does not grant that caller authority or activate a schedule.
"""

from __future__ import annotations

import hashlib
import io
import json
import os
import stat
import tarfile
import tempfile
from pathlib import Path

CONSUMER = (
    "paired-backup.py", "snapshot-content-and-db.py", "mirror-paired-backup.py",
    "check-paired-backup-health.py", "backup_capacity.py", "mirror-platform-runtime.json",
    "backup_alerts.py", "notify-backup-alert.py", "observe-backup-alerts.py",
)
PLATFORM = (
    "application_recovery_identities.py", "application_recovery_mirror.py",
    "application_recovery_resources.py", "primary_application_archive.py",
)
UNITS = (
    "coulomb-social-backup-freshness.service", "coulomb-social-paired-backup.service",
    "coulomb-social-backup-freshness.timer", "coulomb-social-paired-backup.timer",
    "coulomb-social-backup-alert@.service",
)
TARGETS = {**{"consumer/" + n: "/opt/coulomb-social-backup/" + n for n in CONSUMER},
           **{"platform/" + n: "/opt/railiance-platform/scripts/" + n for n in PLATFORM},
           **{"units/" + n: "/etc/systemd/system/" + n for n in UNITS},
           "configuration/alerting.env": "/etc/coulomb-social-backup/alerting.env",
           "evidence/backup-primary.lock": None,
           "evidence/alertmanager-config.pending.json": None}


class Refused(ValueError):
    pass


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


def read_bundle(data, archive_sha256, manifest_sha256):
    if len(data) > 2 * 1024 * 1024 or sha(data) != archive_sha256:
        raise Refused("archive_digest_or_size")
    with tarfile.open(fileobj=io.BytesIO(data)) as archive:
        members = archive.getmembers()
        names = [m.name for m in members]
        if (len(set(names)) != len(names) or set(names) != set(TARGETS) | {"manifest.json"}
                or any(not m.isfile() or not 0 < m.size <= 512 * 1024 for m in members)):
            raise Refused("unexpected_archive_member")
        contents = {m.name: archive.extractfile(m).read() for m in members}
    if sha(contents["manifest.json"]) != manifest_sha256:
        raise Refused("manifest_digest")
    manifest = json.loads(contents.pop("manifest.json"))
    if manifest.get("schema") != "coulomb.backup-operating-source-bundle.v1":
        raise Refused("manifest_schema")
    files = manifest["files"]
    if len(files) != len(TARGETS) or {f["member"] for f in files} != set(TARGETS):
        raise Refused("manifest_members")
    for entry in files:
        name = entry["member"]
        if entry["target"] != TARGETS[name]:
            raise Refused("target_outside_reviewed_paths")
        if entry["target"] and entry["mode"] != ("0644" if name.startswith("units/") else "0600"):
            raise Refused("unexpected_target_mode")
        if entry["sha256"] != sha(contents[name]) or entry["bytes"] != len(contents[name]):
            raise Refused("member_digest_or_size")
    return manifest, contents


def path_for(root, target):
    if target not in TARGETS.values() or target is None:
        raise Refused("target_outside_reviewed_paths")
    return root / target.lstrip("/")


def trusted_parents(root, path):
    current = path.parent
    while True:
        if current.exists() or current.is_symlink():
            info = current.lstat()
            if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid()
                    or info.st_mode & 0o022):
                raise Refused("unsafe_target_parent")
        if current == root:
            break
        if current == current.parent:
            raise Refused("target_not_under_root")
        current = current.parent


def fingerprint(path):
    if not path.exists() and not path.is_symlink():
        return {"exists": False}
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
    with os.fdopen(fd, "rb") as stream:
        info = os.fstat(stream.fileno())
        if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_nlink != 1:
            raise Refused("unsafe_target_file")
        data = stream.read(512 * 1024 + 1)
        after = os.fstat(stream.fileno())
    if len(data) > 512 * 1024 or (info.st_size, info.st_mtime_ns) != (
            after.st_size, after.st_mtime_ns):
        raise Refused("target_size_or_read_race")
    return {"exists": True, "sha256": sha(data), "mode": stat.S_IMODE(info.st_mode),
            "uid": info.st_uid, "gid": info.st_gid, "inode": info.st_ino,
            "device": info.st_dev, "mtime_ns": info.st_mtime_ns}


def observe(root=Path("/")):
    result = {}
    for target in TARGETS.values():
        if target is not None:
            path = path_for(root, target)
            trusted_parents(root, path)
            result[target] = fingerprint(path)
    return result


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


def journal(directory, record):
    fd, name = tempfile.mkstemp(prefix=".journal-", dir=directory)
    with os.fdopen(fd, "w") as stream:
        json.dump(record, stream, indent=2)
        stream.flush()
        os.fsync(stream.fileno())
    os.replace(name, directory / "receipt.json")
    sync_directory(directory)


def stage(path, data, mode, gid=None):
    fd, name = tempfile.mkstemp(prefix=".coulomb-install-", dir=path.parent)
    try:
        with os.fdopen(fd, "wb") as stream:
            stream.write(data)
            if gid is not None:
                os.fchown(stream.fileno(), os.geteuid(), gid)
            os.fchmod(stream.fileno(), mode)
            stream.flush()
            os.fsync(stream.fileno())
    except BaseException:
        Path(name).unlink(missing_ok=True)
        raise
    return Path(name)


def restore(root, directory, record, reload_manager):
    """Explicit rollback: refuse interference before undoing any installed file."""
    if len({c["target"] for c in record["changes"]}) != len(record["changes"]):
        raise Refused("duplicate_rollback_target")
    for item in record["changes"]:
        path = path_for(root, item["target"])
        trusted_parents(root, path)
        preimage = item["preimage"]
        if not preimage.endswith(".before") or not preimage.removesuffix(".before").isdecimal():
            raise Refused("invalid_preimage_path")
        for key in ("temporary", "rollback_temporary"):
            if key in item:
                temporary = Path(item[key])
                if (temporary.parent != path.parent
                        or not temporary.name.startswith(".coulomb-install-")):
                    raise Refused("staging_outside_reviewed_parent")
        current = fingerprint(path_for(root, item["target"]))
        if current not in (item["before"], item["staged"], item.get("restored")):
            raise Refused("rollback_target_changed_requires_review")
        if item["before"]["exists"]:
            if sha((directory / item["preimage"]).read_bytes()) != item["before"]["sha256"]:
                raise Refused("rollback_preimage_differs")
    # Persist expected restoration identities before any replacement/unlink, so
    # an interrupted rollback can distinguish its own writes from interference.
    for item in record["changes"]:
        path = path_for(root, item["target"])
        if fingerprint(path) != item["staged"] or item.get("restored") is not None:
            continue
        before = item["before"]
        if before["exists"]:
            temporary = stage(path, (directory / item["preimage"]).read_bytes(),
                              before["mode"], before["gid"])
            item.update(rollback_temporary=str(temporary), restored=fingerprint(temporary))
        else:
            item["restored"] = {"exists": False}
    record["status"] = "rollback_prepared"
    journal(directory, record)
    for item in reversed(record["changes"]):
        path = path_for(root, item["target"])
        if fingerprint(path) == item["staged"]:
            before = item["before"]
            if before["exists"]:
                temporary = Path(item["rollback_temporary"])
                if fingerprint(temporary) != item["restored"]:
                    raise Refused("rollback_staging_changed_requires_review")
                os.replace(temporary, path)
            else:
                path.unlink()
            sync_directory(path.parent)
        temporary = Path(item["temporary"])
        if temporary.exists():
            if fingerprint(temporary) != item["staged"]:
                raise Refused("rollback_staging_changed_requires_review")
            temporary.unlink()
    reload_manager()
    for item in record["changes"]:
        now, before = fingerprint(path_for(root, item["target"])), item["before"]
        keys = ("exists", "sha256", "mode", "uid", "gid") if before["exists"] else ("exists",)
        if any(now.get(k) != before.get(k) for k in keys):
            raise Refused("rollback_readback_failed")
    record["status"] = "prior_files_restored_verified"
    journal(directory, record)
    return record


def install(manifest, contents, expected_before, directory, *, root, reload_manager,
            verify_inactive):
    """Caller supplies admitted bytes, exact native preimage and locked execution context.

    Shared existing platform helpers must already match their pinned content.
    Unchanged files keep their inode/mode; changed existing Python files retain
    their owner's execute bit. All installs finish with inactive-unit verification.
    """
    if observe(root) != expected_before:
        raise Refused("installed_source_drift")
    verify_inactive()
    for entry in manifest["files"]:
        if entry["member"].startswith("platform/"):
            before = expected_before[entry["target"]]
            if before["exists"] and before["sha256"] != entry["sha256"]:
                raise Refused("shared_platform_source_drift")
    directory.mkdir(mode=0o700, parents=True, exist_ok=False)
    record = {"schema": "coulomb.backup-source-installation.v1", "status": "staging",
              "changes": [], "timers_activated": False, "capture_executed": False}
    journal(directory, record)
    try:
        for entry in manifest["files"]:
            target = entry["target"]
            if target is None:
                continue
            before = expected_before[target]
            if before["exists"] and before["sha256"] == entry["sha256"]:
                continue
            path = path_for(root, target)
            trusted_parents(root, path)
            path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
            preimage = str(len(record["changes"])) + ".before"
            if before["exists"]:
                data = path.read_bytes()
                if sha(data) != before["sha256"]:
                    raise Refused("preimage_changed")
                fd = os.open(directory / preimage, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
                with os.fdopen(fd, "wb") as stream:
                    stream.write(data)
                    stream.flush()
                    os.fsync(stream.fileno())
            mode = int(entry["mode"], 8)
            if target.endswith(".py") and before["exists"]:
                mode |= before["mode"] & 0o100
            temporary = stage(path, contents[entry["member"]], mode,
                              before["gid"] if before["exists"] else None)
            record["changes"].append({
                "target": target, "before": before, "preimage": preimage,
                "temporary": str(temporary), "staged": fingerprint(temporary)})
            journal(directory, record)
        if observe(root) != expected_before:
            raise Refused("source_changed_during_staging")
        record["status"] = "installing"
        journal(directory, record)
        for item in record["changes"]:
            if fingerprint(path_for(root, item["target"])) != item["before"]:
                raise Refused("source_changed_before_replace")
            os.replace(item["temporary"], path_for(root, item["target"]))
            sync_directory(path_for(root, item["target"]).parent)
        reload_manager()
        verify_inactive()
        for item in record["changes"]:
            if fingerprint(path_for(root, item["target"])) != item["staged"]:
                raise Refused("installed_readback_differs")
        expected_after = dict(expected_before)
        expected_after.update({item["target"]: item["staged"] for item in record["changes"]})
        if observe(root) != expected_after:
            raise Refused("preserved_source_changed")
        record["status"] = "sources_installed_units_inactive_verified"
        record["verified_targets"] = len(expected_after)
        journal(directory, record)
    except Exception as error:
        record.update(status="installation_requires_review", failure_type=type(error).__name__)
        journal(directory, record)
        raise
    return record