climperor/web/cloudflare/put_pages_secrets.py

"""Put STEAM_API_KEY + SESSION_SECRET on climperor-relations Pages (prod+preview).

Reads secrets from env or staged temp files under web/.refresh/ (no secret echo).
Prefers `wrangler pages secret put`; falls back to Pages project PATCH.
"""

from __future__ import annotations

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

ACCT = "510534f7f6284344aadaf2f5a0794d48"
PROJ = "climperor-relations"
HERE = Path(__file__).resolve().parent
FRONTEND = HERE.parents[0] / "frontend"
REFRESH = Path(__file__).resolve().parents[1] / ".refresh"
STEAM_TMP = REFRESH / "steam_key.tmp"
SESSION_TMP = REFRESH / "session_secret.tmp"


def _creds() -> tuple[str, str]:
    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 credentials")
    return email, key


def _secret(name: str, env_keys: list[str], staged: Path) -> str:
    for k in env_keys:
        v = (os.environ.get(k) or "").strip()
        if v:
            return v
    if staged.is_file():
        v = staged.read_text(encoding="utf-8").strip()
        try:
            staged.unlink()
        except OSError:
            pass
        if v:
            return v
    raise SystemExit(f"missing {name}")


def cf(method: str, path: str, body: dict | None = None) -> dict:
    email, key = _creds()
    data = None if body is None else json.dumps(body).encode("utf-8")
    req = urllib.request.Request(
        f"https://api.cloudflare.com/client/v4{path}",
        data=data,
        method=method,
        headers={
            "X-Auth-Email": email,
            "X-Auth-Key": key,
            "Content-Type": "application/json",
            "User-Agent": "climperor-put-pages-secrets",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        raise SystemExit(f"CF API {e.code}: {raw[:500]}") from e


def _wrangler_put(name: str, value: str) -> None:
    email, key = _creds()
    env = os.environ.copy()
    env["CLOUDFLARE_EMAIL"] = email
    env["CLOUDFLARE_API_KEY"] = key
    env["CLOUDFLARE_ACCOUNT_ID"] = ACCT
    env["CI"] = "true"
    print(f"wrangler pages secret put {name} …", flush=True)
    cmd = (
        f'npx --yes wrangler@3 pages secret put {name} '
        f'--project-name {PROJ}'
    )
    proc = subprocess.run(
        cmd,
        input=value + "\n",
        text=True,
        capture_output=True,
        env=env,
        cwd=str(FRONTEND if FRONTEND.is_dir() else HERE),
        shell=True,
    )
    if proc.returncode != 0:
        print(proc.stderr or proc.stdout, file=sys.stderr)
        raise SystemExit(proc.returncode)
    tail = (proc.stdout or "").strip().splitlines()
    if tail:
        print(tail[-1], flush=True)


def main() -> int:
    steam = _secret(
        "STEAM_API_KEY",
        ["STEAM_API_KEY", "KEYZOO_ASSET_SECRET_WEB_API_KEY"],
        STEAM_TMP,
    )
    session = _secret(
        "SESSION_SECRET",
        ["SESSION_SECRET", "KEYZOO_ASSET_SECRET_SESSION_SECRET"],
        SESSION_TMP,
    )
    # Official path: wrangler pages secret put (production).
    _wrangler_put("STEAM_API_KEY", steam)
    _wrangler_put("SESSION_SECRET", session)

    # Also patch preview so *.pages.dev preview deploys work.
    proj = cf("GET", f"/accounts/{ACCT}/pages/projects/{PROJ}")
    result = proj.get("result") or {}
    dc = result.get("deployment_configs") or {}
    secrets = {
        "STEAM_API_KEY": {"type": "secret_text", "value": steam},
        "SESSION_SECRET": {"type": "secret_text", "value": session},
    }
    preview_cfg = dc.get("preview") or {}
    preview_entry: dict = {"env_vars": secrets}
    wch = preview_cfg.get("wrangler_config_hash")
    if wch:
        preview_entry["wrangler_config_hash"] = wch
    out = cf(
        "PATCH",
        f"/accounts/{ACCT}/pages/projects/{PROJ}",
        {"deployment_configs": {"preview": preview_entry}},
    )
    if not out.get("success"):
        print("preview patch soft-fail", flush=True)

    check = cf("GET", f"/accounts/{ACCT}/pages/projects/{PROJ}")
    cdc = (check.get("result") or {}).get("deployment_configs") or {}
    for env_name in ("production", "preview"):
        ev = ((cdc.get(env_name) or {}).get("env_vars") or {})
        names = sorted(ev.keys())
        types = {
            k: (ev[k] or {}).get("type") for k in ("STEAM_API_KEY", "SESSION_SECRET")
        }
        print(f"{env_name}: env_vars={names} secret_types={types}", flush=True)
    print("ok", flush=True)
    return 0


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