nas-tools/dashboard/backend/routers/system.py

import os
import platform
import shutil
import time

import httpx
import psutil
from fastapi import APIRouter, Depends, HTTPException, Query, Request

import auth_service as auth
import config
from config import OPENCLAW_GATEWAY_TOKEN, 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 = []
    seen_devices = set()
    for mount in psutil.disk_partitions():
        if mount.mountpoint.startswith(("/volume",)) and mount.device not in seen_devices:
            seen_devices.add(mount.device)
            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.1)
    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), current_user=Depends(auth.get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    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"


@router.get("/openclaw-token")
def openclaw_token(request: Request):
    user = request.state.user
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")

    ip = config.get_client_ip(request)
    if not config.is_lan_ip(ip):
        raise HTTPException(status_code=403, detail="Access denied")
    return {"token": OPENCLAW_GATEWAY_TOKEN}


@router.get("/external-services")
def external_services():
    """Return external service URLs and network config for the frontend sidebar."""
    return {
        "lan_ip": config.LAN_IP,
        "ts_ip": config.TS_IP,
        "services": config.EXTERNAL_SERVICES,
    }