"""Authentication endpoints.""" from fastapi import APIRouter, Depends, HTTPException, Request, Response from sqlalchemy.orm import Session from app.auth.dependencies import get_current_user, verify_csrf from app.auth.hashing import hash_password, verify_password from app.auth.rate_limit import check_rate_limit, clear_attempts, record_failed_attempt from app.auth.session import clear_session, create_session, verify_session from app.config import settings from app.db import get_db from app.models.settings import Settings from app.schemas.auth import AuthStatusResponse, ChangePasswordRequest, LoginRequest, SetupRequest router = APIRouter(prefix="/auth", tags=["auth"]) def get_client_ip(request: Request) -> str: """Helper to extract client IP for rate limiting.""" forwarded = request.headers.get("X-Forwarded-For") if forwarded: return forwarded.split(",")[0].strip() return request.client.host if request.client else "unknown" @router.get("/status", response_model=AuthStatusResponse) def get_auth_status(request: Request, db: Session = Depends(get_db)) -> AuthStatusResponse: """Check if app requires setup and if user is logged in.""" app_settings = db.query(Settings).first() if not app_settings: app_settings = Settings() db.add(app_settings) db.commit() setup_required = app_settings.password_hash is None token = request.cookies.get(settings.COOKIE_NAME) logged_in = False if token: user_id = verify_session(token) if user_id is not None: logged_in = True return AuthStatusResponse(setupRequired=setup_required, loggedIn=logged_in) @router.post("/setup", dependencies=[Depends(verify_csrf)]) def setup_password( data: SetupRequest, response: Response, db: Session = Depends(get_db) ) -> dict[str, str]: """First-run setup.""" app_settings = db.query(Settings).first() if app_settings and app_settings.password_hash: raise HTTPException(status_code=409, detail="Password already set") if not app_settings: app_settings = Settings() db.add(app_settings) app_settings.password_hash = hash_password(data.password) db.commit() # Log them in automatically create_session(response) return {"status": "ok"} @router.post("/login", dependencies=[Depends(verify_csrf)]) def login( request: Request, data: LoginRequest, response: Response, db: Session = Depends(get_db) ) -> dict[str, str]: """Validate password and issue session cookie.""" ip = get_client_ip(request) check_rate_limit(ip) app_settings = db.query(Settings).first() if not app_settings or not app_settings.password_hash: record_failed_attempt(ip) raise HTTPException(status_code=401, detail="Invalid credentials") if not verify_password(app_settings.password_hash, data.password): record_failed_attempt(ip) raise HTTPException(status_code=401, detail="Invalid credentials") clear_attempts(ip) create_session(response) return {"status": "ok"} @router.post("/logout", dependencies=[Depends(verify_csrf)]) def logout(response: Response) -> dict[str, str]: """Clear session cookie.""" clear_session(response) return {"status": "ok"} @router.put("/password", dependencies=[Depends(verify_csrf), Depends(get_current_user)]) def change_password( request: Request, data: ChangePasswordRequest, db: Session = Depends(get_db) ) -> dict[str, str]: """Change an existing password.""" ip = get_client_ip(request) check_rate_limit(ip) app_settings = db.query(Settings).first() if not app_settings or not app_settings.password_hash: raise HTTPException(status_code=400, detail="Setup required first") if not verify_password(app_settings.password_hash, data.current_password): record_failed_attempt(ip) raise HTTPException(status_code=401, detail="Invalid current password") clear_attempts(ip) app_settings.password_hash = hash_password(data.new_password) db.commit() return {"status": "ok"}