808fbf5c7c
<message>Update the _set_golden_from_path function to improve the handling of existing golden image files. Replace the existing unlink logic with a more robust method that safely removes files or broken symlinks using the missing_ok parameter. This change enhances the reliability of the backup upload process by ensuring that stale references are properly cleared before setting a new golden image path.
151 lines
4.2 KiB
Python
151 lines
4.2 KiB
Python
#!/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}
|
|
|