85 lines
3.2 KiB
Python
85 lines
3.2 KiB
Python
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials
|
|
|
|
from src.server.models.auth import AuthStatus, LoginRequest, LoginResponse, SetupRequest
|
|
from src.server.services.auth_service import AuthError, LockedOutError, auth_service
|
|
|
|
# NOTE: import dependencies (optional_auth, security) lazily inside handlers
|
|
# to avoid importing heavyweight modules (e.g. sqlalchemy) at import time.
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
|
|
@router.post("/setup", status_code=status.HTTP_201_CREATED)
|
|
def setup_auth(req: SetupRequest):
|
|
"""Initial setup endpoint to configure the master password."""
|
|
if auth_service.is_configured():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Master password already configured",
|
|
)
|
|
|
|
try:
|
|
auth_service.setup_master_password(req.master_password)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
def login(req: LoginRequest):
|
|
"""Validate master password and return JWT token."""
|
|
# Use a simple identifier for failed attempts; prefer IP in a real app
|
|
identifier = "global"
|
|
|
|
try:
|
|
valid = auth_service.validate_master_password(req.password, identifier=identifier)
|
|
except AuthError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except LockedOutError as e:
|
|
raise HTTPException(status_code=429, detail=str(e))
|
|
|
|
if not valid:
|
|
raise HTTPException(status_code=401, detail="Invalid credentials")
|
|
|
|
token = auth_service.create_access_token(subject="master", remember=bool(req.remember))
|
|
return token
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(credentials: HTTPAuthorizationCredentials = None):
|
|
"""Logout by revoking token (no-op for stateless JWT)."""
|
|
# Import security dependency lazily to avoid heavy imports during test
|
|
if credentials is None:
|
|
from fastapi import Depends
|
|
|
|
from src.server.utils.dependencies import security as _security
|
|
|
|
# Trigger dependency resolution during normal request handling
|
|
credentials = Depends(_security)
|
|
|
|
# If a plain credentials object was provided, extract token
|
|
token = getattr(credentials, "credentials", None)
|
|
# Placeholder; auth_service.revoke_token can be expanded to persist revocations
|
|
auth_service.revoke_token(token)
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/status", response_model=AuthStatus)
|
|
def status(auth: Optional[dict] = None):
|
|
"""Return whether master password is configured and if caller is authenticated."""
|
|
# Lazy import to avoid pulling in database/sqlalchemy during module import
|
|
from fastapi import Depends
|
|
try:
|
|
from src.server.utils.dependencies import optional_auth as _optional_auth
|
|
except Exception:
|
|
_optional_auth = None
|
|
|
|
# If dependency injection didn't provide auth, attempt to resolve optionally
|
|
if auth is None and _optional_auth is not None:
|
|
auth = Depends(_optional_auth)
|
|
return AuthStatus(configured=auth_service.is_configured(), authenticated=bool(auth))
|