Sécuriser Gitea et alerter sur les échecs CI

ops/setup-gitea-webhooks.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+import json
+import os
+import secrets
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+
+GITEA_URL = os.environ.get("GITEA_URL", "https://git.recette.santinova-soft.org").rstrip("/")
+GITEA_USER = os.environ.get("GITEA_USERNAME", "manus-admin")
+DASHBOARD_URL = os.environ.get("DASHBOARD_URL", "https://dashboard.recette.santinova-soft.org").rstrip("/")
+DASHBOARD_APP_DIR = Path(os.environ.get("DASHBOARD_APP_DIR", "/opt/manus-deploy/apps/manus-dashboard"))
+ENV_FILE = DASHBOARD_APP_DIR / ".env"
+WEBHOOK_ENDPOINT = f"{DASHBOARD_URL}/api/webhook/gitea"
+
+
+def read_env() -> dict[str, str]:
+    values: dict[str, str] = {}
+    if ENV_FILE.exists():
+        for line in ENV_FILE.read_text(encoding="utf-8").splitlines():
+            if line and not line.lstrip().startswith("#") and "=" in line:
+                key, value = line.split("=", 1)
+                values[key.strip()] = value.strip()
+    return values
+
+
+def update_env(key: str, value: str) -> None:
+    lines = ENV_FILE.read_text(encoding="utf-8").splitlines() if ENV_FILE.exists() else []
+    updated: list[str] = []
+    found = False
+    for line in lines:
+        if line.startswith(f"{key}="):
+            updated.append(f"{key}={value}")
+            found = True
+        else:
+            updated.append(line)
+    if not found:
+        updated.append(f"{key}={value}")
+    temporary = ENV_FILE.with_suffix(".env.next")
+    temporary.write_text("\n".join(updated).rstrip() + "\n", encoding="utf-8")
+    temporary.chmod(0o600)
+    temporary.replace(ENV_FILE)
+
+
+def api(token: str, method: str, path: str, payload: dict | None = None):
+    data = json.dumps(payload).encode() if payload is not None else None
+    request = urllib.request.Request(
+        f"{GITEA_URL}/api/v1{path}",
+        method=method,
+        data=data,
+        headers={"Authorization": f"token {token}", "Content-Type": "application/json"},
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=20) as response:
+            body = response.read()
+            return json.loads(body) if body else None
+    except urllib.error.HTTPError as error:
+        detail = error.read().decode(errors="replace")
+        raise RuntimeError(f"Gitea {method} {path} : HTTP {error.code} {detail[:300]}") from error
+
+
+def ensure_dashboard_secret(values: dict[str, str]) -> tuple[dict[str, str], bool]:
+    if values.get("WEBHOOK_SECRET"):
+        return values, False
+    update_env("WEBHOOK_SECRET", secrets.token_hex(32))
+    return read_env(), True
+
+
+def main() -> None:
+    values, generated = ensure_dashboard_secret(read_env())
+    token = values.get("GITEA_TOKEN", "")
+    secret = values.get("WEBHOOK_SECRET", "")
+    if not token or not secret:
+        raise RuntimeError("GITEA_TOKEN et WEBHOOK_SECRET sont requis dans le .env serveur.")
+    if generated:
+        subprocess.run(["docker", "compose", "up", "-d", "--force-recreate", "dashboard"], cwd=DASHBOARD_APP_DIR, check=True)
+        time.sleep(8)
+    identity = api(token, "GET", "/user")
+    if identity.get("login") != GITEA_USER:
+        raise RuntimeError("Le jeton Gitea n’appartient pas au compte attendu.")
+    repositories = api(token, "GET", "/user/repos?limit=100&page=1") or []
+    configured = 0
+    for repository in repositories:
+        owner = repository.get("owner", {}).get("login")
+        name = repository.get("name")
+        if owner != GITEA_USER or not name:
+            continue
+        hooks = api(token, "GET", f"/repos/{owner}/{name}/hooks") or []
+        existing = next((hook for hook in hooks if hook.get("config", {}).get("url") == WEBHOOK_ENDPOINT), None)
+        payload = {
+            "active": True,
+            "branch_filter": "main",
+            "config": {"url": WEBHOOK_ENDPOINT, "content_type": "json", "secret": secret},
+            "events": ["push", "workflow_run"],
+        }
+        if existing:
+            api(token, "PATCH", f"/repos/{owner}/{name}/hooks/{existing['id']}", payload)
+        else:
+            api(token, "POST", f"/repos/{owner}/{name}/hooks", {"type": "gitea", **payload})
+        configured += 1
+        print(f"WEBHOOK_OK {owner}/{name}")
+    if configured == 0:
+        raise RuntimeError("Aucun dépôt Gitea n’a été configuré.")
+    print(f"GITEA_WEBHOOKS_CONFIGURED {configured}")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as exc:
+        print(f"GITEA_WEBHOOK_SETUP_FAILED: {exc}", file=sys.stderr)
+        raise SystemExit(1)