auth.py (3138B)
1 """Auth routes: setup, login, logout, status.""" 2 3 import json 4 import os 5 import time 6 from pathlib import Path 7 8 from fastapi import APIRouter, HTTPException, Request, Response 9 10 from app.auth import ( 11 clear_jwt_cookie, 12 create_jwt, 13 get_jwt_from_cookie, 14 hash_password, 15 set_jwt_cookie, 16 verify_jwt, 17 verify_password, 18 ) 19 from app.config import DashboardConfig 20 from app.models import AuthResponse, AuthStatus, LoginRequest, SetupRequest 21 22 router = APIRouter(prefix="/api/auth", tags=["auth"]) 23 24 _CONFIG_PATH = os.environ.get("THEO_CONFIG_PATH", "data/config.json") 25 _AUTH_DB: dict[str, str] = {} 26 27 28 def _load_auth_config() -> dict[str, str] | None: # pragma: no mutate: block 29 """Load stored auth config from disk.""" 30 config_path = Path(_CONFIG_PATH) 31 if not config_path.exists(): 32 return None 33 return json.loads(config_path.read_text()) 34 35 36 def _save_auth_config(password_hash: str, api_key: str) -> None: # pragma: no mutate: block 37 """Persist auth config to disk.""" 38 config_path = Path(_CONFIG_PATH) 39 config_path.parent.mkdir(parents=True, exist_ok=True) 40 config_path.write_text(json.dumps({ 41 "hermes_api_key": api_key, 42 "_password_hash": password_hash, 43 "setup_at": int(time.time()), 44 })) 45 46 47 @router.get("/status", response_model=AuthStatus) 48 async def auth_status(request: Request) -> AuthStatus: # pragma: no mutate: block 49 """Check if auth is configured.""" 50 config = _load_auth_config() 51 authenticated = False 52 token = get_jwt_from_cookie(request) 53 if token: 54 try: 55 verify_jwt(token) 56 authenticated = True 57 except Exception: 58 pass 59 return AuthStatus(configured=config is not None, authenticated=authenticated) 60 61 62 @router.post("/setup", response_model=AuthResponse) 63 async def auth_setup(payload: SetupRequest, response: Response) -> AuthResponse: # pragma: no mutate: block 64 """First-run: set password and API key.""" 65 existing = _load_auth_config() 66 if existing is not None: 67 raise HTTPException(status_code=400, detail="Already configured") 68 hashed = hash_password(payload.password) 69 _save_auth_config(hashed, payload.api_key) 70 token = create_jwt({"sub": "admin"}) 71 set_jwt_cookie(response, token) 72 return AuthResponse(token=token) 73 74 75 @router.post("/login", response_model=AuthResponse) 76 async def auth_login(payload: LoginRequest, response: Response) -> AuthResponse: # pragma: no mutate: block 77 """Login with password, returns JWT.""" 78 config = _load_auth_config() 79 if config is None: 80 raise HTTPException(status_code=401, detail="Not configured") 81 if not verify_password(payload.password, config["_password_hash"]): 82 raise HTTPException(status_code=401, detail="Invalid password") 83 token = create_jwt({"sub": "admin"}) 84 set_jwt_cookie(response, token) 85 return AuthResponse(token=token) 86 87 88 @router.post("/logout") 89 async def auth_logout() -> Response: # pragma: no mutate: block 90 """Clear the JWT cookie.""" 91 resp = Response(content='{"ok": true}', media_type="application/json") 92 clear_jwt_cookie(resp) 93 return resp