robot-u-site/scripts/bootstrap_lxc_deploy_key.py

#!/usr/bin/env python3
from __future__ import annotations

import base64
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen


DEFAULT_FORGEJO_BASE_URL = "https://aksal.cloud"
DEFAULT_REPO = "Robot-U/robot-u-site"
DEFAULT_DEPLOY_HOST = "192.168.1.220"
DEFAULT_DEPLOY_PORT = "22"
DEFAULT_DEPLOY_USER = "root"
DEFAULT_KEY_COMMENT = "robot-u-site-actions-deploy"
DEFAULT_SECRET_NAME = "DEPLOY_SSH_KEY"


def main() -> int:
    root_dir = Path(__file__).resolve().parents[1]
    _load_env_file(root_dir / ".env")
    _load_env_file(root_dir / ".env.local")

    token = os.getenv("FORGEJO_API_TOKEN") or os.getenv("FORGEJO_TOKEN")
    if not token:
        print(
            "Set FORGEJO_API_TOKEN to a Forgejo token that can manage repo secrets.",
            file=sys.stderr,
        )
        return 1

    base_url = (os.getenv("FORGEJO_BASE_URL") or DEFAULT_FORGEJO_BASE_URL).rstrip("/")
    repo = os.getenv("FORGEJO_REPO") or DEFAULT_REPO
    deploy_host = os.getenv("LXC_DEPLOY_HOST") or DEFAULT_DEPLOY_HOST
    deploy_port = os.getenv("LXC_DEPLOY_PORT") or DEFAULT_DEPLOY_PORT
    deploy_user = os.getenv("LXC_DEPLOY_USER") or DEFAULT_DEPLOY_USER
    key_comment = os.getenv("LXC_DEPLOY_KEY_COMMENT") or DEFAULT_KEY_COMMENT
    secret_name = os.getenv("LXC_DEPLOY_SECRET_NAME") or DEFAULT_SECRET_NAME
    owner, repo_name = _repo_parts(repo)

    ssh_keygen = shutil.which("ssh-keygen")
    ssh = shutil.which("ssh")
    if not ssh_keygen:
        print("ssh-keygen is required.", file=sys.stderr)
        return 1
    if not ssh:
        print("ssh is required.", file=sys.stderr)
        return 1

    with tempfile.TemporaryDirectory(prefix="robot-u-lxc-deploy-key.") as temp_dir:
        key_path = Path(temp_dir) / "id_ed25519"
        subprocess.run(
            [
                ssh_keygen,
                "-t",
                "ed25519",
                "-N",
                "",
                "-C",
                key_comment,
                "-f",
                str(key_path),
            ],
            check=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )

        public_key = key_path.with_suffix(".pub").read_text(encoding="utf-8").strip()
        private_key = key_path.read_text(encoding="utf-8")

        _install_authorized_key(ssh, deploy_user, deploy_host, deploy_port, public_key)
        _test_ssh_login(ssh, deploy_user, deploy_host, deploy_port, key_path)

    _put_actions_secret(base_url, token, owner, repo_name, secret_name, private_key)

    print(f"Installed an SSH public key for {deploy_user}@{deploy_host}.")
    print(f"Updated repository Actions secret {secret_name} for {repo}.")
    print("Private key material was only held in memory and a temporary directory.")
    return 0


def _load_env_file(path: Path) -> None:
    if not path.exists():
        return

    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue

        key, value = line.split("=", 1)
        key = key.strip().removeprefix("export ").strip()
        value = value.strip().strip('"').strip("'")
        os.environ.setdefault(key, value)


def _repo_parts(repo: str) -> tuple[str, str]:
    parts = repo.split("/", 1)
    if len(parts) != 2 or not parts[0] or not parts[1]:
        raise SystemExit("FORGEJO_REPO must use owner/repo format.")
    return parts[0], parts[1]


def _install_authorized_key(
    ssh: str,
    user: str,
    host: str,
    port: str,
    public_key: str,
) -> None:
    encoded_key = base64.b64encode(public_key.encode("utf-8")).decode("ascii")
    remote_script = r"""
set -euo pipefail
key="$(printf '%s' "$1" | base64 -d)"
mkdir -p ~/.ssh
chmod 700 ~/.ssh
touch ~/.ssh/authorized_keys
chmod 600 ~/.ssh/authorized_keys
if ! grep -Fqx "$key" ~/.ssh/authorized_keys; then
  printf '%s\n' "$key" >> ~/.ssh/authorized_keys
fi
"""
    subprocess.run(
        [
            ssh,
            "-p",
            port,
            "-o",
            "StrictHostKeyChecking=accept-new",
            f"{user}@{host}",
            "bash",
            "-s",
            "--",
            encoded_key,
        ],
        input=remote_script,
        text=True,
        check=True,
    )


def _test_ssh_login(
    ssh: str,
    user: str,
    host: str,
    port: str,
    key_path: Path,
) -> None:
    subprocess.run(
        [
            ssh,
            "-p",
            port,
            "-i",
            str(key_path),
            "-o",
            "BatchMode=yes",
            "-o",
            "StrictHostKeyChecking=accept-new",
            f"{user}@{host}",
            "true",
        ],
        check=True,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )


def _put_actions_secret(
    base_url: str,
    token: str,
    owner: str,
    repo: str,
    name: str,
    value: str,
) -> None:
    path = f"/api/v1/repos/{quote(owner)}/{quote(repo)}/actions/secrets/{quote(name)}"
    _request_json(base_url, token, "PUT", path, {"data": value})


def _request_json(
    base_url: str,
    token: str,
    method: str,
    path: str,
    payload: dict[str, object],
) -> dict[str, object]:
    data = json.dumps(payload).encode("utf-8")
    request = Request(
        f"{base_url}{path}",
        data=data,
        method=method,
        headers={
            "Authorization": f"token {token}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        },
    )
    try:
        with urlopen(request, timeout=20) as response:
            if response.status == 204:
                return {}
            raw_body = response.read()
    except HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")
        print(f"Forgejo API returned {error.code}: {detail}", file=sys.stderr)
        raise SystemExit(1) from error

    if not raw_body:
        return {}

    decoded = json.loads(raw_body.decode("utf-8"))
    if isinstance(decoded, dict):
        return decoded
    return {}


if __name__ == "__main__":
    raise SystemExit(main())