104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
"""Forgejo / Gitea HTTP API: Authorization: token <secret>."""
|
||
|
||
import httpx
|
||
|
||
|
||
def _auth(token: str) -> dict[str, str]:
|
||
return {
|
||
"Authorization": f"token {token}",
|
||
"Accept": "application/json",
|
||
}
|
||
|
||
|
||
async def probe(base_url: str, token: str) -> dict:
|
||
"""
|
||
Проверка токена. Сначала GET /api/v1/user (нужен scope read:user).
|
||
При 403 из‑за scope — fallback: GET /api/v1/admin/config (часто доступен с write:admin).
|
||
"""
|
||
if not base_url.strip() or not token.strip():
|
||
return {"status": "error", "detail": "forgejo url or token empty"}
|
||
|
||
base = base_url.rstrip("/")
|
||
h = _auth(token)
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15.0, verify=True, follow_redirects=True) as client:
|
||
ru = await client.get(f"{base}/api/v1/user", headers=h)
|
||
if ru.status_code == 200:
|
||
try:
|
||
u = ru.json()
|
||
except Exception:
|
||
u = {}
|
||
login = u.get("login")
|
||
out: dict = {
|
||
"status": "ok",
|
||
"url": base,
|
||
"api": "authenticated",
|
||
"scope": "read:user",
|
||
}
|
||
if login:
|
||
out["login"] = login
|
||
return out
|
||
|
||
if ru.status_code == 403:
|
||
for path, name in (
|
||
("/api/v1/admin/config", "admin_config"),
|
||
("/api/v1/notifications?limit=1", "notifications"),
|
||
):
|
||
rx = await client.get(f"{base}{path}", headers=h)
|
||
if rx.status_code == 200:
|
||
return {
|
||
"status": "ok",
|
||
"url": base,
|
||
"api": "authenticated",
|
||
"scope_note": (
|
||
"в токене нет scope read:user — в Forgejo включи read:user у PAT "
|
||
"или создай новый токен с read:user, чтобы отображался login"
|
||
),
|
||
"fallback": name,
|
||
}
|
||
|
||
body = (ru.text or "")[:500]
|
||
return {
|
||
"status": "error",
|
||
"url": base,
|
||
"detail": f"http {ru.status_code}: {body}",
|
||
}
|
||
except Exception as e:
|
||
return {"status": "error", "detail": str(e), "url": base_url.rstrip("/")}
|
||
|
||
|
||
async def ping(base_url: str, token: str) -> tuple[bool, str | None]:
|
||
"""Обёртка для совместимости: True если status ok."""
|
||
r = await probe(base_url, token)
|
||
if r.get("status") == "ok":
|
||
return True, None
|
||
return False, r.get("detail", "unknown")
|
||
|
||
|
||
async def get_user(base_url: str, token: str) -> tuple[dict | None, str | None]:
|
||
base = base_url.rstrip("/")
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15.0, verify=True, follow_redirects=True) as client:
|
||
r = await client.get(f"{base}/api/v1/user", headers=_auth(token))
|
||
except Exception as e:
|
||
return None, str(e)
|
||
if r.status_code != 200:
|
||
return None, f"http {r.status_code}"
|
||
try:
|
||
return r.json(), None
|
||
except Exception:
|
||
return None, "invalid json"
|
||
|
||
|
||
async def health_public(base_url: str) -> tuple[bool, str | None]:
|
||
base = base_url.rstrip("/")
|
||
try:
|
||
async with httpx.AsyncClient(timeout=10.0, verify=True) as client:
|
||
r = await client.get(f"{base}/api/v1/version")
|
||
except Exception as e:
|
||
return False, str(e)
|
||
if r.status_code == 200:
|
||
return True, None
|
||
return False, f"http {r.status_code}"
|