tegwick
9b31f229e8
Add runbook coverage for REUSE_SURFACE_TOKEN and REUSE_SURFACE_FORGEJO_WEBHOOK_SECRET in reuse-surface-env, plus an idempotent make reuse-forgejo-webhook target that wires the coulomb org push webhook to the hub receiver.
80 lines
No EOL
2.6 KiB
Bash
Executable file
80 lines
No EOL
2.6 KiB
Bash
Executable file
#!/usr/bin/env bash
|
|
# Idempotent org-level Forgejo webhook for reuse-surface hub recompose (REUSE-WP-0019-T03).
|
|
# Reads the HMAC secret from the live cluster Secret — never prints it.
|
|
set -euo pipefail
|
|
|
|
FORGEJO_API="${FORGEJO_API:-https://forgejo.coulomb.social/api/v1}"
|
|
FORGEJO_ORG="${FORGEJO_ORG:-coulomb}"
|
|
WEBHOOK_URL="${REUSE_WEBHOOK_URL:-https://reuse.coulomb.social/v1/webhooks/forgejo}"
|
|
KUBECONFIG_PATH="${KUBECONFIG:-$HOME/.kube/config-hosteurope}"
|
|
SECRET_NS="${REUSE_SECRET_NAMESPACE:-reuse}"
|
|
SECRET_NAME="${REUSE_SECRET_NAME:-reuse-surface-env}"
|
|
SECRET_KEY="${REUSE_WEBHOOK_SECRET_KEY:-REUSE_SURFACE_FORGEJO_WEBHOOK_SECRET}"
|
|
TOKEN_FILE="${FORGEJO_TOKEN_FILE:-/tmp/forgejo-tegwick-api-token}"
|
|
|
|
if [[ -n "${FORGEJO_ADMIN_TOKEN:-}" ]]; then
|
|
TOKEN="${FORGEJO_ADMIN_TOKEN}"
|
|
elif [[ -f "${TOKEN_FILE}" ]]; then
|
|
TOKEN="$(cat "${TOKEN_FILE}")"
|
|
else
|
|
echo "Set FORGEJO_ADMIN_TOKEN or create ${TOKEN_FILE}" >&2
|
|
exit 1
|
|
fi
|
|
|
|
WEBHOOK_SECRET="$(
|
|
kubectl --kubeconfig "${KUBECONFIG_PATH}" get secret "${SECRET_NAME}" -n "${SECRET_NS}" \
|
|
-o "jsonpath={.data.${SECRET_KEY}}" | base64 -d
|
|
)"
|
|
if [[ -z "${WEBHOOK_SECRET}" ]]; then
|
|
echo "Missing ${SECRET_KEY} in ${SECRET_NS}/${SECRET_NAME}" >&2
|
|
exit 1
|
|
fi
|
|
|
|
export FORGEJO_API FORGEJO_ORG WEBHOOK_URL TOKEN WEBHOOK_SECRET
|
|
python3 <<'PY'
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
api = os.environ["FORGEJO_API"].rstrip("/")
|
|
org = os.environ["FORGEJO_ORG"]
|
|
url = os.environ["WEBHOOK_URL"]
|
|
token = os.environ["TOKEN"]
|
|
secret = os.environ["WEBHOOK_SECRET"]
|
|
headers = {"Authorization": f"token {token}"}
|
|
|
|
def api_json(method: str, path: str, payload: dict | None = None):
|
|
data = None if payload is None else json.dumps(payload).encode()
|
|
req = urllib.request.Request(
|
|
f"{api}{path}",
|
|
data=data,
|
|
headers={**headers, **({} if payload is None else {"Content-Type": "application/json"})},
|
|
method=method,
|
|
)
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.load(resp)
|
|
|
|
hooks = api_json("GET", f"/orgs/{org}/hooks")
|
|
for hook in hooks:
|
|
cfg = hook.get("config") or {}
|
|
if cfg.get("url") == url and hook.get("active") and "push" in (hook.get("events") or []):
|
|
print(f"==> Forgejo org webhook already configured (id={hook['id']})")
|
|
raise SystemExit(0)
|
|
|
|
created = api_json(
|
|
"POST",
|
|
f"/orgs/{org}/hooks",
|
|
{
|
|
"type": "gitea",
|
|
"config": {
|
|
"url": url,
|
|
"content_type": "json",
|
|
"secret": secret,
|
|
},
|
|
"events": ["push"],
|
|
"active": True,
|
|
},
|
|
)
|
|
print(f"==> Forgejo org webhook created (id={created['id']}) -> {url}")
|
|
PY |