JiriUhlir
290 lines
10 KiB
Python
290 lines
10 KiB
Python
import json
|
||
import secrets
|
||
import string
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.parse import quote, urlencode
|
||
from urllib.request import Request as UrlRequest
|
||
from urllib.request import urlopen
|
||
|
||
from app.config import get_gitea_admin_token, get_gitea_server_url
|
||
from app.db.users import update_gitea_sync_state
|
||
|
||
GITEA_ACCOUNT_ROLES = {"admin", "developer"}
|
||
|
||
|
||
def desired_gitea_username(user: dict) -> str:
|
||
"""Cílové Gitea uživatelské jméno: vlastní (nastavené adminem), jinak automatické portal-{id}."""
|
||
custom = (user.get("gitea_username") or "").strip()
|
||
if custom:
|
||
return custom
|
||
return _gitea_username(int(user["id"]))
|
||
|
||
|
||
def apply_gitea_credentials(user: dict, new_username: str = "", new_password: str = "") -> dict:
|
||
"""Admin nastaví Gitea uživatelské jméno a/nebo heslo přes admin API.
|
||
Vrací {"status": "synced"/"error", "applied": [...], "gitea_username", "gitea_user_id"}.
|
||
Heslo se nikam neukládá – jen se předá do Gitea."""
|
||
new_username = (new_username or "").strip()
|
||
new_password = new_password or ""
|
||
|
||
if not new_username and not new_password:
|
||
return {"status": "error", "error": "Nebyla zadána žádná změna."}
|
||
if not _requires_gitea_account(user):
|
||
return {
|
||
"status": "error",
|
||
"error": "Uživatel nemá mít Gitea účet (potřebuje roli admin/developer a aktivní účet).",
|
||
}
|
||
|
||
try:
|
||
current_username = desired_gitea_username(user)
|
||
gitea_user = _ensure_gitea_user(user, current_username)
|
||
current_login = _gitea_login(gitea_user) or current_username
|
||
|
||
applied: list[str] = []
|
||
if new_username and new_username != current_login:
|
||
_rename_gitea_user(current_login, new_username)
|
||
current_login = new_username
|
||
applied.append("username")
|
||
if new_password:
|
||
_set_gitea_password(current_login, user, new_password, gitea_user=gitea_user)
|
||
applied.append("password")
|
||
|
||
return {
|
||
"status": "synced",
|
||
"applied": applied,
|
||
"gitea_username": current_login,
|
||
"gitea_user_id": int(gitea_user["id"]),
|
||
}
|
||
except Exception as exc:
|
||
return {"status": "error", "error": str(exc)[:500]}
|
||
|
||
|
||
def _rename_gitea_user(current_username: str, new_username: str) -> dict:
|
||
# Gitea admin rename endpoint (Gitea >= 1.20). Starší Gitea endpoint nemá → vrátí HTTP 404,
|
||
# což apply_gitea_credentials zachytí a nahlásí jako srozumitelnou chybu.
|
||
return _request_json(
|
||
"POST",
|
||
f"/api/v1/admin/users/{quote(current_username)}/rename",
|
||
{"new_username": new_username},
|
||
expected=(200, 204),
|
||
)
|
||
|
||
|
||
def _set_gitea_password(username: str, user: dict, password: str, gitea_user: dict | None = None) -> dict:
|
||
payload = {
|
||
"login_name": _gitea_login_name(gitea_user) or username,
|
||
"source_id": _gitea_source_id(gitea_user),
|
||
"email": _email(user),
|
||
"full_name": _display_name(user),
|
||
"password": password,
|
||
"must_change_password": False,
|
||
}
|
||
return _request_json("PATCH", f"/api/v1/admin/users/{quote(username)}", payload, expected=(200,))
|
||
|
||
|
||
def sync_gitea_user(user: dict) -> dict:
|
||
user_id = int(user["id"])
|
||
desired_active = _requires_gitea_account(user)
|
||
username = desired_gitea_username(user)
|
||
|
||
update_gitea_sync_state(user_id, "syncing")
|
||
try:
|
||
if desired_active:
|
||
gitea_user = _ensure_gitea_user(user, username)
|
||
gitea_username = _gitea_login(gitea_user) or username
|
||
_update_gitea_user(gitea_username, user, active=True, gitea_user=gitea_user)
|
||
update_gitea_sync_state(
|
||
user_id,
|
||
"synced",
|
||
gitea_user_id=int(gitea_user["id"]),
|
||
synced=True,
|
||
)
|
||
return {"status": "synced", "action": "activated", "gitea_user_id": gitea_user["id"]}
|
||
|
||
if not user.get("gitea_user_id"):
|
||
update_gitea_sync_state(user_id, "not_required", synced=True)
|
||
return {"status": "not_required", "action": "none"}
|
||
|
||
gitea_user = _find_linked_gitea_user(user, username)
|
||
if gitea_user:
|
||
gitea_username = _gitea_login(gitea_user) or username
|
||
_update_gitea_user(gitea_username, user, active=False, gitea_user=gitea_user)
|
||
update_gitea_sync_state(
|
||
user_id,
|
||
"not_required",
|
||
gitea_user_id=int(gitea_user["id"]),
|
||
synced=True,
|
||
)
|
||
return {"status": "not_required", "action": "deactivated", "gitea_user_id": gitea_user["id"]}
|
||
|
||
update_gitea_sync_state(user_id, "not_required", synced=True)
|
||
return {"status": "not_required", "action": "none"}
|
||
except Exception as exc:
|
||
error = str(exc)[:500]
|
||
update_gitea_sync_state(user_id, "error", error=error)
|
||
return {"status": "error", "error": error}
|
||
|
||
|
||
def _requires_gitea_account(user: dict) -> bool:
|
||
role = (user.get("role") or "").lower()
|
||
return role in GITEA_ACCOUNT_ROLES and bool(user.get("is_enabled", 1)) and bool(user.get("is_active", 1))
|
||
|
||
|
||
def _gitea_username(user_id: int) -> str:
|
||
return f"portal-{user_id}"
|
||
|
||
|
||
def _ensure_gitea_user(user: dict, username: str) -> dict:
|
||
existing = _get_gitea_user(username, missing_ok=True)
|
||
if existing:
|
||
return existing
|
||
|
||
existing = _find_gitea_user_by_email(_email(user))
|
||
if existing:
|
||
return existing
|
||
|
||
return _create_gitea_user(user, username)
|
||
|
||
|
||
def _create_gitea_user(user: dict, username: str) -> dict:
|
||
payload = {
|
||
"username": username,
|
||
"login_name": username,
|
||
"source_id": 0,
|
||
"email": _email(user),
|
||
"full_name": _display_name(user),
|
||
"password": _random_password(),
|
||
"must_change_password": False,
|
||
"send_notify": False,
|
||
"visibility": "private",
|
||
}
|
||
return _request_json("POST", "/api/v1/admin/users", payload, expected=(201,))
|
||
|
||
|
||
def _update_gitea_user(username: str, user: dict, active: bool, gitea_user: dict | None = None) -> dict:
|
||
payload = {
|
||
"email": _email(user),
|
||
"full_name": _display_name(user),
|
||
"login_name": _gitea_login_name(gitea_user) or username,
|
||
"source_id": _gitea_source_id(gitea_user),
|
||
"active": active,
|
||
"admin": bool((gitea_user or {}).get("is_admin", False)),
|
||
"visibility": "private",
|
||
}
|
||
return _request_json("PATCH", f"/api/v1/admin/users/{quote(username)}", payload, expected=(200,))
|
||
|
||
|
||
def _get_gitea_user(username: str, missing_ok: bool = False) -> dict | None:
|
||
try:
|
||
return _request_json("GET", f"/api/v1/users/{quote(username)}", expected=(200,))
|
||
except RuntimeError as exc:
|
||
if missing_ok and "HTTP 404" in str(exc):
|
||
return None
|
||
raise
|
||
|
||
|
||
def _find_linked_gitea_user(user: dict, username: str) -> dict | None:
|
||
existing = _get_gitea_user(username, missing_ok=True)
|
||
if existing:
|
||
return existing
|
||
|
||
gitea_user_id = user.get("gitea_user_id")
|
||
if not gitea_user_id:
|
||
return None
|
||
|
||
existing = _find_gitea_user_by_email(_email(user))
|
||
if existing and int(existing.get("id") or 0) == int(gitea_user_id):
|
||
return existing
|
||
|
||
return None
|
||
|
||
|
||
def _find_gitea_user_by_email(email: str) -> dict | None:
|
||
email = (email or "").strip().lower()
|
||
if not email:
|
||
return None
|
||
|
||
response = _request_json("GET", f"/api/v1/admin/users?{urlencode({'q': email})}", expected=(200,))
|
||
users = response if isinstance(response, list) else response.get("data", [])
|
||
for gitea_user in users:
|
||
if (gitea_user.get("email") or "").strip().lower() == email:
|
||
return gitea_user
|
||
|
||
return None
|
||
|
||
|
||
def _gitea_login(gitea_user: dict | None) -> str:
|
||
if not gitea_user:
|
||
return ""
|
||
return (gitea_user.get("login") or gitea_user.get("username") or "").strip()
|
||
|
||
|
||
def _gitea_login_name(gitea_user: dict | None) -> str:
|
||
if not gitea_user:
|
||
return ""
|
||
return (gitea_user.get("login_name") or _gitea_login(gitea_user)).strip()
|
||
|
||
|
||
def _gitea_source_id(gitea_user: dict | None) -> int:
|
||
if not gitea_user:
|
||
return 0
|
||
try:
|
||
return int(gitea_user.get("source_id") or 0)
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
|
||
|
||
def _request_json(method: str, path: str, payload: dict | None = None, expected: tuple[int, ...] = (200,)) -> dict:
|
||
base_url = get_gitea_server_url().rstrip("/")
|
||
token = get_gitea_admin_token()
|
||
if not base_url:
|
||
raise RuntimeError("Gitea server URL is not configured")
|
||
if not token:
|
||
raise RuntimeError("Gitea admin token is not configured")
|
||
|
||
data = json.dumps(payload).encode("utf-8") if payload is not None else None
|
||
request = UrlRequest(
|
||
f"{base_url}{path}",
|
||
data=data,
|
||
headers={
|
||
"Accept": "application/json",
|
||
"Authorization": f"token {token}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
method=method,
|
||
)
|
||
try:
|
||
with urlopen(request, timeout=10) as response:
|
||
body = response.read().decode("utf-8")
|
||
if response.status not in expected:
|
||
raise RuntimeError(f"Gitea API returned HTTP {response.status}")
|
||
return json.loads(body) if body else {}
|
||
except HTTPError as exc:
|
||
detail = exc.read().decode("utf-8", errors="replace")
|
||
raise RuntimeError(f"Gitea API returned HTTP {exc.code}: {detail}") from exc
|
||
except (URLError, TimeoutError, json.JSONDecodeError) as exc:
|
||
raise RuntimeError(f"Gitea API request failed for {base_url}{path}: {exc}") from exc
|
||
|
||
|
||
def _email(user: dict) -> str:
|
||
email = (user.get("email") or "").strip().lower()
|
||
if email:
|
||
return email
|
||
return f"portal-{int(user['id'])}@localhost"
|
||
|
||
|
||
def _display_name(user: dict) -> str:
|
||
return (user.get("display_name") or user.get("username") or _email(user)).strip()
|
||
|
||
|
||
def _random_password() -> str:
|
||
alphabet = string.ascii_letters + string.digits
|
||
required = [
|
||
secrets.choice(string.ascii_lowercase),
|
||
secrets.choice(string.ascii_uppercase),
|
||
secrets.choice(string.digits),
|
||
]
|
||
required.extend(secrets.choice(alphabet) for _ in range(29))
|
||
secrets.SystemRandom().shuffle(required)
|
||
return "".join(required)
|