116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
#!/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()
|