import asyncio import logging import time import traceback as tb from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone import docker.errors import httpx from fastapi import Depends, FastAPI, Request from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles from slowapi import Limiter from slowapi.errors import RateLimitExceeded from slowapi.util import get_remote_address import auth_service as auth_module import config from config import CORS_ORIGINS, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, VOLUME_ROOT, get_client_ip from rbac import require_page from routers import auth as auth_router from routers import ( cc_connect, chat_summary, conversation_tracker, docker_router, files, gitea, info_engine, litellm, opc_agents, opc_projects, opc_tasks, opc_ws, passkey, security, system, terminal, totp, transmission, ) _tz_cst = timezone(timedelta(hours=8)) @asynccontextmanager async def lifespan(app: FastAPI): # Startup print("=== Application Startup ===") # Initialize OPC database from db import opc_db try: await opc_db.init_db() print("OPC database initialized") except Exception as e: print(f"Failed to initialize OPC database: {e}") import traceback traceback.print_exc() # Start agent executor service from services import agent_executor executor_task = None try: executor_task = asyncio.create_task(agent_executor.start_executor()) print("Agent executor service started") except Exception as e: print(f"Failed to start agent executor: {e}") import traceback traceback.print_exc() monitor_task = asyncio.create_task(monitor_containers()) yield # Shutdown print("=== Application Shutdown ===") if executor_task: executor_task.cancel() monitor_task.cancel() limiter = Limiter(key_func=get_remote_address) app = FastAPI(title="NAS Dashboard", lifespan=lifespan) app.state.limiter = limiter app.add_middleware( CORSMiddleware, allow_origins=CORS_ORIGINS, allow_credentials=True, allow_methods=["GET", "POST", "DELETE", "OPTIONS"], allow_headers=["Authorization", "Content-Type"], ) # Compress responses (static files and API responses) app.add_middleware(GZipMiddleware, minimum_size=1000) @app.exception_handler(RateLimitExceeded) async def rate_limit_handler(request: Request, exc: RateLimitExceeded): return JSONResponse(status_code=429, content={"detail": "Too many requests, try again later"}) @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): """Handle request validation errors with detailed information.""" logging.error(f"Validation error on {request.method} {request.url.path}: {exc.errors()}") return JSONResponse(status_code=422, content={"detail": exc.errors()}) @app.exception_handler(docker.errors.NotFound) async def docker_not_found_handler(request: Request, exc: docker.errors.NotFound): """Handle Docker container/image not found errors.""" logging.warning(f"Docker resource not found: {exc}") return JSONResponse(status_code=404, content={"detail": "Docker resource not found"}) @app.exception_handler(docker.errors.APIError) async def docker_api_error_handler(request: Request, exc: docker.errors.APIError): """Handle Docker API errors.""" logging.error(f"Docker API error: {exc}") return JSONResponse(status_code=503, content={"detail": "Docker service unavailable"}) @app.exception_handler(Exception) async def generic_exception_handler(request: Request, exc: Exception): """Handle all unhandled exceptions.""" logging.error(f"Unhandled exception on {request.method} {request.url.path}: {exc}") logging.error(tb.format_exc()) return JSONResponse(status_code=500, content={"detail": "Internal server error"}) @app.middleware("http") async def security_headers(request: Request, call_next): response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["X-XSS-Protection"] = "1; mode=block" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" response.headers["Content-Security-Policy"] = ( "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self' wss: ws:; frame-ancestors 'none'" ) # Cache static assets with hashed filenames (Vite generates these) if request.url.path.startswith("/assets/"): response.headers["Cache-Control"] = "public, max-age=31536000, immutable" return response # Audit logger import os _audit_log_path = os.path.join(VOLUME_ROOT, "docker/nas-dashboard/audit.log") os.makedirs(os.path.dirname(_audit_log_path), exist_ok=True) _audit_logger = logging.getLogger("audit") _audit_logger.setLevel(logging.INFO) _audit_handler = logging.FileHandler(_audit_log_path) _audit_handler.setFormatter(logging.Formatter("%(message)s")) _audit_logger.addHandler(_audit_handler) _audit_logger.propagate = False @app.middleware("http") async def audit_log(request: Request, call_next): start = time.time() response = await call_next(request) duration = int((time.time() - start) * 1000) user = getattr(request.state, "user", None) username = user.username if user else "-" _audit_logger.info( "%s %s %s %s %s %d %dms", datetime.now(_tz_cst).strftime("%Y-%m-%dT%H:%M:%S"), get_client_ip(request), username, request.method, request.url.path, response.status_code, duration, ) return response async def monitor_containers(): import docker from config import DOCKER_HOST client = docker.DockerClient(base_url=DOCKER_HOST) # Store initial states previous_states = {} while True: try: containers = await asyncio.to_thread(client.containers.list, all=True) for c in containers: current_status = c.status if c.id in previous_states: prev_status = previous_states[c.id] # Alert if it transitioned from running to exited or unhealthy if prev_status == "running" and current_status in ("exited", "dead"): msg = f"⚠️ *Container Alert* ⚠️\nContainer `{c.name}` just stopped unexpectedly (Status: {current_status})." # Use httpx directly as system.send_notification expects a FastAPI request object usually if TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID: proxy_url = os.environ.get("TELEGRAM_PROXY") async with httpx.AsyncClient(proxy=proxy_url) as hc: await hc.post( f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage", data={"chat_id": TELEGRAM_CHAT_ID, "text": msg, "parse_mode": "Markdown"}, ) previous_states[c.id] = current_status except Exception as e: logging.getLogger(__name__).error("Error in container monitor: %s", e) await asyncio.sleep(60) @app.get("/api/health") async def health(): return {"status": "ok"} @app.get("/api/client-ip") async def client_ip(request: Request): ip = get_client_ip(request) # LAN detection logic: # - True LAN IP (192.168.31.x) → lan=True # - Tailscale IP with router reachable → lan=True (user at home with Tailscale) # - Tailscale IP without router reachable → lan=False (remote via VPS proxy) # - Other IPs → lan=False (public internet) if config.is_lan_ip(ip): lan = True elif config.is_tailscale_ip(ip): lan = _router_reachable() else: lan = False return {"lan": lan} def _router_reachable(): import socket try: s = socket.socket() s.settimeout(1) s.connect(("192.168.31.1", 80)) s.close() return True except Exception: return False # Dependency to store user in request.state for rbac dependencies async def _inject_user(request: Request, user=Depends(auth_module.get_current_user)): request.state.user = user return user # Public auth endpoints app.include_router(auth_router.router, prefix="/api/auth", tags=["auth"]) app.include_router(passkey.router, prefix="/api/auth", tags=["passkey"]) app.include_router(totp.router, prefix="/api/auth", tags=["totp"]) # Protected endpoints with page-level RBAC app.include_router( docker_router.router, prefix="/api/docker", dependencies=[Depends(_inject_user), Depends(require_page("docker"))] ) app.include_router( gitea.router, prefix="/api/gitea", dependencies=[Depends(_inject_user), Depends(require_page("gitea"))] ) app.include_router( files.router, prefix="/api/files", dependencies=[Depends(_inject_user), Depends(require_page("files"))] ) app.include_router( system.router, prefix="/api/system", dependencies=[Depends(_inject_user), Depends(require_page("dashboard"))] ) app.include_router( litellm.router, prefix="/api/litellm", dependencies=[Depends(_inject_user), Depends(require_page("dashboard"))] ) app.include_router( cc_connect.router, prefix="/api/cc-connect", dependencies=[Depends(_inject_user), Depends(require_page("dashboard"))], ) app.include_router( chat_summary.router, prefix="/api/chat-summary", dependencies=[Depends(_inject_user), Depends(require_page("chat-digest"))], ) app.include_router( conversation_tracker.router, prefix="/api/conversations", dependencies=[Depends(_inject_user), Depends(require_page("conversations"))], ) app.include_router( info_engine.router, prefix="/api/info-engine", dependencies=[Depends(_inject_user), Depends(require_page("dashboard"))], ) app.include_router( security.router, prefix="/api/security", dependencies=[Depends(_inject_user), Depends(require_page("security"))] ) app.include_router( transmission.router, dependencies=[Depends(_inject_user), Depends(require_page("transmission"))] ) # OPC endpoints app.include_router( opc_tasks.router, prefix="/api/opc", dependencies=[Depends(_inject_user), Depends(require_page("opc"))] ) app.include_router( opc_agents.router, prefix="/api/opc", dependencies=[Depends(_inject_user), Depends(require_page("opc"))] ) app.include_router( opc_projects.router, prefix="/api/opc", dependencies=[Depends(_inject_user), Depends(require_page("opc"))] ) # WebSocket endpoint (auth handled inside handler via auth cookie; query token fallback temporary) app.add_api_websocket_route("/ws/terminal", terminal.ws_endpoint) app.add_api_websocket_route("/ws/opc", opc_ws.websocket_endpoint) # Mount static files (skip if directory doesn't exist, e.g., in test environments) import os if os.path.isdir("/app/static"): app.mount("/", StaticFiles(directory="/app/static", html=True), name="static") else: # In test environment, create a minimal fallback import tempfile _test_static = tempfile.mkdtemp() with open(os.path.join(_test_static, "index.html"), "w") as f: f.write("Test Environment") app.mount("/", StaticFiles(directory=_test_static, html=True), name="static")