Přidáno:

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
This commit is contained in:
JiriUhlir
2026-06-01 13:53:06 +02:00
parent 6aa5e69ee4
commit 786d055f6b
2 changed files with 216 additions and 3 deletions
+59
View File
@@ -35,6 +35,65 @@ def get_user_by_username(username: str) -> dict[str, Any] | None:
return dict(row) if row else 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: def get_user_by_id(user_id: int) -> dict[str, Any] | None:
con = get_connection() con = get_connection()
+157 -3
View File
@@ -1,13 +1,21 @@
import html 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 import APIRouter, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse 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.db.audit import log_audit_event
from app.templates.layout import page from app.templates.layout import page
router = APIRouter() router = APIRouter()
DEFAULT_GITEA_URL = "http://192.168.66.130:3000"
@router.get("/login", response_class=HTMLResponse) @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) 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") @router.post("/logout")
def logout(request: Request): def logout(request: Request):
user = current_user(request) user = current_user(request)
@@ -52,6 +136,71 @@ def logout(request: Request):
return RedirectResponse(url="/portal/login", status_code=303) 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: def _render_login(error: str | None = None) -> str:
error_html = "" error_html = ""
if error: if error:
@@ -62,10 +211,15 @@ def _render_login(error: str | None = None) -> str:
f""" f"""
<div class="auth-card card"> <div class="auth-card card">
<h2>CSBot Services Portal</h2> <h2>CSBot Services Portal</h2>
<p class="muted">Přihlaste se interm účtem.</p> <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} {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"> <form method="post" action="/portal/login">
<h3>Nouzové lokální přihlášení</h3>
<p> <p>
<label>Uživatelské jméno</label><br> <label>Uživatelské jméno</label><br>
<input name="username" autocomplete="username" required autofocus> <input name="username" autocomplete="username" required autofocus>
@@ -76,7 +230,7 @@ def _render_login(error: str | None = None) -> str:
<input type="password" name="password" autocomplete="current-password" required> <input type="password" name="password" autocomplete="current-password" required>
</p> </p>
<button type="submit">Přihlásit</button> <button type="submit" class="btn-secondary">Přihlásit lokálně</button>
</form> </form>
</div> </div>
""", """,