117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
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")}
|