appfactory-portal/app/routes/auth.py

import html
import json
import secrets
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request as UrlRequest
from urllib.request import urlopen

from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse

from app.auth import authenticate_user, current_user, find_or_create_oauth_user
from app.config import read_env_value
from app.db.audit import log_audit_event
from app.templates.layout import page

router = APIRouter()
DEFAULT_GITEA_URL = "http://192.168.66.130:3000"


@router.get("/login", response_class=HTMLResponse)
def login_form(request: Request):
    if current_user(request):
        return RedirectResponse(url="/portal/operations", status_code=303)

    return _render_login()


@router.post("/login", response_class=HTMLResponse)
def login(request: Request, username: str = Form(...), password: str = Form(...)):
    user = authenticate_user(username, password)

    if not user:
        return _render_login("Neplatné přihlašovací údaje.")

    request.session.clear()
    request.session["user_id"] = user["id"]
    log_audit_event(
        user,
        action="login_success",
        target_type="user",
        target_id=user.get("id"),
        metadata={"username": user.get("username")},
    )

    return RedirectResponse(url="/portal/operations", status_code=303)


@router.get("/auth/gitea/login")
def gitea_login(request: Request):
    if current_user(request):
        return RedirectResponse(url="/portal/operations", status_code=303)

    gitea_url = read_env_value("GITEA_URL", DEFAULT_GITEA_URL).rstrip("/")
    client_id = read_env_value("GITEA_OAUTH_CLIENT_ID", "")
    redirect_uri = read_env_value("GITEA_OAUTH_REDIRECT_URI", "")
    if not client_id or not redirect_uri:
        _log_gitea_failure("missing_oauth_config")
        return _render_login("Přihlášení přes Gitea není správně nastavené.")

    state = secrets.token_urlsafe(32)
    request.session["gitea_oauth_state"] = state
    log_audit_event(
        None,
        action="auth.gitea.login.started",
        target_type="auth",
        metadata={"provider": "gitea"},
    )

    params = urlencode(
        {
            "client_id": client_id,
            "redirect_uri": redirect_uri,
            "response_type": "code",
            "scope": "read:user",
            "state": state,
        }
    )
    return RedirectResponse(url=f"{gitea_url}/login/oauth/authorize?{params}", status_code=303)


@router.get("/auth/gitea/callback", response_class=HTMLResponse)
def gitea_callback(request: Request, code: str = "", state: str = "", error: str = ""):
    expected_state = request.session.pop("gitea_oauth_state", None)
    if error:
        _log_gitea_failure("provider_error", error=error)
        return _render_login("Přihlášení přes Gitea se nepodařilo.")
    if not expected_state or not state or not secrets.compare_digest(expected_state, state):
        _log_gitea_failure("invalid_state")
        return _render_login("Přihlášení přes Gitea se nepodařilo.")
    if not code:
        _log_gitea_failure("missing_code")
        return _render_login("Přihlášení přes Gitea se nepodařilo.")

    try:
        token = _exchange_gitea_code(code)
        gitea_user = _fetch_gitea_user(token)
        username = (gitea_user.get("login") or gitea_user.get("username") or "").strip()
        if not username:
            _log_gitea_failure("missing_username")
            return _render_login("Přihlášení přes Gitea se nepodařilo.")

        display_name = (gitea_user.get("full_name") or username).strip()
        email = (gitea_user.get("email") or "").strip()
        user = find_or_create_oauth_user(username, display_name, email)
        if not user.get("is_active"):
            _log_gitea_failure("disabled_user", username=username)
            return _render_login("Uživatel je v portálu vypnutý.")

        request.session.clear()
        request.session["user_id"] = user["id"]
        log_audit_event(
            user,
            action="auth.gitea.login.success",
            target_type="user",
            target_id=user.get("id"),
            metadata={"username": user.get("username"), "provider": "gitea"},
        )
        return RedirectResponse(url="/portal", status_code=303)
    except Exception:
        _log_gitea_failure("callback_failed")
        return _render_login("Přihlášení přes Gitea se nepodařilo.")


@router.post("/logout")
def logout(request: Request):
    user = current_user(request)
    log_audit_event(
        user,
        action="logout",
        target_type="user",
        target_id=user.get("id") if user else None,
        metadata={"username": user.get("username")} if user else {},
    )
    request.session.clear()
    return RedirectResponse(url="/portal/login", status_code=303)


def _exchange_gitea_code(code: str) -> str:
    gitea_url = read_env_value("GITEA_URL", DEFAULT_GITEA_URL).rstrip("/")
    client_id = read_env_value("GITEA_OAUTH_CLIENT_ID", "")
    client_secret = read_env_value("GITEA_OAUTH_CLIENT_SECRET", "")
    redirect_uri = read_env_value("GITEA_OAUTH_REDIRECT_URI", "")
    if not client_id or not client_secret or not redirect_uri:
        raise RuntimeError("Missing Gitea OAuth configuration")

    payload = urlencode(
        {
            "client_id": client_id,
            "client_secret": client_secret,
            "code": code,
            "grant_type": "authorization_code",
            "redirect_uri": redirect_uri,
        }
    ).encode("utf-8")
    request = UrlRequest(
        f"{gitea_url}/login/oauth/access_token",
        data=payload,
        headers={
            "Accept": "application/json",
            "Content-Type": "application/x-www-form-urlencoded",
        },
        method="POST",
    )
    data = _read_json(request)
    token = data.get("access_token")
    if not token:
        raise RuntimeError("Gitea OAuth token response did not contain access_token")
    return token


def _fetch_gitea_user(access_token: str) -> dict:
    gitea_url = read_env_value("GITEA_URL", DEFAULT_GITEA_URL).rstrip("/")
    request = UrlRequest(
        f"{gitea_url}/api/v1/user",
        headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {access_token}",
        },
        method="GET",
    )
    return _read_json(request)


def _read_json(request: UrlRequest) -> dict:
    try:
        with urlopen(request, timeout=10) as response:
            return json.loads(response.read().decode("utf-8"))
    except (HTTPError, URLError, TimeoutError, json.JSONDecodeError) as exc:
        raise RuntimeError("Gitea OAuth request failed") from exc


def _log_gitea_failure(reason: str, **metadata):
    data = {"reason": reason, "provider": "gitea"}
    data.update(metadata)
    log_audit_event(
        None,
        action="auth.gitea.login.failed",
        target_type="auth",
        metadata=data,
    )


def _render_login(error: str | None = None) -> str:
    error_html = ""
    if error:
        error_html = f'<p class="alert alert-danger">{html.escape(error)}</p>'

    return page(
        "Přihlášení",
        f"""
        <div class="auth-card card">
            <h2>CSBot Services Portal</h2>
            <p class="muted">Doporučené přihlášení je přes Gitea. Lokální účet slouží pouze jako nouzový administrátorský přístup.</p>
            {error_html}

            <p>
                <a class="btn" href="/portal/auth/gitea/login">Přihlásit přes Gitea</a>
            </p>

            <form method="post" action="/portal/login">
                <h3>Nouzové lokální přihlášení</h3>
                <p>
                    <label>Uživatelské jméno</label><br>
                    <input name="username" autocomplete="username" required autofocus>
                </p>

                <p>
                    <label>Heslo</label><br>
                    <input type="password" name="password" autocomplete="current-password" required>
                </p>

                <button type="submit" class="btn-secondary">Přihlásit lokálně</button>
            </form>
        </div>
        """,
    )