bootstrap.py

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tarfile
import urllib.request
from pathlib import Path


def required(name: str) -> str:
    value = os.getenv(name, "").strip()
    if not value:
        raise RuntimeError(f"missing required environment variable: {name}")
    return value


def safe_extract(archive: Path, destination: Path) -> None:
    destination_resolved = destination.resolve()
    with tarfile.open(archive, "r:gz") as bundle:
        for member in bundle.getmembers():
            target = (destination / member.name).resolve()
            if destination_resolved not in target.parents and target != destination_resolved:
                raise RuntimeError(f"unsafe bundle member: {member.name}")
        bundle.extractall(destination)


def main() -> None:
    bundle_url = required("PVCE_BUNDLE_URL")
    bundle_token = required("PVCE_BUNDLE_BEARER_TOKEN")
    bundle_version = required("PVCE_BUNDLE_VERSION")

    state_root = Path(os.getenv("PVCE_RUNTIME_ROOT", "/var/data/pvce-runtime"))
    app_root = state_root / "app"
    archive = state_root / "bundle.tar.gz"
    marker = state_root / "bundle-state.json"
    state_root.mkdir(parents=True, exist_ok=True)

    remote_state = {"version": bundle_version}

    local_state = {}
    if marker.exists():
        try:
            local_state = json.loads(marker.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            local_state = {}

    if local_state != remote_state or not (app_root / "pvce_system" / "server.py").exists():
        request = urllib.request.Request(
            bundle_url,
            headers={
                "Accept": "application/octet-stream",
                "Authorization": f"Bearer {bundle_token}",
                "User-Agent": "brn-private-runtime-bootstrap",
                "X-GitHub-Api-Version": "2022-11-28",
            },
        )
        with urllib.request.urlopen(request, timeout=180) as response, archive.open("wb") as output:
            shutil.copyfileobj(response, output)
        fresh = state_root / "app.new"
        shutil.rmtree(fresh, ignore_errors=True)
        fresh.mkdir(parents=True, exist_ok=True)
        safe_extract(archive, fresh)
        if not (fresh / "pvce_system" / "server.py").exists():
            raise RuntimeError("private bundle is missing pvce_system/server.py")
        previous = state_root / "app.previous"
        shutil.rmtree(previous, ignore_errors=True)
        if app_root.exists():
            app_root.replace(previous)
        fresh.replace(app_root)
        marker.write_text(json.dumps(remote_state, sort_keys=True), encoding="utf-8")
        archive.unlink(missing_ok=True)
        shutil.rmtree(previous, ignore_errors=True)

    env = os.environ.copy()
    env["PYTHONPATH"] = str(app_root)
    env.setdefault("PVCE_DB_PATH", "/var/data/pvce.sqlite3")
    os.chdir(app_root)
    completed = subprocess.run([sys.executable, "-m", "pvce_system.server"], env=env)
    raise SystemExit(completed.returncode)


if __name__ == "__main__":
    main()