"""FlowDeck — Auth routes: login, callback, logout.""" from __future__ import annotations import logging import secrets from fastapi import APIRouter, Body, Query, Request from fastapi.responses import HTMLResponse, RedirectResponse from app.auth.session import SessionManager from app.config import settings from app.templating import CSP_NONCE logger = logging.getLogger(__name__) router = APIRouter(tags=["auth"], prefix="/auth") def get_redirect_uri(request: Request) -> str: """OAuth redirect URI for this request. Explicit `OAUTH_REDIRECT_URI` env override wins (must be registered in the provider's OAuth app). Otherwise it is derived from the request so it always matches the URL the user actually used: scheme from `X-Forwarded-Proto` (reverse proxies) falling back to the request scheme, host from `X-Forwarded-Host` falling back to the `Host` header. """ if settings.oauth_redirect_uri: return settings.oauth_redirect_uri proto = request.headers.get("x-forwarded-proto", "") scheme = proto.split(",")[0].strip() or request.url.scheme or "http" fwd_host = request.headers.get("x-forwarded-host", "") host = fwd_host.split(",")[0].strip() or request.headers.get("host", "localhost:8080") return f"{scheme}://{host}/auth/callback" def _with_nonce(html: str) -> str: """A20 : injecte le nonce CSP au moment du rendu. `LOCAL_LOGIN_HTML` est une constante de module — le nonce, lui, est par requête, donc il ne peut être figé qu'ici. """ return html.replace(" """ @router.get("/register") def register_page(request: Request): """Show the registration page (local login page with register tab active).""" return HTMLResponse(_with_nonce(LOCAL_LOGIN_HTML.replace( 'class="tab active" onclick="switchTab(\'login\')"', 'class="tab" onclick="switchTab(\'login\')"' ).replace( 'class="tab" onclick="switchTab(\'register\')"', 'class="tab active" onclick="switchTab(\'register\')"' ).replace( 'let mode=\'login\';', 'let mode=\'register\';' ).replace( 'id="name-group" style="display:none"', 'id="name-group" style="display:block"' ).replace( 'id="submit-btn">Login<', 'id="submit-btn">Register<' )), status_code=200) @router.get("/login") def login(request: Request, provider: str = Query("gitea")): """Redirect to OAuth2 authorize page or show local login page.""" # Local login page (POST handled by /auth/local-login) from fastapi.responses import HTMLResponse if provider == "local": return HTMLResponse(_with_nonce(LOCAL_LOGIN_HTML), status_code=200) # OAuth flow — check if provider is configured from app.auth.providers import get_provider oauth_provider = get_provider(provider) if not oauth_provider: # OAuth provider not configured — show error instead of auto-creating admin return HTMLResponse( f"""FlowDeck

⚠️ {provider.title()} OAuth not configured

The {provider} integration has not been set up by the server administrator.

↩ Use local login

""", status_code=200, ) state = secrets.token_hex(32) request.session["oauth_state"] = state request.session["oauth_provider"] = provider # Link mode: connect OAuth to current local account instead of creating new user mode = request.query_params.get("mode", "") # Encode auth mode in state to survive session loss during OAuth redirect signed_state = f"{state}:{mode}" if mode else state request.session["oauth_mode"] = mode # Redirect URI derived from the incoming request (scheme-aware); stored in # session so the callback reuses the EXACT same URI for token exchange redirect_uri = get_redirect_uri(request) request.session["oauth_redirect_uri"] = redirect_uri auth_url = oauth_provider.get_authorize_url(signed_state, redirect_uri=redirect_uri, force_login=(mode == "link")) return RedirectResponse(url=auth_url, status_code=302) @router.post("/register") def register(request: Request, body: dict = Body(default={})): """Register a new local account.""" from app.db import get_conn from app.password_utils import hash_password email = body.get("email", "").strip() password = body.get("password", "").strip() name = body.get("name", email.split("@")[0] if "@" in email else email) if not email or not password: from fastapi.responses import JSONResponse return JSONResponse({"error": "Email and password required"}, status_code=400) if len(password) < 6: from fastapi.responses import JSONResponse return JSONResponse({"error": "Password must be at least 6 characters"}, status_code=400) # SSO-only instance (v6.7.0): local registration is refused — accounts are # auto-provisioned by the IdP instead (admins still come from Settings). from app.services.sso_provisioning import is_sso_only if is_sso_only(): from fastapi.responses import JSONResponse return JSONResponse( {"error": "Registration is disabled — sign in with your organization SSO"}, status_code=403, ) with get_conn() as conn: existing = conn.execute("SELECT id FROM users WHERE login=?", (email,)).fetchone() if existing: from fastapi.responses import JSONResponse return JSONResponse({"error": "Account already exists"}, status_code=409) # First real user (excluding default admin with no password) is admin real_user_count = conn.execute( "SELECT COUNT(*) FROM users WHERE password_hash IS NOT NULL AND password_hash != ''" ).fetchone()[0] is_admin = 1 if real_user_count == 0 else 0 conn.execute( "INSERT INTO users (login, full_name, email, password_hash, is_admin) VALUES (?, ?, ?, ?, ?)", (email, name, email, hash_password(password), is_admin), ) conn.commit() user = conn.execute("SELECT * FROM users WHERE login=?", (email,)).fetchone() user_data = dict(user) # Log login _log_login(user_data["id"], request) session = SessionManager.create_session(user_data, request) from fastapi.responses import JSONResponse response = JSONResponse({"status": "ok", "user": {"login": email, "name": name}}) response.set_cookie("flowdeck_session", session, httponly=True, max_age=86400 * 7, samesite="lax", path="/") return response @router.post("/local-login") def local_login(request: Request, body: dict = Body(default={})): """Login with email + password.""" import time from fastapi.responses import JSONResponse from app.db import get_conn from app.password_utils import is_locked, verify_password email = body.get("email", "").strip() password = body.get("password", "").strip() if not email or not password: return JSONResponse({"error": "Email and password required"}, status_code=400) with get_conn() as conn: user = conn.execute("SELECT * FROM users WHERE login=?", (email,)).fetchone() if not user: return JSONResponse({"error": "Invalid credentials"}, status_code=401) ud = dict(user) if not ud.get("is_active"): return JSONResponse({"error": "Account disabled"}, status_code=403) if is_locked(ud.get("locked_until")): return JSONResponse({"error": "Account temporarily locked. Try again later."}, status_code=423) if not verify_password(password, ud.get("password_hash", "")): with get_conn() as conn: attempts = (ud.get("login_attempts", 0) or 0) + 1 lock = None if attempts >= 5: lock = str(time.time() + 900) # 15 min lock conn.execute( "UPDATE users SET login_attempts=?, locked_until=? WHERE id=?", (attempts, lock, ud["id"]), ) conn.commit() return JSONResponse({"error": "Invalid credentials"}, status_code=401) # Successful local login — SSO-only instances keep a way in for admins # only (every other account must use the IdP, design §7.1). from app.services.sso_provisioning import is_sso_only if is_sso_only() and not ud.get("is_admin"): return JSONResponse( {"error": "Local login is disabled on this instance — sign in with SSO"}, status_code=403, ) # v7.2.0: verified domain with SSO enforcement (admins keep local access). if not ud.get("is_admin"): with get_conn() as conn: dom = (ud.get("email") or "").split("@")[-1].lower() if "@" in (ud.get("email") or "") else "" if dom: enforced = conn.execute( "SELECT id FROM domain_claims WHERE domain=? AND verified=1" " AND enforce_sso=1", (dom,)).fetchone() if enforced: return JSONResponse( {"error": "Local login is disabled for your domain — sign in with SSO"}, status_code=403) with get_conn() as conn: conn.execute( "UPDATE users SET login_attempts=0, locked_until=NULL, last_login=? WHERE id=?", (str(time.time()), ud["id"]), ) conn.commit() # v7.2.0: TOTP 2FA — password OK, but hold the session until code check. from app.services import two_factor as _2fa if _2fa.is_enabled(ud["id"]): return JSONResponse({"status": "2fa_required", "pending": _2fa.mint_pending(ud["id"])}) session = SessionManager.create_session(ud, request) _log_login(ud["id"], request) response = JSONResponse({"status": "ok", "user": {"login": ud["login"], "name": ud["full_name"]}}) response.set_cookie("flowdeck_session", session, httponly=True, max_age=86400 * 7, samesite="lax", path="/") return response @router.get("/callback") async def callback( request: Request, code: str = Query(...), state: str = Query(...), ): """Handle OAuth2 callback from Gitea.""" # Recover mode from state suffix (state:mode format), then validate expected_state = request.session.get("oauth_state", "") provider_name = request.session.get("oauth_provider", "gitea") auth_mode = "" raw_state = state if ":" in state: raw_state, auth_mode = state.rsplit(":", 1) if auth_mode: request.session["oauth_mode"] = auth_mode # Validate: state token must match session, unless session lost and mode present if expected_state and raw_state != expected_state: return HTMLResponse("

Invalid state

", status_code=400) if not expected_state and not auth_mode: return HTMLResponse("

Session expired — please try connecting again

", status_code=400) from app.auth.providers import get_provider oauth_provider = get_provider(provider_name) if not oauth_provider: return HTMLResponse(f"

Unknown provider: {provider_name}

", status_code=400) # Exchange code for token — reuse the redirect URI from the authorize step # (stored in session), falling back to deriving it from this request redirect_uri = request.session.get("oauth_redirect_uri") or get_redirect_uri(request) token_data = await oauth_provider.exchange_code(code, redirect_uri=redirect_uri) if not token_data: return HTMLResponse("

Token exchange failed

", status_code=400) access_token = token_data.get("access_token") if not access_token: return HTMLResponse("

No access token

", status_code=400) # Get user info oauth_user = await oauth_provider.get_user(access_token) if not oauth_user: return HTMLResponse("

Failed to get user

", status_code=400) # Link mode: connect OAuth to current session user (for Settings → Integrations) oauth_mode = request.session.pop("oauth_mode", "") if oauth_mode == "link": from app.auth.session import get_current_user as gcu current = gcu(request) if not current: return HTMLResponse("

Not logged in — please log in first

", status_code=400) from app.db import get_conn as _gc with _gc() as conn: conn.execute( """INSERT OR REPLACE INTO user_oauth_tokens (user_id, provider, access_token, refresh_token, expires_at, updated_at) VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)""", (current["id"], provider_name, access_token, token_data.get("refresh_token"), token_data.get("expires_at")), ) conn.commit() return RedirectResponse(url="/settings#integrations", status_code=302) # Store user in DB from app.db import get_conn login_id = f"{provider_name}_{oauth_user['login']}" with get_conn() as conn: conn.execute( """INSERT INTO users (login, full_name, email, avatar_url, auth_method) VALUES (?, ?, ?, ?, ?) ON CONFLICT(login) DO UPDATE SET full_name=excluded.full_name, email=excluded.email, avatar_url=excluded.avatar_url, auth_method=excluded.auth_method""", (login_id, oauth_user.get("full_name", ""), oauth_user.get("email", ""), oauth_user.get("avatar_url", ""), provider_name), ) conn.commit() # Store OAuth token uid = conn.execute("SELECT id FROM users WHERE login=?", (login_id,)).fetchone() if uid: conn.execute( """INSERT OR REPLACE INTO user_oauth_tokens (user_id, provider, access_token, refresh_token, expires_at, updated_at) VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)""", (uid["id"], provider_name, access_token, token_data.get("refresh_token"), token_data.get("expires_at")), ) conn.commit() user = conn.execute("SELECT * FROM users WHERE id=?", (uid["id"],)).fetchone() else: user = conn.execute("SELECT * FROM users WHERE login=?", (login_id,)).fetchone() user_data = dict(user) if user else oauth_user # Create session session = SessionManager.create_session(user_data, request) _log_login(user_data["id"], request) response = RedirectResponse(url="/workspaces", status_code=302) response.set_cookie("flowdeck_session", session, httponly=True, max_age=86400 * 7, samesite="lax", path="/") return response @router.get("/logout") def logout(request: Request): """Clear session and redirect to login page. SAML sessions additionally hand over to the IdP's Single Logout when one is configured (the actual cookie clearing happens on the SLO route). """ cookie = request.cookies.get("flowdeck_session", "") user = SessionManager.decode_session(cookie) if cookie else None local_target = "/auth/login?provider=local" if user and user.get("_sso_name_id"): # SSO session → let /auth/saml/logout revoke locally + notify the IdP. return RedirectResponse(url=f"/auth/saml/logout?next={local_target}", status_code=302) response = RedirectResponse(url=local_target, status_code=302) response.delete_cookie("flowdeck_session") return response @router.get("/user") def current_user(request: Request): """Return current user info as JSON.""" from app.auth.session import get_current_user as gcu user = gcu(request) if not user: return {"authenticated": False} return {"authenticated": True, "user": user} # ── v7.2.0 — TOTP 2FA ───────────────────────────────────────────────────── @router.post("/local-verify") def local_verify(request: Request, body: dict = Body(default={})): """Exchange a 2FA ``pending`` token + TOTP/backup code for a session.""" from fastapi.responses import JSONResponse from app.db import get_conn from app.services import two_factor as _2fa user_id = _2fa.redeem_pending(body.get("pending", "")) if not user_id: return JSONResponse({"error": "Challenge expired — log in again"}, status_code=401) if not _2fa.verify_code(user_id, body.get("code", "")): return JSONResponse({"error": "Invalid code"}, status_code=401) with get_conn() as conn: row = conn.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone() if not row or not row["is_active"]: return JSONResponse({"error": "Account disabled"}, status_code=403) ud = dict(row) session = SessionManager.create_session(ud, request) _log_login(ud["id"], request) response = JSONResponse({"status": "ok", "user": {"login": ud["login"]}}) response.set_cookie("flowdeck_session", session, httponly=True, max_age=86400 * 7, samesite="lax", path="/") return response def _session_user_or_401(request: Request) -> dict: from fastapi import HTTPException user = SessionManager.decode_session(request.cookies.get("flowdeck_session", "")) if not user or not user.get("id"): raise HTTPException(401, "Authentication required") return user @router.get("/2fa/status") def twofa_status(request: Request): from app.services import two_factor as _2fa user = _session_user_or_401(request) return {"enabled": _2fa.is_enabled(user["id"]), "backup_remaining": _2fa.remaining_backup_codes(user["id"])} @router.post("/2fa/setup") def twofa_setup(request: Request): from app.services import two_factor as _2fa user = _session_user_or_401(request) return _2fa.setup_secret(user["id"]) @router.post("/2fa/activate") def twofa_activate(request: Request, body: dict = Body(default={})): from fastapi.responses import JSONResponse from app.services import two_factor as _2fa user = _session_user_or_401(request) try: codes = _2fa.activate_secret(user["id"], body.get("secret", ""), body.get("code", "")) except ValueError: return JSONResponse({"error": "Invalid code — secret not activated"}, status_code=400) return {"status": "enabled", "backup_codes": codes} @router.post("/2fa/disable") def twofa_disable(request: Request): from app.services import two_factor as _2fa user = _session_user_or_401(request) _2fa.disable(user["id"]) return {"status": "disabled"} # ── Helpers ── def _log_login(user_id: int, request: Request): """Record login in history.""" try: from app.db import get_conn ip = request.client.host if request.client else '' ua = request.headers.get('user-agent', '')[:500] with get_conn() as conn: conn.execute( "INSERT INTO login_history (user_id, ip_address, user_agent) VALUES (?, ?, ?)", (user_id, ip, ua), ) conn.commit() except Exception: logger.exception("_log_login")