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_google_user, find_or_create_oauth_user, mark_last_login from app.config import ( get_auth_mode, get_gitea_public_url, get_gitea_redirect_uri, get_gitea_server_url, get_google_redirect_uri, is_gitea_oauth_button_enabled, is_google_oauth_button_enabled, read_env_value, ) from app.db.audit import log_audit_event from app.templates.layout import page router = APIRouter() GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth" GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token" GOOGLE_USERINFO_URL = "https://openidconnect.googleapis.com/v1/userinfo" @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(...)): if get_auth_mode() == "google": return _render_login("Local login is disabled. Use Google sign in.") 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) if get_auth_mode() != "mixed": return _render_login("Gitea login is not enabled in this auth mode.") gitea_url = get_gitea_public_url() client_id = read_env_value("GITEA_OAUTH_CLIENT_ID", "") client_secret = read_env_value("GITEA_OAUTH_CLIENT_SECRET", "") redirect_uri = get_gitea_redirect_uri() if not gitea_url or not client_id or not client_secret or not redirect_uri: _log_gitea_failure("missing_oauth_config") return _render_login("Gitea login is not fully configured.") 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 = ""): if get_auth_mode() != "mixed": return _render_login("Gitea login is not enabled in this auth mode.") 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_enabled", 1): _log_gitea_failure("disabled_user", username=username) return _render_login("User is disabled in the portal.") 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"] mark_last_login(int(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.get("/auth/google/login") def google_login(request: Request): if current_user(request): return RedirectResponse(url="/portal/operations", status_code=303) if get_auth_mode() == "local": return _render_login("Google login is not enabled in local auth mode.") if not is_google_oauth_button_enabled(): _log_google_failure("missing_oauth_config") return _render_login("Google login is not configured.") redirect_uri = get_google_redirect_uri() if not redirect_uri: _log_google_failure("missing_redirect_uri") return _render_login("Google login is missing redirect URI.") state = secrets.token_urlsafe(32) request.session["google_oauth_state"] = state log_audit_event( None, action="auth.google.login.started", target_type="auth", metadata={"provider": "google"}, ) params = urlencode( { "client_id": read_env_value("GOOGLE_CLIENT_ID", ""), "redirect_uri": redirect_uri, "response_type": "code", "scope": "openid email profile", "state": state, "access_type": "online", "prompt": "select_account", } ) return RedirectResponse(url=f"{GOOGLE_AUTH_URL}?{params}", status_code=303) @router.get("/auth/google/callback", response_class=HTMLResponse) def google_callback(request: Request, code: str = "", state: str = "", error: str = ""): if get_auth_mode() == "local": return _render_login("Google login is not enabled in local auth mode.") expected_state = request.session.pop("google_oauth_state", None) if error: _log_google_failure("provider_error", error=error) return _render_login("Google login failed.") if not expected_state or not state or not secrets.compare_digest(expected_state, state): _log_google_failure("invalid_state") return _render_login("Google login failed.") if not code: _log_google_failure("missing_code") return _render_login("Google login failed.") try: token = _exchange_google_code(code) google_user = _fetch_google_user(token) email = (google_user.get("email") or "").strip().lower() if not email: _log_google_failure("missing_email") return _render_login("Google account did not return an email address.") allowed_domain = read_env_value("GOOGLE_ALLOWED_DOMAIN", "").strip().lower() email_domain = email.rsplit("@", 1)[1] if "@" in email else "" hosted_domain = (google_user.get("hd") or "").strip().lower() if allowed_domain and allowed_domain not in {email_domain, hosted_domain}: _log_google_failure("domain_not_allowed", email_domain=email_domain) return _render_login("Google account is not allowed for this portal.") username = email display_name = (google_user.get("name") or email).strip() provider_subject = (google_user.get("sub") or "").strip() avatar_url = (google_user.get("picture") or "").strip() user, created = find_or_create_google_user(email, display_name, provider_subject, avatar_url) if not user.get("is_enabled", 1): _log_google_failure("disabled_user", username=username) return _render_login("User is disabled in the portal.") if not user.get("is_active"): _log_google_failure("disabled_user", username=username) return _render_login("UĹľivatel je v portálu vypnutĂ˝.") request.session.clear() request.session["user_id"] = user["id"] mark_last_login(int(user["id"])) if created: log_audit_event( user, action="auth.google.user.created", target_type="user", target_id=user.get("id"), metadata={ "username": user.get("username"), "email": user.get("email"), "role": user.get("role"), "provider": "google", }, ) log_audit_event( user, action="auth.google.login.success", target_type="user", target_id=user.get("id"), metadata={"username": user.get("username"), "provider": "google"}, ) return RedirectResponse(url="/portal", status_code=303) except Exception: _log_google_failure("callback_failed") return _render_login("Google login failed.") @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 = get_gitea_server_url() client_id = read_env_value("GITEA_OAUTH_CLIENT_ID", "") client_secret = read_env_value("GITEA_OAUTH_CLIENT_SECRET", "") redirect_uri = get_gitea_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 = get_gitea_server_url() request = UrlRequest( f"{gitea_url}/api/v1/user", headers={ "Accept": "application/json", "Authorization": f"Bearer {access_token}", }, method="GET", ) return _read_json(request) def _exchange_google_code(code: str) -> str: client_id = read_env_value("GOOGLE_CLIENT_ID", "") client_secret = read_env_value("GOOGLE_CLIENT_SECRET", "") redirect_uri = get_google_redirect_uri() if not client_id or not client_secret or not redirect_uri: raise RuntimeError("Missing Google 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( GOOGLE_TOKEN_URL, 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("Google OAuth token response did not contain access_token") return token def _fetch_google_user(access_token: str) -> dict: request = UrlRequest( GOOGLE_USERINFO_URL, 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 _log_google_failure(reason: str, **metadata): data = {"reason": reason, "provider": "google"} data.update(metadata) log_audit_event( None, action="auth.google.login.failed", target_type="auth", metadata=data, ) def _render_login(error: str | None = None) -> str: error_html = "" if error: error_html = f'

{html.escape(error)}

' auth_mode = get_auth_mode() gitea_login_html = "" if auth_mode == "mixed" and is_gitea_oauth_button_enabled(): gitea_login_html = """

Sign in with Gitea

""" google_login_html = "" if auth_mode in {"google", "mixed"} and is_google_oauth_button_enabled(): google_login_html = """

Sign in with Google

""" elif auth_mode in {"google", "mixed"}: google_login_html = '

Google login is not fully configured.

' local_login_html = "" if auth_mode in {"mixed", "local"}: local_login_html = """

Local login



""" return page( "Sign in", f"""

CSBot Services Portal

{error_html} {google_login_html} {gitea_login_html} {local_login_html}
""", ) return page( "Přihlášení", f"""

CSBot Services Portal

Doporučené přihlášení je přes Gitea. Lokální účet slouží pouze jako nouzový administrátorský přístup.

{error_html} {gitea_login_html} {google_login_html}

Nouzové lokální přihlášení



""", )