| 1 | #!/usr/bin/env python3 |
| 2 | import json |
| 3 | import sys |
| 4 | import urllib.error |
| 5 | import urllib.request |
| 6 | |
| 7 | from api import instance |
| 8 | |
| 9 | |
| 10 | data = json.load(sys.stdin) |
| 11 | request = data["request"] |
| 12 | if request["kind"] != "oauth-client": |
| 13 | raise ValueError("unsupported Forgejo input") |
| 14 | stage = data.get("stageId") |
| 15 | existing = data.get("existing") or {} |
| 16 | if data.get("operation") == "delete" and not stage: |
| 17 | raise ValueError("refusing to delete a production OAuth client") |
| 18 | _, address = instance() |
| 19 | authorization = "token " + data["providerSecrets"]["automation_token"] |
| 20 | |
| 21 | |
| 22 | def api(path, method="GET", body=None, missing_ok=False): |
| 23 | query = urllib.request.Request( |
| 24 | address + "/api/v1" + path, |
| 25 | data=json.dumps(body).encode() if body is not None else None, |
| 26 | method=method, |
| 27 | headers={"Authorization": authorization, "Content-Type": "application/json"}, |
| 28 | ) |
| 29 | try: |
| 30 | with urllib.request.urlopen(query, timeout=10) as response: |
| 31 | return json.load(response) if response.status != 204 else None |
| 32 | except urllib.error.HTTPError as error: |
| 33 | if error.code == 404 and (method == "DELETE" or missing_ok): |
| 34 | return None |
| 35 | raise RuntimeError(f"Forgejo OAuth API returned HTTP {error.code}") from error |
| 36 | |
| 37 | |
| 38 | application_id = existing.get("applicationId") |
| 39 | if data.get("operation") == "delete": |
| 40 | if application_id: |
| 41 | api("/user/applications/oauth2/" + application_id, "DELETE") |
| 42 | sys.exit(0) |
| 43 | |
| 44 | suffix = request["name"] + (":" + stage if stage else "") |
| 45 | name = "Snow Globe: " + suffix |
| 46 | desired = {"name": name, "redirect_uris": request["redirectUris"], "confidential_client": True} |
| 47 | if application_id: |
| 48 | current = api("/user/applications/oauth2/" + application_id, missing_ok=True) |
| 49 | if current and current["client_id"] == existing.get("clientId") and existing.get("clientSecret"): |
| 50 | if any(current[key] != value for key, value in desired.items()): |
| 51 | api("/user/applications/oauth2/" + application_id, "PATCH", desired) |
| 52 | client_id = existing["clientId"] |
| 53 | client_secret = existing["clientSecret"] |
| 54 | else: |
| 55 | application_id = None |
| 56 | if not application_id: |
| 57 | applications = api("/user/applications/oauth2?limit=100") |
| 58 | if any(item["name"] in (name, "studio:" + suffix) for item in applications): |
| 59 | raise ValueError("Forgejo OAuth client exists without its stored secret") |
| 60 | created = api("/user/applications/oauth2", "POST", desired) |
| 61 | application_id = str(created["id"]) |
| 62 | client_id = created["client_id"] |
| 63 | client_secret = created["client_secret"] |
| 64 | if not client_id or not client_secret: |
| 65 | raise ValueError("Forgejo OAuth client has no credentials") |
| 66 | print(json.dumps({"applicationId": application_id, "clientId": client_id, |
| 67 | "clientSecret": client_secret, "providerUrl": "https://" + data["host"]})) |