94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
|
"""Nexus — Auth API Routes (Login / TOTP / Session management)
|
||
|
|
Presentation layer — receives HTTP requests, delegates to AuthService.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from typing import Optional
|
||
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
||
|
|
|
||
|
|
from server.api.dependencies import get_auth_service
|
||
|
|
from server.application.services.auth_service import AuthService
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/login", response_model=dict)
|
||
|
|
async def login(
|
||
|
|
request: Request,
|
||
|
|
payload: dict,
|
||
|
|
service: AuthService = Depends(get_auth_service),
|
||
|
|
):
|
||
|
|
"""Authenticate admin user (password + optional TOTP)
|
||
|
|
|
||
|
|
Body: {
|
||
|
|
"username": "admin",
|
||
|
|
"password": "...",
|
||
|
|
"totp_code": "123456" // Optional, required if TOTP enabled
|
||
|
|
}
|
||
|
|
"""
|
||
|
|
ip_address = request.client.host if request.client else "unknown"
|
||
|
|
result = await service.login(
|
||
|
|
username=payload.get("username", ""),
|
||
|
|
password=payload.get("password", ""),
|
||
|
|
ip_address=ip_address,
|
||
|
|
totp_code=payload.get("totp_code"),
|
||
|
|
)
|
||
|
|
if not result["success"]:
|
||
|
|
status_code = 401
|
||
|
|
if result.get("reason") == "account_locked":
|
||
|
|
status_code = 429
|
||
|
|
raise HTTPException(status_code=status_code, detail=result.get("message", "Login failed"))
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/totp/setup", response_model=dict)
|
||
|
|
async def setup_totp(
|
||
|
|
payload: dict,
|
||
|
|
service: AuthService = Depends(get_auth_service),
|
||
|
|
):
|
||
|
|
"""Generate TOTP secret for admin user (requires valid session token)
|
||
|
|
|
||
|
|
Body: {"admin_id": 1}
|
||
|
|
"""
|
||
|
|
# TODO: Verify session token from Authorization header
|
||
|
|
admin_id = payload.get("admin_id")
|
||
|
|
if not admin_id:
|
||
|
|
raise HTTPException(status_code=400, detail="admin_id required")
|
||
|
|
result = await service.setup_totp(admin_id)
|
||
|
|
if not result["success"]:
|
||
|
|
raise HTTPException(status_code=400, detail=result.get("reason", "Setup failed"))
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/totp/enable", response_model=dict)
|
||
|
|
async def enable_totp(
|
||
|
|
payload: dict,
|
||
|
|
service: AuthService = Depends(get_auth_service),
|
||
|
|
):
|
||
|
|
"""Enable TOTP after verification (requires valid session token)
|
||
|
|
|
||
|
|
Body: {"admin_id": 1, "totp_code": "123456"}
|
||
|
|
"""
|
||
|
|
admin_id = payload.get("admin_id")
|
||
|
|
totp_code = payload.get("totp_code")
|
||
|
|
if not admin_id or not totp_code:
|
||
|
|
raise HTTPException(status_code=400, detail="admin_id and totp_code required")
|
||
|
|
result = await service.enable_totp(admin_id, totp_code)
|
||
|
|
if not result["success"]:
|
||
|
|
raise HTTPException(status_code=400, detail=result.get("message", "Enable failed"))
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/totp/disable", response_model=dict)
|
||
|
|
async def disable_totp(
|
||
|
|
payload: dict,
|
||
|
|
service: AuthService = Depends(get_auth_service),
|
||
|
|
):
|
||
|
|
"""Disable TOTP for admin user (requires valid session token)
|
||
|
|
|
||
|
|
Body: {"admin_id": 1}
|
||
|
|
"""
|
||
|
|
admin_id = payload.get("admin_id")
|
||
|
|
if not admin_id:
|
||
|
|
raise HTTPException(status_code=400, detail="admin_id required")
|
||
|
|
result = await service.disable_totp(admin_id)
|
||
|
|
return result
|