appfactory-portal/app/gitea_provisioning.py

import json
import secrets
import string
from urllib.error import HTTPError, URLError
from urllib.parse import quote
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 sync_gitea_user(user: dict) -> dict:
    user_id = int(user["id"])
    desired_active = _requires_gitea_account(user)
    username = _gitea_username(user_id)

    update_gitea_sync_state(user_id, "syncing")
    try:
        if desired_active:
            gitea_user = _ensure_gitea_user(user, username)
            _update_gitea_user(username, user, active=True)
            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 = _get_gitea_user(username, missing_ok=True)
        if gitea_user:
            _update_gitea_user(username, user, active=False)
            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

    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) -> dict:
    payload = {
        "email": _email(user),
        "full_name": _display_name(user),
        "login_name": username,
        "active": active,
        "admin": False,
        "visibility": "private",
    }
    return _request_json("PUT", 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/admin/users/{quote(username)}", expected=(200,))
    except RuntimeError as exc:
        if missing_ok and "HTTP 404" in str(exc):
            return None
        raise


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: {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)