Gan, Jimmy
319 lines
11 KiB
Python
319 lines
11 KiB
Python
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
|
|
_tz_cst = timezone(timedelta(hours=8))
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
# Startup
|
|
print("=== Application Startup ===")
|
|
|
|
# Warn if cookies are not marked Secure (must be behind TLS-terminating proxy)
|
|
import auth_service as auth_svc
|
|
|
|
if not auth_svc.COOKIE_SECURE and not os.environ.get("ALLOW_INSECURE_COOKIES"):
|
|
print("WARNING: COOKIE_SECURE=False — auth cookies not marked Secure. "
|
|
"This requires a TLS-terminating reverse proxy (e.g. Caddy) in front. "
|
|
"Set ALLOW_INSECURE_COOKIES=true to suppress this warning.")
|
|
|
|
# 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
|
|
import time
|
|
import logging
|
|
|
|
from config import DOCKER_HOST
|
|
|
|
_monitor_logger = logging.getLogger(__name__)
|
|
|
|
def _docker_retry(fn, max_retries=3, base_delay=1):
|
|
for attempt in range(max_retries):
|
|
try:
|
|
return fn()
|
|
except Exception as e:
|
|
delay = base_delay * (2 ** attempt)
|
|
_monitor_logger.warning("Docker monitor error (attempt %d/%d): %s. Retrying in %ds...", attempt+1, max_retries, e, delay)
|
|
if attempt < max_retries - 1:
|
|
time.sleep(delay)
|
|
_monitor_logger.error("Docker monitor failed after %d attempts", max_retries)
|
|
return None
|
|
|
|
client = docker.DockerClient(base_url=DOCKER_HOST)
|
|
|
|
# Store initial states
|
|
previous_states = {}
|
|
|
|
while True:
|
|
try:
|
|
containers = await asyncio.to_thread(
|
|
lambda: docker_retry(lambda: client.containers.list(all=True)) or []
|
|
)
|
|
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
|
|
|
|
|
|
# Auto-discovered router registration (from ROUTER_CONFIGS manifest)
|
|
from routers import ROUTER_CONFIGS, WS_ROUTES
|
|
import importlib
|
|
import re
|
|
|
|
for cfg in ROUTER_CONFIGS:
|
|
mod = importlib.import_module('routers.' + cfg['module'])
|
|
router_attr = cfg.get('router_attr', 'router')
|
|
router_obj = getattr(mod, router_attr)
|
|
prefix = cfg.get('prefix')
|
|
tags = cfg.get('tags')
|
|
deps = cfg.get('deps', [])
|
|
dep_objects = []
|
|
for d in deps:
|
|
if d == '_inject_user':
|
|
dep_objects.append(Depends(_inject_user))
|
|
elif d.startswith('require_page('):
|
|
# Extract page name from require_page('gitea') or require_page(gitea)
|
|
m = re.search(r'require_page\(["\'](.*?)["\']\)', d)
|
|
page_name = m.group(1) if m else d.split('(')[1].rstrip(')')
|
|
dep_objects.append(Depends(require_page(page_name)))
|
|
kwargs = {'dependencies': dep_objects} if dep_objects else {}
|
|
if prefix is not None:
|
|
kwargs['prefix'] = prefix
|
|
if tags:
|
|
kwargs['tags'] = tags
|
|
app.include_router(router_obj, **kwargs)
|
|
|
|
for ws in WS_ROUTES:
|
|
mod = importlib.import_module('routers.' + ws['module'])
|
|
handler = getattr(mod, ws['handler'])
|
|
app.add_api_websocket_route(ws['path'], handler)
|
|
|
|
# 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("<html><body>Test Environment</body></html>")
|
|
app.mount("/", StaticFiles(directory=_test_static, html=True), name="static")
|