95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app import config
|
|
from app.auth import (
|
|
authenticate_user,
|
|
create_session,
|
|
delete_session,
|
|
check_login_rate_limit,
|
|
record_login_failure,
|
|
record_login_success,
|
|
change_user_password,
|
|
require_user,
|
|
)
|
|
|
|
router = APIRouter(tags=["Auth"])
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str = Field(..., min_length=1)
|
|
password: str = Field(..., min_length=1)
|
|
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
old_password: str = Field(..., min_length=1)
|
|
new_password: str = Field(..., min_length=6)
|
|
|
|
|
|
@router.post("/api/login")
|
|
async def api_login(req: LoginRequest, request: Request, response: Response):
|
|
forwarded = request.headers.get("x-forwarded-for")
|
|
if forwarded:
|
|
ip = forwarded.split(",")[0].strip()
|
|
else:
|
|
ip = request.headers.get("x-real-ip") or (request.client.host if request.client else "unknown")
|
|
|
|
if not check_login_rate_limit(ip):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail="Too many login attempts. Please try again in a minute.",
|
|
)
|
|
|
|
user = authenticate_user(req.username, req.password)
|
|
if not user:
|
|
record_login_failure(ip)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid username or password",
|
|
)
|
|
|
|
record_login_success(ip)
|
|
user_agent = request.headers.get("User-Agent", "")
|
|
session_token = create_session(user["id"], user_agent=user_agent, ip=ip)
|
|
|
|
is_secure = request.headers.get("x-forwarded-proto", "").lower() == "https" or request.url.scheme == "https"
|
|
response.set_cookie(
|
|
key=config.SESSION_COOKIE_NAME,
|
|
value=session_token,
|
|
max_age=config.SESSION_MAX_AGE,
|
|
httponly=True,
|
|
samesite="lax",
|
|
secure=is_secure,
|
|
)
|
|
return {
|
|
"status": "ok",
|
|
"user": {"id": user["id"], "username": user["username"]},
|
|
"token": session_token,
|
|
}
|
|
|
|
|
|
@router.post("/api/logout")
|
|
async def api_logout(request: Request, response: Response):
|
|
token = request.cookies.get(config.SESSION_COOKIE_NAME)
|
|
if token:
|
|
delete_session(token)
|
|
response.delete_cookie(key=config.SESSION_COOKIE_NAME)
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/api/me")
|
|
async def api_me(user: dict = Depends(require_user)):
|
|
return {"status": "ok", "user": {"id": user["id"], "username": user["username"]}}
|
|
|
|
|
|
@router.post("/api/change-password")
|
|
@router.post("/api/auth/change-password")
|
|
async def api_change_password(req: ChangePasswordRequest, user: dict = Depends(require_user)):
|
|
ok, msg = change_user_password(user["id"], req.old_password, req.new_password)
|
|
if not ok:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=msg or "Current password does not match",
|
|
)
|
|
return {"status": "ok", "message": "Password changed successfully"}
|