JiriUhlir
7dffff17c7
DEFAULT_APPFACTORY_HOST = "192.168.66.130" je pryč z config.py (line 10). DEFAULT_GITEA_URL = "http://192.168.66.130:3000" je pryč z auth.py (line 24). Doplněné chování: APPFACTORY_HOST se odvozuje přes helper: env APPFACTORY_HOST, potom APPFACTORY_PORTAL_PUBLIC_URL, potom request host. Gitea browser redirect používá jen explicitní veřejnou URL z GITEA_URL nebo GITEA_ROOT_URL. Server-side Gitea token/userinfo requesty používají GITEA_URL, potom GITEA_ROOT_URL, jinak interní Docker URL http://appfactory-gitea:3000. Gitea login tlačítko se zobrazí jen při kompletní konfiguraci: public URL, client id, client secret a redirect URI. Přímý Gitea login bez kompletní konfigurace vrací čitelnou chybu.
404 lines
14 KiB
Python
404 lines
14 KiB
Python
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 (
|
|
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(...)):
|
|
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 = 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 = ""):
|
|
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.get("/auth/google/login")
|
|
def google_login(request: Request):
|
|
if current_user(request):
|
|
return RedirectResponse(url="/portal/operations", status_code=303)
|
|
|
|
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 = ""):
|
|
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()
|
|
user = find_or_create_oauth_user(username, display_name, email)
|
|
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"]
|
|
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'<p class="alert alert-danger">{html.escape(error)}</p>'
|
|
gitea_login_html = ""
|
|
if is_gitea_oauth_button_enabled():
|
|
gitea_login_html = """
|
|
<p>
|
|
<a class="btn" href="/portal/auth/gitea/login">Sign in with Gitea</a>
|
|
</p>
|
|
"""
|
|
google_login_html = ""
|
|
if is_google_oauth_button_enabled():
|
|
google_login_html = """
|
|
<p>
|
|
<a class="btn" href="/portal/auth/google/login">Sign in with Google</a>
|
|
</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}
|
|
{gitea_login_html}
|
|
|
|
{google_login_html}
|
|
|
|
<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>
|
|
""",
|
|
)
|