3ee63b3018
New /api/v1/infra/stack/status probes all stack apps/APIs/SW; Infra UI groups proc-cards by VM; wire vm123 router; menu INFRA COD and Serviços IaaS · Infra as Code labels. Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
"""OpenPanel Community bridge client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
BRIDGE_URL = os.getenv("OPENPANEL_BRIDGE_URL", "http://10.10.10.123:18087").rstrip("/")
|
|
BRIDGE_TOKEN = os.getenv("OPENPANEL_BRIDGE_TOKEN", "")
|
|
OPENADMIN_URL = os.getenv("OPENADMIN_URL", "https://admin.openpanel.ligbox.com.br")
|
|
OPENPANEL_URL = os.getenv("OPENPANEL_URL", "https://openpanel.ligbox.com.br")
|
|
DEFAULT_PLAN = os.getenv("OPENPANEL_DEFAULT_PLAN", "ligbox-site-cms")
|
|
|
|
|
|
class OpenPanelBridgeError(Exception):
|
|
pass
|
|
|
|
|
|
def bridge_configured() -> bool:
|
|
return bool(BRIDGE_TOKEN)
|
|
|
|
|
|
def _headers() -> dict[str, str]:
|
|
if not bridge_configured():
|
|
raise OpenPanelBridgeError("OPENPANEL_BRIDGE_TOKEN ausente")
|
|
return {"Authorization": f"Bearer {BRIDGE_TOKEN}"}
|
|
|
|
|
|
def autologin_payload(username: str) -> dict[str, Any]:
|
|
"""MVP: devolve URL OpenAdmin + instrução CONNECT (Enterprise futuro)."""
|
|
return {
|
|
"username": username,
|
|
"openadmin_url": OPENADMIN_URL,
|
|
"openpanel_url": OPENPANEL_URL,
|
|
"note": "CONNECT autologin requer OpenPanel Enterprise API — use OpenAdmin manualmente",
|
|
"bridge_configured": bridge_configured(),
|
|
}
|
|
|
|
|
|
def health() -> dict[str, Any]:
|
|
if not bridge_configured():
|
|
return {"ok": False, "reason": "OPENPANEL_BRIDGE_TOKEN ausente", "bridge_url": BRIDGE_URL}
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.get(f"{BRIDGE_URL}/api", headers=_headers())
|
|
body: dict[str, Any] = {}
|
|
try:
|
|
body = res.json()
|
|
except Exception:
|
|
pass
|
|
return {
|
|
"ok": res.status_code < 400,
|
|
"status": res.status_code,
|
|
"bridge_url": BRIDGE_URL,
|
|
"bridge": body.get("bridge"),
|
|
}
|
|
except Exception as exc:
|
|
return {"ok": False, "reason": str(exc), "bridge_url": BRIDGE_URL}
|
|
|
|
|
|
def list_users() -> dict[str, Any]:
|
|
with httpx.Client(timeout=30.0) as client:
|
|
res = client.get(f"{BRIDGE_URL}/api/users", headers=_headers())
|
|
if res.status_code >= 400:
|
|
raise OpenPanelBridgeError(f"bridge list_users HTTP {res.status_code}: {res.text[:300]}")
|
|
return res.json()
|
|
|
|
|
|
def get_user(username: str) -> dict[str, Any]:
|
|
with httpx.Client(timeout=30.0) as client:
|
|
res = client.get(f"{BRIDGE_URL}/api/users/{username}", headers=_headers())
|
|
if res.status_code >= 400:
|
|
raise OpenPanelBridgeError(f"bridge get_user HTTP {res.status_code}: {res.text[:300]}")
|
|
return res.json()
|
|
|
|
|
|
def provision_user(
|
|
*,
|
|
username: str,
|
|
password: str,
|
|
email: str,
|
|
domain: str,
|
|
plan_name: str | None = None,
|
|
) -> dict[str, Any]:
|
|
payload = {
|
|
"username": username.strip().lower(),
|
|
"password": password,
|
|
"email": email,
|
|
"domain": domain.strip().lower(),
|
|
"plan_name": plan_name or DEFAULT_PLAN,
|
|
}
|
|
with httpx.Client(timeout=180.0) as client:
|
|
res = client.post(f"{BRIDGE_URL}/api/users", headers=_headers(), json=payload)
|
|
data = res.json() if res.content else {}
|
|
if res.status_code >= 400 or not data.get("success", True):
|
|
raise OpenPanelBridgeError(data.get("error") or f"bridge provision HTTP {res.status_code}")
|
|
return data
|
|
|
|
|
|
def add_domain(*, username: str, domain: str) -> dict[str, Any]:
|
|
payload = {"username": username.strip().lower(), "domain": domain.strip().lower()}
|
|
with httpx.Client(timeout=180.0) as client:
|
|
res = client.post(f"{BRIDGE_URL}/api/domains", headers=_headers(), json=payload)
|
|
data = res.json() if res.content else {}
|
|
if res.status_code >= 400 or not data.get("success", True):
|
|
raise OpenPanelBridgeError(data.get("error") or f"bridge add_domain HTTP {res.status_code}")
|
|
return data
|
|
|
|
|
|
def delete_user(username: str) -> dict[str, Any]:
|
|
with httpx.Client(timeout=120.0) as client:
|
|
res = client.delete(f"{BRIDGE_URL}/api/users/{username.strip().lower()}", headers=_headers())
|
|
data = res.json() if res.content else {}
|
|
if res.status_code >= 400 or not data.get("success", True):
|
|
raise OpenPanelBridgeError(data.get("error") or f"bridge delete HTTP {res.status_code}")
|
|
return data
|