MoShiro/scripts/test-container-updater.py

#!/usr/bin/env python3
"""Signed updater regression checks using fake engine/service CLIs and real signatures."""
import base64
import json
import os
from pathlib import Path
import shlex
import subprocess
import tempfile

SCRIPT = Path(__file__).resolve().parents[1] / "packaging/update/moshiro-update.sh"
with tempfile.TemporaryDirectory(prefix="moshiro-updater-test-") as directory:
    root = Path(directory)
    (root / "bin").mkdir()
    (root / "data/update").mkdir(parents=True)
    (root / "conf").mkdir()
    key = root / "key.pem"
    pub = root / "key.pub"
    subprocess.run(["openssl", "genpkey", "-algorithm", "EC", "-pkeyopt", "ec_paramgen_curve:P-256", "-out", str(key)], check=True, capture_output=True)
    subprocess.run(["openssl", "pkey", "-in", str(key), "-pubout", "-out", str(pub)], check=True, capture_output=True)
    log = root / "calls"
    mode = root / "mode"
    # Only external effects are replaced; verification and state handling run unchanged.
    for name in ("podman", "docker", "systemctl", "sleep"):
        tool = root / "bin" / name
        tool.write_text("""#!/bin/sh
printf '%s %s\n' "${0##*/}" "$*" >> "$TEST_LOG"
case "${0##*/}" in
  sleep) exit 0 ;;
  podman|docker)
    [ "$(cat "$TEST_MODE")" != pull-fail ] || exit 1
    if [ -n "${DOCKER_CONFIG:-}" ]; then test -r "$DOCKER_CONFIG/config.json" || exit 1; fi ;;
  systemctl)
    [ "${1:-}" != --user ] || shift
    case "$1" in
      show) echo 0 ;;
      is-active) [ "$(cat "$TEST_MODE")" != unhealthy ] ;;
    esac ;;
esac
""")
        tool.chmod(0o755)
    env = dict(os.environ, PATH=str(root / "bin") + os.pathsep + os.environ["PATH"], MOSHIRO_UPDATE_CONF_DIR=str(root / "conf"), TEST_LOG=str(log), TEST_MODE=str(mode))
    repo = "registry.example/moshiro/moshiro-agent"
    digest = "sha256:" + "a" * 64
    image_file = root / "agent.image"
    state = root / "state"
    auth = root / "auth file.json"
    auth.write_text('{"auths":{}}')

    def configure(engine="podman", scope="system", policy="manual", explicit=True):
        values = dict(DATA_DIR=str(root / "data"), IMAGE_REPO=repo, UNIT="moshiro-agent.service", PUBKEY=str(pub), ENV_FILE=str(image_file), STATE_DIR=str(state), POLICY=policy, AUTHFILE=str(auth))
        if explicit:
            values.update(ENGINE=engine, SERVICE_SCOPE=scope)
        (root / "conf/agent.conf").write_text("".join(k + "=" + shlex.quote(v) + "\n" for k, v in values.items()))
        image_file.write_text("IMAGE=old-image\n")
        (state / "agent.installed").unlink(missing_ok=True)
        log.write_text("")
        mode.write_text("healthy")

    def offer(seq=1, tamper=False):
        manifest = root / "data/update/offer.manifest"
        manifest.write_text(json.dumps(dict(component="agent", image=repo, digest=digest, tag="v1.0.0", version="1.0.0", seq=seq), separators=(",", ":")))
        sig = subprocess.check_output(["openssl", "dgst", "-sha256", "-sign", str(key), str(manifest)])
        (root / "data/update/offer.sig").write_bytes(base64.b64encode(sig))
        if tamper:
            manifest.write_text(manifest.read_text().replace("1.0.0", "9.0.0"))

    def invoke(command):
        result = subprocess.run(["sh", str(SCRIPT), command, "agent"], env=env, capture_output=True, text=True)
        status = json.loads((root / "data/update-status/status.json").read_text())
        return result, status

    configure(explicit=False)
    offer()
    result, status = invoke("apply")
    assert result.returncode == 0 and status["state"] == "up_to_date", result.stderr
    assert "systemctl restart" in log.read_text() and "podman pull --quiet --authfile" in log.read_text()
    print("PASS legacy rootful Podman config")

    configure(scope="user", policy="auto")
    offer()
    result, status = invoke("check")
    assert result.returncode == 0 and status["state"] == "up_to_date", result.stderr
    assert "systemctl --user restart" in log.read_text()
    assert image_file.read_text() == "IMAGE=" + repo + "@" + digest + "\n"
    print("PASS rootless Podman auto policy and user service restart")
    log.write_text("")
    _, status = invoke("check")
    assert status["state"] == "up_to_date" and not log.read_text()
    print("PASS replayed manifest does not pull or restart")

    configure(engine="docker")
    offer()
    result, status = invoke("apply")
    assert result.returncode == 0 and status["state"] == "up_to_date", result.stderr
    assert "docker pull --quiet" in log.read_text() and "--authfile" not in log.read_text()
    assert "systemctl --user" not in log.read_text()
    print("PASS Docker auth config and system service restart")

    configure(scope="user")
    offer(tamper=True)
    result, status = invoke("apply")
    assert result.returncode != 0 and status["state"] == "invalid" and not log.read_text()
    print("PASS invalid signature never invokes engine/service")

    configure(scope="user")
    offer()
    mode.write_text("unhealthy")
    result, status = invoke("apply")
    assert result.returncode != 0 and status["state"] == "rolled_back"
    assert image_file.read_text() == "IMAGE=old-image\n"
    assert log.read_text().count("systemctl --user restart") == 2
    assert not (state / "agent.installed").exists()
    print("PASS unhealthy image rolls back without advancing sequence")

    configure(engine="docker")
    offer()
    mode.write_text("pull-fail")
    result, status = invoke("apply")
    assert result.returncode != 0 and status["state"] == "failed"
    assert image_file.read_text() == "IMAGE=old-image\n" and "systemctl" not in log.read_text()
    print("PASS failed pull leaves active image unchanged")