feat: Add session-based admin authentication

pyproject.toml

@@ -66,6 +66,7 @@ dependencies = [
   "packaging>=25.0",
   "paho-mqtt",
   "pandas",
+  "passlib[argon2]>=1.7.4",
   "plotly",
   "psutil>=7.0.0",
   "pydub>=0.25.1",
@@ -79,6 +80,7 @@ dependencies = [
   "sqladmin>=0.21.0",
   "sqlalchemy",
   "sqlmodel>=0.0.24",
+  "starsessions[redis]>=2.2.1",
   "structlog>=25.4.0",
   "suntime",
   "tqdm>=4.67.1",

src/birdnetpi/utils/auth.py

@@ -0,0 +1,182 @@
+"""Authentication utilities for BirdNET-Pi admin interface.
+
+Provides session-based authentication using Starlette's built-in
+authentication system with Redis-backed sessions.
+"""
+
+from collections.abc import Awaitable, Callable
+from datetime import datetime
+
+from passlib.context import CryptContext
+from pydantic import BaseModel
+from starlette.authentication import (
+    AuthCredentials,
+    AuthenticationBackend,
+    SimpleUser,
+)
+from starlette.requests import HTTPConnection
+from starlette.responses import RedirectResponse
+from starsessions import load_session
+
+from birdnetpi.system.path_resolver import PathResolver
+
+# Password hashing context using Argon2
+pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
+
+
+def require_admin_relative(
+    redirect_path: str = "/admin/login",
+) -> Callable[[Callable[..., Awaitable[object]]], Callable[..., Awaitable[object]]]:
+    """Create authentication decorator that uses relative URLs for redirects.
+
+    Unlike Starlette's @requires which generates absolute URLs, this decorator
+    uses relative paths to avoid issues with proxies and URL parsing.
+
+    Args:
+        redirect_path: Relative path to redirect to if not authenticated
+
+    Returns:
+        Decorator function that wraps route handlers
+    """
+    from functools import wraps
+    from urllib.parse import urlencode
+
+    def decorator(
+        func: Callable[..., Awaitable[object]],
+    ) -> Callable[..., Awaitable[object]]:
+        @wraps(func)
+        async def wrapper(request: HTTPConnection, *args: object, **kwargs: object) -> object:
+            # Check if user has required authentication scope
+            # This uses request.auth which works with or without AuthenticationMiddleware
+            if "authenticated" not in request.auth.scopes:
+                # Build relative redirect URL with next parameter
+                next_qparam = urlencode({"next": str(request.url.path)})
+                if request.url.query:
+                    next_qparam = urlencode({"next": f"{request.url.path}?{request.url.query}"})
+
+                redirect_url = f"{redirect_path}?{next_qparam}"
+                return RedirectResponse(url=redirect_url, status_code=303)
+
+            return await func(request, *args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
+# Syntactic sugar for common authentication requirement
+# Usage: @require_admin decorator on admin view routes
+require_admin = require_admin_relative()
+
+
+class AdminUser(BaseModel):
+    """Admin user model for file-based storage."""
+
+    username: str
+    password_hash: str
+    created_at: datetime
+
+
+class AuthService:
+    """Handles admin user file operations and password hashing.
+
+    Stores a single admin user in a JSON file with permissions set to 0600.
+    """
+
+    def __init__(self, path_resolver: PathResolver) -> None:
+        """Initialize auth service.
+
+        Args:
+            path_resolver: PathResolver instance for determining file paths
+        """
+        self.admin_file = path_resolver.get_data_dir() / "admin_user.json"
+
+    def load_admin_user(self) -> AdminUser | None:
+        """Load admin user from JSON file.
+
+        Returns:
+            AdminUser if file exists and is valid, None otherwise
+        """
+        if not self.admin_file.exists():
+            return None
+
+        import json
+
+        try:
+            with open(self.admin_file) as f:
+                data = json.load(f)
+            return AdminUser(**data)
+        except (json.JSONDecodeError, ValueError):
+            return None
+
+    def save_admin_user(self, username: str, password: str) -> None:
+        """Hash password and save to JSON with 0600 permissions.
+
+        Args:
+            username: Admin username
+            password: Plain text password (will be hashed)
+        """
+        import json
+
+        admin = AdminUser(
+            username=username, password_hash=pwd_context.hash(password), created_at=datetime.now()
+        )
+
+        # Write to file
+        with open(self.admin_file, "w") as f:
+            json.dump(admin.model_dump(), f, default=str, indent=2)
+
+        # Set restrictive permissions (owner read/write only)
+        self.admin_file.chmod(0o600)
+
+    def verify_password(self, password: str, password_hash: str) -> bool:
+        """Verify password against hash.
+
+        Args:
+            password: Plain text password to verify
+            password_hash: Argon2 hash to verify against
+
+        Returns:
+            True if password matches, False otherwise
+        """
+        return pwd_context.verify(password, password_hash)
+
+    def admin_exists(self) -> bool:
+        """Check if admin user file exists.
+
+        Returns:
+            True if admin_user.json exists, False otherwise
+        """
+        return self.admin_file.exists()
+
+
+class SessionAuthBackend(AuthenticationBackend):
+    """Session-based authentication backend for Starlette.
+
+    Checks for username in session and returns appropriate credentials.
+    """
+
+    async def authenticate(self, conn: HTTPConnection) -> tuple[AuthCredentials, SimpleUser] | None:
+        """Authenticate request based on session data.
+
+        Called by AuthenticationMiddleware on every request. Explicitly loads
+        session from starsessions middleware before accessing it.
+
+        Args:
+            conn: HTTP connection (request or WebSocket)
+
+        Returns:
+            Tuple of (AuthCredentials, SimpleUser) if authenticated,
+            None if not authenticated
+        """
+        # Load session from starsessions middleware
+        await load_session(conn)
+
+        # Get username from session
+        username = conn.session.get("username")
+        if not username:
+            return None  # Not authenticated
+
+        # Return authenticated user with "authenticated" scope
+        # The scope is used by @requires decorator for authorization
+        return AuthCredentials(["authenticated"]), SimpleUser(username)

src/birdnetpi/web/core/container.py

@@ -27,6 +27,7 @@
 from birdnetpi.system.log_reader import LogReaderService
 from birdnetpi.system.path_resolver import PathResolver
 from birdnetpi.system.system_control import SystemControlService
+from birdnetpi.utils.auth import AuthService
 from birdnetpi.utils.cache import Cache
 from birdnetpi.web.core.config import get_config
 
@@ -128,6 +129,19 @@ class Container(containers.DeclarativeContainer):
         enable_cache_warming=True,
     )
 
+    # Authentication services
+    auth_service = providers.Singleton(
+        AuthService,
+        path_resolver=path_resolver,
+    )
+
+    # Redis client for session storage - singleton
+    redis_client = providers.Singleton(
+        lambda: __import__("redis.asyncio", fromlist=["Redis"]).Redis.from_url(
+            "redis://127.0.0.1:6379"
+        ),
+    )
+
     # Core business services - singletons
     file_manager = providers.Singleton(
         FileManager,

src/birdnetpi/web/core/factory.py

@@ -3,18 +3,24 @@
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import HTMLResponse
+from starlette.middleware.authentication import AuthenticationMiddleware
+from starsessions import SessionMiddleware
+from starsessions.stores.redis import RedisStore
 
 from birdnetpi.config.manager import ConfigManager
 from birdnetpi.i18n.translation_manager import setup_jinja2_i18n
 from birdnetpi.system.status import SystemInspector
+from birdnetpi.utils.auth import SessionAuthBackend
 from birdnetpi.utils.language import get_user_language
 from birdnetpi.web.core.container import Container
 from birdnetpi.web.core.lifespan import lifespan
 from birdnetpi.web.middleware.i18n import LanguageMiddleware
 from birdnetpi.web.middleware.request_logging import StructuredRequestLoggingMiddleware
+from birdnetpi.web.middleware.setup_redirect import SetupRedirectMiddleware
 from birdnetpi.web.middleware.update_banner import add_update_status_to_templates
 from birdnetpi.web.routers import (
     analysis_api_routes,
+    auth_routes,
     detections_api_routes,
     health_api_routes,
     i18n_api_routes,
@@ -75,6 +81,37 @@ def create_app() -> FastAPI:
         expose_headers=["*"],  # Expose all headers including Content-Type
     )
 
+    # Authentication and session middleware
+    # NOTE: Middleware is stacked in reverse order - last added runs first!
+    # Desired execution order: Session → Auth → SetupRedirect → App
+    # So add in reverse: SetupRedirect, Auth, Session
+
+    # 1. Setup redirect (added first, runs last before app)
+    auth_service = container.auth_service()
+    app.add_middleware(SetupRedirectMiddleware, auth_service=auth_service)
+
+    # 2. Authentication (added second, runs after session loads)
+    app.add_middleware(
+        AuthenticationMiddleware,
+        backend=SessionAuthBackend(),
+    )
+
+    # 3. Session middleware (added last, runs first to load session)
+    redis_client = container.redis_client()
+    session_store = RedisStore(
+        connection=redis_client,
+        prefix="birdnetpi:",
+        gc_ttl=86400,  # 24 hours
+    )
+    app.add_middleware(
+        SessionMiddleware,
+        store=session_store,
+        lifetime=86400,  # 24 hours
+        rolling=True,  # Extend session on each request
+        cookie_https_only=False,  # TODO: Enable in production with HTTPS
+        cookie_name="birdnetpi_session",
+    )
+
     # Add LanguageMiddleware
     app.add_middleware(LanguageMiddleware)
 
@@ -102,6 +139,7 @@ def create_app() -> FastAPI:
         modules=[
             "birdnetpi.web.core.factory",  # Wire factory for root route
             "birdnetpi.web.routers.analysis_api_routes",
+            "birdnetpi.web.routers.auth_routes",
             "birdnetpi.web.routers.detections_api_routes",
             "birdnetpi.web.routers.health_api_routes",
             "birdnetpi.web.routers.i18n_api_routes",  # Wire i18n API routes
@@ -125,6 +163,13 @@ def create_app() -> FastAPI:
 
     # === API Routes (included in documentation) ===
 
+    # Authentication routes (setup, login, logout)
+    app.include_router(
+        auth_routes.router,
+        tags=["Authentication"],
+        include_in_schema=False,  # Exclude from API docs
+    )
+
     # Analysis API routes for progressive loading
     app.include_router(analysis_api_routes.router, prefix="/api", tags=["Analysis API"])
 
@@ -205,6 +250,12 @@ def create_app() -> FastAPI:
     # Database administration interface
     sqladmin_view_routes.setup_sqladmin(app)
 
+    # Cleanup Redis connection on shutdown
+    @app.on_event("shutdown")
+    async def shutdown() -> None:
+        """Clean up resources on application shutdown."""
+        await redis_client.close()
+
     # Root route (excluded from API documentation)
     @app.get("/", response_class=HTMLResponse, include_in_schema=False)
     async def read_root(request: Request) -> HTMLResponse:

src/birdnetpi/web/middleware/setup_redirect.py

@@ -0,0 +1,64 @@
+"""Setup redirect middleware for BirdNET-Pi.
+
+Redirects all requests to the setup wizard if no admin user exists.
+"""
+
+from collections.abc import Callable
+
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import RedirectResponse, Response
+from starlette.types import ASGIApp
+
+from birdnetpi.utils.auth import AuthService
+
+
+class SetupRedirectMiddleware(BaseHTTPMiddleware):
+    """Redirect to setup wizard if no admin user exists.
+
+    This middleware checks if an admin user has been created. If not,
+    it redirects all requests to /admin/setup except for:
+    - The setup page itself
+    - The login page
+    - Static files
+    - Health check endpoints
+    """
+
+    def __init__(self, app: ASGIApp, auth_service: AuthService):
+        """Initialize middleware with auth service.
+
+        Args:
+            app: ASGI application
+            auth_service: AuthService instance for checking admin existence
+        """
+        super().__init__(app)
+        self.auth_service = auth_service
+
+    async def dispatch(self, request: Request, call_next: Callable) -> Response:
+        """Process request and redirect to setup if needed.
+
+        Args:
+            request: Incoming HTTP request
+            call_next: Next middleware/endpoint in chain
+
+        Returns:
+            Response from next handler or redirect to setup
+        """
+        # Paths that should not trigger setup redirect
+        exempt_paths = [
+            "/admin/setup",
+            "/admin/login",
+            "/static/",
+            "/api/health",
+        ]
+
+        # Allow exempt paths through
+        if any(request.url.path.startswith(path) for path in exempt_paths):
+            return await call_next(request)
+
+        # Check if admin user exists
+        if not self.auth_service.admin_exists():
+            return RedirectResponse(url="/admin/setup", status_code=303)
+
+        # Admin exists, continue normally
+        return await call_next(request)