TorbatYar/backend/services/identity-access/scripts/patch_keycloak_client.py
Mortezakoohjani 800b0ba2c5 Deploy TorbatYar for torbatyar.ir with nginx multi-tenant routing.
Wire production domain, CORS for tenant subdomains, celery volume mounts, and nginx reverse proxy configs for apex, API, identity, auth, and wildcard tenants.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-21 21:43:33 +03:30

101 lines
3.4 KiB
Python

"""یک‌بار اجرا کنید تا redirect URI دامنه dev به کلاینت frontend در Keycloak اضافه شود."""
from __future__ import annotations
import os
import sys
from urllib.parse import urlparse
import httpx
KEYCLOAK_URL = os.getenv("KEYCLOAK_SERVER_URL", "http://localhost:8080")
REALM = os.getenv("KEYCLOAK_REALM", "superapp")
ADMIN_USER = os.getenv("KEYCLOAK_ADMIN", "admin")
ADMIN_PASS = os.getenv("KEYCLOAK_ADMIN_PASSWORD", "admin")
CLIENT_ID = os.getenv("KEYCLOAK_FRONTEND_CLIENT_ID", "superapp-frontend")
CALLBACK_URL = os.getenv(
"FRONTEND_CALLBACK_URL", "http://torbatyar.xyz:3000/auth/callback"
)
def _dev_origin(callback_url: str) -> str:
parsed = urlparse(callback_url)
if not parsed.scheme or not parsed.netloc:
raise ValueError(f"FRONTEND_CALLBACK_URL نامعتبر است: {callback_url}")
return f"{parsed.scheme}://{parsed.netloc}"
def main() -> int:
origin = _dev_origin(CALLBACK_URL)
extra_redirect = f"{origin}/*"
extra_origin = origin
token_url = f"{KEYCLOAK_URL}/realms/master/protocol/openid-connect/token"
with httpx.Client(timeout=15.0) as client:
token_resp = client.post(
token_url,
data={
"grant_type": "password",
"client_id": "admin-cli",
"username": ADMIN_USER,
"password": ADMIN_PASS,
},
)
token_resp.raise_for_status()
token = token_resp.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
clients_resp = client.get(
f"{KEYCLOAK_URL}/admin/realms/{REALM}/clients",
params={"clientId": CLIENT_ID},
headers=headers,
)
clients_resp.raise_for_status()
clients = clients_resp.json()
if not clients:
print(f"Client not found: {CLIENT_ID}", file=sys.stderr)
return 1
client_uuid = clients[0]["id"]
detail_resp = client.get(
f"{KEYCLOAK_URL}/admin/realms/{REALM}/clients/{client_uuid}",
headers=headers,
)
detail_resp.raise_for_status()
data = detail_resp.json()
redirects = list(data.get("redirectUris") or [])
origins = list(data.get("webOrigins") or [])
changed = False
if extra_redirect not in redirects:
redirects.append(extra_redirect)
changed = True
if extra_origin not in origins:
origins.append(extra_origin)
changed = True
# حذف اجبار PKCE (تبادل توکن سمت سرور توسط identity انجام می‌شود و
# crypto.subtle روی HTTP لوکال در دسترس نیست).
attributes = dict(data.get("attributes") or {})
if attributes.pop("pkce.code.challenge.method", None) is not None:
data["attributes"] = attributes
changed = True
if not changed:
print(f"Keycloak client already configured for {origin}.")
return 0
data["redirectUris"] = redirects
data["webOrigins"] = origins
update_resp = client.put(
f"{KEYCLOAK_URL}/admin/realms/{REALM}/clients/{client_uuid}",
headers=headers,
json=data,
)
update_resp.raise_for_status()
print(f"Keycloak frontend client updated for {origin}.")
return 0
if __name__ == "__main__":
raise SystemExit(main())