appfactory-portal/app/gitea_provisioning.py

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
from app.logging_config import get_logger

logger = get_logger(__name__)

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.

    Heslo a přejmenování se aplikují NEZÁVISLE – když jedno selže, druhé se i tak provede a do
    výsledku se vrátí přesný důvod selhání (chybí endpoint na staré Gitea, jméno obsazené apod.).
    Vrací: {"status": synced|partial|error, "applied": [...], "errors": [...],
            "gitea_username", "gitea_user_id", "renamed": bool}.
    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.", "errors": [], "applied": [], "renamed": False}
    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).",
            "errors": [],
            "applied": [],
            "renamed": False,
        }

    user_id = int(user["id"])
    try:
        current_username = desired_gitea_username(user)
        gitea_user = _ensure_gitea_user(user, current_username)
    except Exception as exc:
        message = f"Gitea účet se nepodařilo načíst: {str(exc)[:400]}"
        logger.error("apply_gitea_credentials: %s (user id=%s)", message, user_id)
        return {
            "status": "error",
            "error": message,
            "errors": [],
            "applied": [],
            "renamed": False,
        }

    current_login = _gitea_login(gitea_user) or current_username
    final_username = current_login
    applied: list[str] = []
    errors: list[str] = []

    # 1) Heslo – nezávisle na přejmenování (vždy se zkusí, i kdyby rename neexistoval).
    if new_password:
        try:
            _set_gitea_password(current_login, user, new_password, gitea_user=gitea_user)
            applied.append("heslo")
            logger.info("Gitea heslo nastaveno pro %s (user id=%s)", current_login, user_id)
        except Exception as exc:
            detail = str(exc)[:300]
            errors.append(f"heslo: {detail}")
            logger.error("Gitea nastavení hesla selhalo pro %s (user id=%s): %s", current_login, user_id, detail)

    # 2) Přejmenování (Gitea >= 1.21). Na starší Gitea vrátí HTTP 404 → zobrazí se srozumitelně.
    if new_username and new_username != current_login:
        try:
            _rename_gitea_user(current_login, new_username)
            final_username = new_username
            applied.append("jméno")
            logger.info("Gitea účet přejmenován %s -> %s (user id=%s)", current_login, new_username, user_id)
        except Exception as exc:
            detail = str(exc)[:300]
            errors.append(f"jméno: {detail}")
            logger.error("Gitea přejmenování %s -> %s selhalo (user id=%s): %s", current_login, new_username, user_id, detail)

    if applied and not errors:
        status = "synced"
    elif applied:
        status = "partial"
    else:
        status = "error"

    return {
        "status": status,
        "applied": applied,
        "errors": errors,
        "gitea_username": final_username,
        "gitea_user_id": int(gitea_user["id"]),
        "renamed": "jméno" in applied,
    }


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]
        logger.error("Gitea sync selhal pro uživatele id=%s (%s): %s", user_id, username, error)
        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)