climperor/_gitea_actions_secrets.py

"""Stage / apply Gitea Actions secrets for site-traffic-notify.

Pass 1 (feishu):     python _gitea_actions_secrets.py stash-feishu
Pass 2 (cloudflare):  python _gitea_actions_secrets.py stash-cf
Pass 3 (gitea PAT):  python _gitea_actions_secrets.py apply

Staging file is user-only and deleted after apply. Never prints secret values.
"""

from __future__ import annotations

import json
import os
import stat
import sys
import urllib.error
import urllib.request
from pathlib import Path

STASH = Path(os.environ.get("TEMP") or os.environ.get("TMP") or ".") / "climperor_actions_secrets.json"
GITEA_URL = os.environ.get("GITEA_URL") or os.environ.get("KEYZOO_ASSET_META_URL") or "https://gitea.refining.dev"
OWNER = "refining"
REPO = "climperor"
SECRET_NAMES = (
    "CLOUDFLARE_EMAIL",
    "CLOUDFLARE_API_KEY",
    "FEISHU_WEBHOOK_URL",
)


def _load() -> dict:
    if not STASH.exists():
        return {}
    return json.loads(STASH.read_text(encoding="utf-8"))


def _save(data: dict) -> None:
    STASH.write_text(json.dumps(data), encoding="utf-8")
    try:
        os.chmod(STASH, stat.S_IRUSR | stat.S_IWUSR)
    except OSError:
        pass


def stash_feishu() -> None:
    url = os.environ.get("KEYZOO_ASSET_SECRET_FEISHU_WEBHOOK_URL") or os.environ.get(
        "FEISHU_WEBHOOK_URL"
    )
    if not url:
        raise SystemExit("missing FEISHU webhook in env")
    data = _load()
    data["FEISHU_WEBHOOK_URL"] = url
    _save(data)
    print(f"stashed FEISHU_WEBHOOK_URL ({STASH.name})")


def stash_cf() -> None:
    email = os.environ.get("CLOUDFLARE_EMAIL") or os.environ.get(
        "KEYZOO_ASSET_META_USERNAME"
    )
    key = os.environ.get("CLOUDFLARE_API_KEY") or os.environ.get(
        "KEYZOO_ASSET_SECRET_GLOBAL_API_KEY"
    )
    if not email or not key:
        raise SystemExit("missing Cloudflare email/key in env")
    data = _load()
    data["CLOUDFLARE_EMAIL"] = email
    data["CLOUDFLARE_API_KEY"] = key
    _save(data)
    print(f"stashed CLOUDFLARE_* ({STASH.name})")


def _token() -> str:
    # Prefer keyzoo-injected secrets over a possibly stale GITEA_TOKEN in the shell.
    candidates = (
        "KEYZOO_ASSET_SECRET_PERSONAL_ACCESS_TOKEN_GITEA_1",
        "KEYZOO_ASSET_SECRET_PERSONAL_ACCESS_TOKEN__GITEA_1",
        "KEYZOO_ASSET_TOKEN",
        "KEYZOO_ASSET_API_KEY",
        "GITEA_TOKEN",
    )
    for name in candidates:
        tok = os.environ.get(name)
        if tok:
            print(f"using token from {name} (len={len(tok)})")
            return tok
    raise SystemExit("missing Gitea token in env")


def _put_secret(name: str, value: str, token: str) -> None:
    # Gitea Actions: PUT /api/v1/repos/{owner}/{repo}/actions/secrets/{secretname}
    url = f"{GITEA_URL.rstrip('/')}/api/v1/repos/{OWNER}/{REPO}/actions/secrets/{name}"
    body = json.dumps({"data": value}).encode()
    req = urllib.request.Request(
        url,
        data=body,
        method="PUT",
        headers={
            "Authorization": f"token {token}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as r:
            code = r.status
            _ = r.read()
    except urllib.error.HTTPError as e:
        raw = e.read().decode(errors="replace")
        raise SystemExit(f"PUT {name} HTTP {e.code}: {raw[:500]}") from e
    if code not in (201, 204, 200):
        raise SystemExit(f"PUT {name} unexpected status {code}")
    print(f"ok: {name}")


def apply() -> None:
    data = _load()
    present = [n for n in SECRET_NAMES if data.get(n)]
    missing = [n for n in SECRET_NAMES if n not in present]
    if not present:
        raise SystemExit(f"stash empty; expected one of {SECRET_NAMES}")
    if missing:
        print(f"note: not in stash (left unchanged on server): {missing}")
    token = _token()
    base = GITEA_URL.rstrip("/")
    # Sanity: can we see the repo?
    req = urllib.request.Request(
        f"{base}/api/v1/repos/{OWNER}/{REPO}",
        headers={"Authorization": f"token {token}", "Accept": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=60) as r:
        repo = json.loads(r.read().decode())
    print(f"repo: {repo.get('full_name')} (id={repo.get('id')})")

    for name in present:
        _put_secret(name, data[name], token)

    try:
        STASH.unlink()
    except OSError:
        pass
    print("stash deleted; secrets applied")


def list_secrets() -> None:
    token = _token()
    url = f"{GITEA_URL.rstrip('/')}/api/v1/repos/{OWNER}/{REPO}/actions/secrets"
    req = urllib.request.Request(
        url,
        headers={"Authorization": f"token {token}", "Accept": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as r:
            rows = json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"list secrets HTTP {e.code}: {e.read().decode()[:500]}") from e
    names = sorted(x.get("name") for x in (rows or []))
    print("secrets:", names)


def main() -> int:
    if len(sys.argv) < 2:
        raise SystemExit("usage: stash-feishu | stash-cf | apply | list")
    cmd = sys.argv[1]
    if cmd == "stash-feishu":
        stash_feishu()
    elif cmd == "stash-cf":
        stash_cf()
    elif cmd == "apply":
        apply()
    elif cmd == "list":
        list_secrets()
    else:
        raise SystemExit(f"unknown cmd: {cmd}")
    return 0


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