#!/usr/bin/env python3 """ Authentication routes for GNSS Guard Server Handles user session authentication for the web UI """ import logging from datetime import datetime, timedelta from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Response, Request from fastapi.responses import RedirectResponse from pydantic import BaseModel from slowapi import Limiter from slowapi.util import get_remote_address from config import get_config logger = logging.getLogger("gnss_guard.server.auth") router = APIRouter(tags=["auth"]) # Rate limiter instance (uses app.state.limiter set in main.py) limiter = Limiter(key_func=get_remote_address) # Simple in-memory session storage (for single-user scenario) # In production with multiple servers, use Redis or database _sessions: dict = {} class LoginRequest(BaseModel): username: str password: str def create_session(username: str) -> str: """Create a new session and return session ID""" import secrets session_id = secrets.token_urlsafe(32) config = get_config() _sessions[session_id] = { "username": username, "created_at": datetime.utcnow(), "expires_at": datetime.utcnow() + timedelta(minutes=config.session_expire_minutes) } return session_id def validate_session(session_id: str) -> Optional[str]: """Validate session and return username if valid""" if not session_id or session_id not in _sessions: return None session = _sessions[session_id] if datetime.utcnow() > session["expires_at"]: del _sessions[session_id] return None return session["username"] def get_current_user(request: Request) -> str: """ Dependency to get current authenticated user. Raises 401 if not authenticated. """ session_id = request.cookies.get("session_id") username = validate_session(session_id) if not username: raise HTTPException( status_code=401, detail="Not authenticated", headers={"WWW-Authenticate": "Bearer"} ) return username def get_optional_user(request: Request) -> Optional[str]: """ Dependency to get current user if authenticated, None otherwise. """ session_id = request.cookies.get("session_id") return validate_session(session_id) @router.post("/login") @limiter.limit("5/minute") # Rate limit: 5 login attempts per minute per IP async def login(request: Request, data: LoginRequest, response: Response): """ Login endpoint - validates credentials and sets session cookie. Rate limited to prevent brute force attacks. """ config = get_config() # Verify credentials against hardcoded user if data.username != config.web_username or data.password != config.web_password: logger.warning(f"Failed login attempt for user: {data.username} from IP: {request.client.host}") raise HTTPException(status_code=401, detail="Invalid credentials") # Create session session_id = create_session(data.username) # Set session cookie # secure=True ensures cookie only sent over HTTPS response.set_cookie( key="session_id", value=session_id, httponly=True, secure=True, # Only send over HTTPS samesite="lax", max_age=config.session_expire_minutes * 60 ) logger.info(f"User logged in: {data.username}") return {"message": "Login successful", "username": data.username} @router.post("/logout") async def logout(request: Request, response: Response): """ Logout endpoint - clears session. """ session_id = request.cookies.get("session_id") if session_id and session_id in _sessions: del _sessions[session_id] response.delete_cookie("session_id") return {"message": "Logged out successfully"} @router.get("/auth/check") async def check_auth(request: Request): """ Check if current session is authenticated. """ session_id = request.cookies.get("session_id") username = validate_session(session_id) if username: return {"authenticated": True, "username": username} else: return {"authenticated": False}