80 lines
2.6 KiB
Bash
80 lines
2.6 KiB
Bash
|
#!/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
|