Add CI deployment to app LXC

scripts/bootstrap_lxc_deploy_key.py

@@ -0,0 +1,227 @@
+#!/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())

scripts/check_python_quality.sh

@@ -12,6 +12,7 @@ python_files=(
   "prototype_cache.py"
   "settings.py"
   "scripts/bootstrap_ci_clone_key.py"
+  "scripts/bootstrap_lxc_deploy_key.py"
   "scripts/check_deploy_config.py"
   "update_events.py"
   "tests"
@@ -57,7 +58,7 @@ run_check \
 run_check \
   "Vulture" \
   uv run --with "vulture>=2.15,<3.0.0" \
-  vulture app.py auth.py calendar_feeds.py forgejo_client.py live_prototype.py prototype_cache.py settings.py scripts/bootstrap_ci_clone_key.py scripts/check_deploy_config.py update_events.py tests --min-confidence 80
+  vulture app.py auth.py calendar_feeds.py forgejo_client.py live_prototype.py prototype_cache.py settings.py scripts/bootstrap_ci_clone_key.py scripts/bootstrap_lxc_deploy_key.py scripts/check_deploy_config.py update_events.py tests --min-confidence 80
 run_check \
   "Backend Tests" \
   "${python_cmd[@]}" -m unittest discover -s tests -p "test_*.py"