Add a stats agent container: FastAPI + psutil behind an API key

Exposes host CPU, memory, disk, and network usage over HTTP for the
dashboard's System Stats block (or anything else that can send a header)
to poll. Auth is a required X-API-Key, auto-generated and persisted on
first run if not supplied, checked with a constant-time comparison and
rate-limited after repeated failures. No CORS, so a browser can't call it
directly from another origin -- only a server-side caller can, keeping
the key out of anyone's network tab.
This commit is contained in:
2026-07-29 20:20:48 +02:00
commit ef6b313d47
10 changed files with 492 additions and 0 deletions

76
app/security.py Normal file
View File

@@ -0,0 +1,76 @@
"""API key auth for the stats endpoint.
The key is either supplied via the API_KEY env var or, if that's unset,
generated once on first run and persisted to AGENT_DATA_DIR so it survives
container restarts without the operator having to pick one themselves.
"""
import os
import secrets
import time
from pathlib import Path
from fastapi import Header, HTTPException, Request
DATA_DIR = Path(os.environ.get("AGENT_DATA_DIR", "/data"))
KEY_FILE = DATA_DIR / "api_key.txt"
# Failed-attempt tracking is per-process, in-memory, and intentionally
# unbounded-but-small: a homelab agent restarts often enough (image
# updates, host reboots) that this never grows large in practice, and
# doesn't need a database to survive a restart.
_WINDOW_SECONDS = 60
_MAX_FAILURES = 10
_failures: dict[str, list[float]] = {}
def _generate_key() -> str:
return secrets.token_urlsafe(32)
def load_or_create_api_key() -> str:
env_key = os.environ.get("API_KEY", "").strip()
if env_key:
return env_key
DATA_DIR.mkdir(parents=True, exist_ok=True)
if KEY_FILE.exists():
existing = KEY_FILE.read_text().strip()
if existing:
return existing
key = _generate_key()
KEY_FILE.write_text(key + "\n")
KEY_FILE.chmod(0o600)
print("=" * 72)
print("No API_KEY set — generated a new one and saved it to")
print(f" {KEY_FILE}")
print(f"API key: {key}")
print("Set API_KEY explicitly instead if you want to choose your own,")
print("or reuse this one from the mounted data volume on next start.")
print("=" * 72)
return key
API_KEY = load_or_create_api_key()
def _client_ip(request: Request) -> str:
return request.client.host if request.client else "unknown"
def _recent_failures(ip: str, now: float) -> list[float]:
attempts = [t for t in _failures.get(ip, []) if now - t < _WINDOW_SECONDS]
_failures[ip] = attempts
return attempts
def require_api_key(request: Request, x_api_key: str | None = Header(default=None, alias="X-API-Key")) -> None:
now = time.time()
ip = _client_ip(request)
if len(_recent_failures(ip, now)) >= _MAX_FAILURES:
raise HTTPException(status_code=429, detail="Too many failed attempts, try again later")
if not x_api_key or not secrets.compare_digest(x_api_key, API_KEY):
_failures.setdefault(ip, []).append(now)
raise HTTPException(status_code=401, detail="Invalid or missing API key")