Minz
ef6b313d47
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.
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
"""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")
|