"""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.config import settings
logger = logging.getLogger(__name__)
router = APIRouter(tags=["auth"], prefix="/auth")
LOCAL_LOGIN_HTML = """
FlowDeck — Login
FlowDeck
Login or create an account to continue
Or continue with
"""
@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)
from fastapi.responses import HTMLResponse
if provider == "local":
return HTMLResponse(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.
""",
status_code=200,
)
state = secrets.token_hex(32)
request.session["oauth_state"] = state
request.session["oauth_provider"] = provider
# Dynamic redirect URI based on incoming Host header
host = request.headers.get("host", "localhost:8080")
dynamic_redirect_uri = f"http://{host}/auth/callback"
auth_url = oauth_provider.get_authorize_url(state, redirect_uri=dynamic_redirect_uri)
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)
# First user is admin
user_count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
is_admin = 1 if 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)
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")
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)
_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."""
# Validate state
expected_state = request.session.get("oauth_state", "")
provider_name = request.session.get("oauth_provider", "gitea")
if not expected_state or state != expected_state:
return HTMLResponse("
Invalid state
", 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 — use dynamic redirect URI matching the authorize step
host = request.headers.get("host", "localhost:8080")
dynamic_redirect_uri = f"http://{host}/auth/callback"
token_data = await oauth_provider.exchange_code(code, redirect_uri=dynamic_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)
# 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)
VALUES (?, ?, ?, ?)
ON CONFLICT(login)
DO UPDATE SET full_name=excluded.full_name, email=excluded.email, avatar_url=excluded.avatar_url""",
(login_id, oauth_user.get("full_name", ""), oauth_user.get("email", ""), oauth_user.get("avatar_url", "")),
)
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)
_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")
async def logout():
"""Clear session and redirect to login page."""
response = RedirectResponse(url="/auth/login?provider=local", 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}
# ── 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:
pass