import os from fastapi import APIRouter, HTTPException, Request, Response from app.models import AuthPayload from app.services.auth_service import ( SESSION_TTL_SECONDS, create_session, create_user, delete_session, get_user_for_session, get_last_session_meta, verify_user, ) from app.services.email_service import send_email router = APIRouter(prefix="/api") SESSION_COOKIE_NAME = "session_id" COOKIE_SECURE = os.getenv("COOKIE_SECURE", "0") == "1" COOKIE_SAMESITE = (os.getenv("COOKIE_SAMESITE") or "lax").lower() def _set_session_cookie(response: Response, session_id: str): same_site = COOKIE_SAMESITE if COOKIE_SAMESITE in {"lax", "strict", "none"} else "lax" response.set_cookie( SESSION_COOKIE_NAME, session_id, httponly=True, samesite=same_site, max_age=SESSION_TTL_SECONDS, secure=COOKIE_SECURE, path="/", ) def _get_identifier(payload: AuthPayload) -> str: identifier = payload.username or payload.email or "" return identifier.strip() @router.post("/signup") def signup(payload: AuthPayload, response: Response): identifier = _get_identifier(payload) if not identifier or not payload.password: raise HTTPException(status_code=400, detail="Email and password are required") user = create_user(identifier, payload.password) if not user: raise HTTPException(status_code=409, detail="User already exists") session_id = create_session(user["id"]) _set_session_cookie(response, session_id) try: body = ( "Welcome to Quantfortune!\n\n" "Your account has been created successfully.\n\n" "You can now log in and start using the platform.\n\n" "Quantfortune Support" ) send_email(user["username"], "Welcome to Quantfortune", body) except Exception: pass return {"id": user["id"], "username": user["username"], "role": user.get("role")} @router.post("/login") def login(payload: AuthPayload, response: Response, request: Request): identifier = _get_identifier(payload) if not identifier or not payload.password: raise HTTPException(status_code=400, detail="Email and password are required") user = verify_user(identifier, payload.password) if not user: raise HTTPException(status_code=401, detail="Invalid email or password") client_ip = request.client.host if request.client else None user_agent = request.headers.get("user-agent") last_meta = get_last_session_meta(user["id"]) if last_meta.get("ip") and ( last_meta.get("ip") != client_ip or last_meta.get("user_agent") != user_agent ): try: body = ( "New login detected on your Quantfortune account.\n\n" f"IP: {client_ip or 'unknown'}\n" f"Device: {user_agent or 'unknown'}\n\n" "If this wasn't you, please reset your password immediately." ) send_email(user["username"], "New login detected", body) except Exception: pass session_id = create_session(user["id"], ip=client_ip, user_agent=user_agent) _set_session_cookie(response, session_id) return {"id": user["id"], "username": user["username"], "role": user.get("role")} @router.post("/logout") def logout(request: Request, response: Response): session_id = request.cookies.get(SESSION_COOKIE_NAME) if session_id: delete_session(session_id) response.delete_cookie(SESSION_COOKIE_NAME, path="/") return {"ok": True} @router.get("/me") def me(request: Request): session_id = request.cookies.get(SESSION_COOKIE_NAME) if not session_id: raise HTTPException(status_code=401, detail="Not authenticated") user = get_user_for_session(session_id) if not user: raise HTTPException(status_code=401, detail="Not authenticated") return {"id": user["id"], "username": user["username"], "role": user.get("role")}