Gan, Jimmy
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
import os
|
|
import shutil
|
|
import psutil
|
|
import platform
|
|
import time
|
|
import httpx
|
|
from fastapi import APIRouter, Query
|
|
from config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, VOLUME_ROOT
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/notify")
|
|
async def send_notification(body: dict):
|
|
"""Send a Telegram notification. Body: {"message": "text"}"""
|
|
msg = body.get("message", "")
|
|
if not msg or not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
|
|
return {"ok": False, "error": "missing message or Telegram config"}
|
|
proxy_url = os.environ.get("TELEGRAM_PROXY")
|
|
async with httpx.AsyncClient(proxy=proxy_url) as client:
|
|
r = await client.post(
|
|
f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
|
|
data={"chat_id": TELEGRAM_CHAT_ID, "text": msg, "parse_mode": "Markdown"},
|
|
)
|
|
return {"ok": r.status_code == 200}
|
|
|
|
|
|
@router.get("/stats")
|
|
def system_stats():
|
|
disk = shutil.disk_usage("/volume1")
|
|
volumes = []
|
|
for mount in psutil.disk_partitions():
|
|
if mount.mountpoint.startswith(("/volume",)):
|
|
try:
|
|
u = shutil.disk_usage(mount.mountpoint)
|
|
volumes.append({"mount": mount.mountpoint, "total": u.total, "used": u.used, "free": u.free, "percent": round(u.used / u.total * 100, 1)})
|
|
except Exception:
|
|
pass
|
|
mem = psutil.virtual_memory()
|
|
cpu_pct = psutil.cpu_percent(interval=0.5)
|
|
load_1, load_5, load_15 = psutil.getloadavg()
|
|
uptime_s = time.time() - psutil.boot_time()
|
|
|
|
days = int(uptime_s // 86400)
|
|
hours = int((uptime_s % 86400) // 3600)
|
|
|
|
return {
|
|
"cpu_percent": cpu_pct,
|
|
"cpu_count": psutil.cpu_count(),
|
|
"load_avg": [round(load_1, 2), round(load_5, 2), round(load_15, 2)],
|
|
"memory": {
|
|
"total": mem.total,
|
|
"used": mem.used,
|
|
"percent": mem.percent,
|
|
},
|
|
"disk": {
|
|
"total": disk.total,
|
|
"used": disk.used,
|
|
"free": disk.free,
|
|
"percent": round(disk.used / disk.total * 100, 1),
|
|
},
|
|
"uptime": f"{days}d {hours}h",
|
|
"platform": platform.platform(),
|
|
"volumes": volumes,
|
|
}
|
|
|
|
|
|
@router.get("/audit-log")
|
|
def audit_log(lines: int = Query(100, le=500)):
|
|
log_path = os.path.join(VOLUME_ROOT, "docker/nas-dashboard/audit.log")
|
|
if not os.path.exists(log_path):
|
|
return {"entries": []}
|
|
with open(log_path, "rb") as f:
|
|
f.seek(0, 2)
|
|
size = f.tell()
|
|
buf = min(size, lines * 200)
|
|
f.seek(max(0, size - buf))
|
|
raw = f.read().decode(errors="replace").splitlines()[-lines:]
|
|
|
|
entries = []
|
|
for line in raw:
|
|
parts = line.split(" ", 6)
|
|
if len(parts) < 7:
|
|
continue
|
|
ts, ip, user, method, path, status_str, dur = parts
|
|
status = int(status_str) if status_str.isdigit() else 0
|
|
level = _classify(method, path, status, user)
|
|
if level == "noise":
|
|
continue
|
|
entries.append({"ts": ts, "ip": ip, "user": user, "method": method, "path": path, "status": status, "duration": dur, "level": level})
|
|
return {"entries": entries}
|
|
|
|
|
|
def _classify(method, path, status, user):
|
|
# High — failed auth attempts
|
|
if "/auth/login" in path and status in (401, 403):
|
|
return "high"
|
|
if "/auth/passkey/login/verify" in path and status == 400:
|
|
return "high"
|
|
# Medium — 403/404 on sensitive paths, or any 5xx
|
|
if status >= 500:
|
|
return "medium"
|
|
if status == 403 and "/files/" in path:
|
|
return "medium"
|
|
# Low — normal auth activity
|
|
if "/auth/" in path and status == 200:
|
|
return "low"
|
|
# Noise — health checks, static assets
|
|
if path in ("/api/health",) or path.startswith("/assets/") or path == "/favicon.ico" or path == "/":
|
|
return "noise"
|
|
# Info — everything else
|
|
return "info"
|