From 786d055f6bd4466263c44fe503887d9acbfb83ea Mon Sep 17 00:00:00 2001 From: JiriUhlir <149317995+JiriUhlir@users.noreply.github.com> Date: Mon, 1 Jun 2026 13:53:06 +0200 Subject: [PATCH] =?UTF-8?q?P=C5=99id=C3=A1no:?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tlačítko Přihlásit přes Gitea na login stránku zachovaný lokální login jako nouzový admin fallback GET /portal/auth/gitea/login GET /portal/auth/gitea/callback OAuth state přes session a ověření přes secrets.compare_digest výměna code za token a načtení /api/v1/user mapování Gitea uživatele do users bez ORM nové Gitea účty dostanou role = developer, existující role se zachová disabled user is_active = 0 se nepřihlásí audit eventy auth.gitea.login.started, auth.gitea.login.success, auth.gitea.login.failed bezpečné chybové hlášky bez detailů a bez zobrazení client secretu --- app/auth.py | 59 +++++++++++++++++ app/routes/auth.py | 160 ++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 216 insertions(+), 3 deletions(-) diff --git a/app/auth.py b/app/auth.py index a8f88fe..e9fe271 100644 --- a/app/auth.py +++ b/app/auth.py @@ -35,6 +35,65 @@ def get_user_by_username(username: str) -> dict[str, Any] | None: return dict(row) if row else None +def find_or_create_oauth_user(username: str, display_name: str, email: str) -> dict[str, Any]: + username = username.strip() + display_name = (display_name or username).strip() or username + email = (email or "").strip() + + con = get_connection() + row = con.execute( + """ + SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at + FROM users + WHERE username = ? + """, + (username,), + ).fetchone() + + if row: + con.execute( + """ + UPDATE users + SET display_name = ?, + email = ?, + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + """, + (display_name, email, row["id"]), + ) + else: + con.execute( + """ + INSERT INTO users ( + username, + display_name, + email, + password_hash, + role, + is_active, + created_at, + updated_at + ) + VALUES (?, ?, ?, '', 'developer', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + """, + (username, display_name, email), + ) + + con.commit() + user_id = row["id"] if row else con.execute("SELECT last_insert_rowid() AS id").fetchone()["id"] + user = con.execute( + """ + SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at + FROM users + WHERE id = ? + """, + (user_id,), + ).fetchone() + + con.close() + return dict(user) + + def get_user_by_id(user_id: int) -> dict[str, Any] | None: con = get_connection() diff --git a/app/routes/auth.py b/app/routes/auth.py index abca15b..7ee612e 100644 --- a/app/routes/auth.py +++ b/app/routes/auth.py @@ -1,13 +1,21 @@ 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 +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) @@ -38,6 +46,82 @@ def login(request: Request, username: str = Form(...), password: str = Form(...) 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) @@ -52,6 +136,71 @@ def logout(request: Request): 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: @@ -62,10 +211,15 @@ def _render_login(error: str | None = None) -> str: f"""

CSBot Services Portal

-

Přihlaste se interním účtem.

+

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

{error_html} +

+ Přihlásit přes Gitea +

+
+

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


@@ -76,7 +230,7 @@ def _render_login(error: str | None = None) -> str:

- +
""",