feat(enterprise): route four domains through lighthouse

deployment/enterprise-domain-bootstrap/enterprise-lighthouse/bootstrap_private_repositories.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+"""Idempotently create the five private enterprise work repositories.
+
+Run on the enterprise node with a short-lived Forgejo admin token stored in a
+root-readable file. The token is never printed. Existing repositories are
+inspected and preserved; a public or wrongly-owned collision fails closed.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+
+def request(base: str, token: str, method: str, path: str, body: dict | None = None):
+    data = json.dumps(body).encode() if body is not None else None
+    call = urllib.request.Request(base + path, data=data, method=method)
+    call.add_header("Authorization", f"token {token}")
+    call.add_header("Accept", "application/json")
+    if data is not None:
+        call.add_header("Content-Type", "application/json")
+    try:
+        with urllib.request.urlopen(call, timeout=20) as response:
+            raw = response.read()
+            return response.status, json.loads(raw) if raw else {}
+    except urllib.error.HTTPError as error:
+        raw = error.read()
+        detail = json.loads(raw) if raw else {}
+        return error.code, detail
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--registry", required=True)
+    parser.add_argument("--token-file", required=True)
+    parser.add_argument("--token-name", required=True)
+    parser.add_argument("--receipt", required=True)
+    parser.add_argument("--base", default="http://127.0.0.1:3341/api/v1")
+    args = parser.parse_args()
+
+    registry = json.loads(Path(args.registry).read_text(encoding="utf-8"))
+    token = Path(args.token_file).read_text(encoding="utf-8").strip()
+    if len(token) < 32:
+        raise SystemExit("short-lived Forgejo token unavailable")
+
+    results = []
+    completed = False
+    try:
+        for human in registry["humans"]:
+            owner, name = human["repository"].split("/", 1)
+            status, existing = request(args.base, token, "GET", f"/repos/{owner}/{name}")
+            action = "PRESERVED"
+            if status == 404:
+                status, existing = request(
+                    args.base,
+                    token,
+                    "POST",
+                    f"/admin/users/{owner}/repos",
+                    {
+                        "name": name,
+                        "description": f"{human['display_name']} · {human['responsibility_domain']} 独立工作仓库",
+                        "private": True,
+                        "auto_init": True,
+                        "default_branch": "main",
+                        "gitignores": "",
+                        "issue_labels": "",
+                        "license": "",
+                        "readme": "Default",
+                    },
+                )
+                action = "CREATED"
+            if status not in (200, 201):
+                raise RuntimeError(f"repository provision failed for {owner}/{name}: HTTP {status}")
+            actual_owner = existing.get("owner", {}).get("login")
+            if actual_owner != owner or existing.get("private") is not True:
+                raise RuntimeError(f"repository boundary invalid for {owner}/{name}")
+            results.append(
+                {
+                    "human_number": human["human_number"],
+                    "repository": f"{owner}/{name}",
+                    "private": True,
+                    "action": action,
+                }
+            )
+        completed = True
+    finally:
+        # Revoke the bootstrap token after success. On failure it remains in the
+        # root-only token file so an operator can inspect and retry deliberately.
+        if completed:
+            revoke_status, _ = request(
+                args.base,
+                token,
+                "DELETE",
+                f"/admin/users/bingshuo/tokens/{args.token_name}",
+            )
+            if revoke_status not in (204, 404):
+                raise RuntimeError(f"bootstrap token revocation failed: HTTP {revoke_status}")
+
+    receipt = {
+        "schema": "guanghu.enterprise-private-repository-bootstrap-receipt/v1",
+        "state": "PASS",
+        "forgejo": "guanghu.chat/code",
+        "repositories": results,
+        "token_revoked": True,
+        "shared_initial_password_used": False,
+        "existing_user_passwords_modified": False,
+    }
+    Path(args.receipt).write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + "\n")
+    print(json.dumps(receipt, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()