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}"
|