111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
# app/api/v1/admin/auth.py
|
|
"""
|
|
Admin authentication endpoints.
|
|
|
|
Implements dual token storage:
|
|
- Sets HTTP-only cookie for browser page navigation
|
|
- Returns token in response for localStorage (API calls)
|
|
"""
|
|
|
|
import logging
|
|
from fastapi import APIRouter, Depends, Response
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.database import get_db
|
|
from app.services.auth_service import auth_service
|
|
from app.exceptions import InvalidCredentialsException
|
|
from models.schema.auth import LoginResponse, UserLogin, UserResponse
|
|
from models.database.user import User
|
|
from app.api.deps import get_current_admin_user
|
|
from app.core.config import settings
|
|
|
|
router = APIRouter(prefix="/auth")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
def admin_login(
|
|
user_credentials: UserLogin,
|
|
response: Response,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Admin login endpoint.
|
|
|
|
Only allows users with 'admin' role to login.
|
|
Returns JWT token for authenticated admin users.
|
|
|
|
Sets token in two places:
|
|
1. HTTP-only cookie (for browser page navigation)
|
|
2. Response body (for localStorage and API calls)
|
|
"""
|
|
# Authenticate user
|
|
login_result = auth_service.login_user(db=db, user_credentials=user_credentials)
|
|
|
|
# Verify user is admin
|
|
if login_result["user"].role != "admin":
|
|
logger.warning(f"Non-admin user attempted admin login: {user_credentials.username}")
|
|
raise InvalidCredentialsException("Admin access required")
|
|
|
|
logger.info(f"Admin login successful: {login_result['user'].username}")
|
|
|
|
# Set HTTP-only cookie for browser navigation
|
|
response.set_cookie(
|
|
key="admin_token",
|
|
value=login_result["token_data"]["access_token"],
|
|
httponly=True, # JavaScript cannot access (XSS protection)
|
|
secure=False, # Set to True in production (requires HTTPS)
|
|
samesite="lax", # CSRF protection
|
|
max_age=login_result["token_data"]["expires_in"], # Match JWT expiry
|
|
path="/", # Available for all routes
|
|
)
|
|
|
|
logger.debug(f"Set admin_token cookie with {login_result['token_data']['expires_in']}s expiry")
|
|
|
|
# Also return token in response for localStorage (API calls)
|
|
return LoginResponse(
|
|
access_token=login_result["token_data"]["access_token"],
|
|
token_type=login_result["token_data"]["token_type"],
|
|
expires_in=login_result["token_data"]["expires_in"],
|
|
user=login_result["user"],
|
|
)
|
|
|
|
|
|
@router.get("/me", response_model=UserResponse)
|
|
def get_current_admin(current_user: User = Depends(get_current_admin_user)):
|
|
"""
|
|
Get current authenticated admin user.
|
|
|
|
This endpoint validates the token and ensures the user has admin privileges.
|
|
Returns the current user's information.
|
|
|
|
Token can come from:
|
|
- Authorization header (API calls)
|
|
- admin_token cookie (browser navigation)
|
|
"""
|
|
logger.info(f"Admin user info requested: {current_user.username}")
|
|
|
|
# Pydantic will automatically serialize the User model to UserResponse
|
|
return current_user
|
|
|
|
|
|
@router.post("/logout")
|
|
def admin_logout(response: Response):
|
|
"""
|
|
Admin logout endpoint.
|
|
|
|
Clears the admin_token cookie.
|
|
Client should also remove token from localStorage.
|
|
"""
|
|
logger.info("Admin logout")
|
|
|
|
# Clear the cookie
|
|
response.delete_cookie(
|
|
key="admin_token",
|
|
path="/",
|
|
)
|
|
|
|
logger.debug("Deleted admin_token cookie")
|
|
|
|
return {"message": "Logged out successfully"}
|