"""FlowDeck — Auth routes: login, callback, logout.""" from __future__ import annotations import logging import secrets from fastapi import APIRouter, Request, Query from fastapi.responses import RedirectResponse, HTMLResponse from app.auth.session import SessionManager from app.auth.oauth import gitea_oauth from app.config import settings logger = logging.getLogger(__name__) router = APIRouter(tags=["auth"], prefix="/auth") LOCAL_LOGIN_HTML = """ FlowDeck — Login """ @router.get("/login") async 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) if provider == "local": from fastapi.responses import HTMLResponse return HTMLResponse(LOCAL_LOGIN_HTML, status_code=200) # OAuth flow if not gitea_oauth.enabled: # Fallback: use global token, create a fake session from app.db import get_conn with get_conn() as conn: user = conn.execute("SELECT * FROM users WHERE login='admin'").fetchone() if not user: conn.execute("INSERT INTO users (login, full_name, email, avatar_url) VALUES ('admin', 'Admin', '', '')") conn.commit() user = conn.execute("SELECT * FROM users WHERE login='admin'").fetchone() user_data = dict(user) session = SessionManager.create_session(user_data) response = RedirectResponse(url="/", status_code=302) response.set_cookie("flowdeck_session", session, httponly=True, max_age=86400 * 7, samesite="lax") return response state = secrets.token_hex(32) request.session["oauth_state"] = state auth_url = gitea_oauth.get_authorize_url(state) return RedirectResponse(url=auth_url, status_code=302) @router.post("/register") async def register(request: Request): """Register a new local account.""" from app.db import get_conn from app.password_utils import hash_password import json try: body = await request.json() except Exception: body = {} 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) 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) conn.execute( "INSERT INTO users (login, full_name, email, password_hash) VALUES (?, ?, ?, ?)", (email, name, email, hash_password(password)), ) conn.commit() user = conn.execute("SELECT * FROM users WHERE login=?", (email,)).fetchone() user_data = dict(user) session = SessionManager.create_session(user_data) 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") return response @router.post("/local-login") async def local_login(request: Request): """Login with email + password.""" from app.db import get_conn from app.password_utils import verify_password, is_locked from fastapi.responses import JSONResponse import json, time try: body = await request.json() except Exception: body = {} 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 login 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() session = SessionManager.create_session(ud) 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") return response @router.get("/callback") async def callback( request: Request, code: str = Query(...), state: str = Query(...), ): """Handle OAuth2 callback from Gitea.""" # Validate state expected_state = request.session.get("oauth_state", "") if not expected_state or state != expected_state: return HTMLResponse("

Invalid state

", status_code=400) # Exchange code for token token_data = await gitea_oauth.exchange_code(code) 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 user = await gitea_oauth.get_user(access_token) if not user: return HTMLResponse("

Failed to get user

", status_code=400) # Store user in DB from app.db import get_conn with get_conn() as conn: conn.execute( """INSERT INTO users (login, full_name, email, avatar_url) VALUES (?, ?, ?, ?) ON CONFLICT(login) DO UPDATE SET full_name=excluded.full_name, email=excluded.email, avatar_url=excluded.avatar_url""", (user["login"], user.get("full_name", ""), user.get("email", ""), user.get("avatar_url", "")), ) conn.commit() # Store token SessionManager.store_token(user["id"], access_token) # Also store user in local DB with get_conn() as conn: db_user = conn.execute("SELECT * FROM users WHERE login=?", (user["login"],)).fetchone() user_data = dict(db_user) if db_user else user # Create session session = SessionManager.create_session(user_data) response = RedirectResponse(url="/", status_code=302) response.set_cookie("flowdeck_session", session, httponly=True, max_age=86400 * 7, samesite="lax") return response @router.get("/logout") async def logout(): """Clear session and redirect to dashboard.""" response = RedirectResponse(url="/", status_code=302) response.delete_cookie("flowdeck_session") return response @router.get("/user") async def current_user(request: Request): """Return current user info as JSON.""" from app.auth.session import get_current_user as gcu user = await gcu(request) if not user: return {"authenticated": False} return {"authenticated": True, "user": user}