homelab-optimized/dashboard/api/routers/network.py

"""Network / Headscale / AdGuard routes."""

from fastapi import APIRouter
import subprocess
import json
import httpx

router = APIRouter(tags=["network"])

CLOUDFLARE_TOKEN = "REDACTED_TOKEN"  # pragma: allowlist secret
CLOUDFLARE_ZONE_ID = "4dbd15d096d71101b7c0c6362b307a66"
AUTHENTIK_URL = "https://sso.vish.gg"
AUTHENTIK_TOKEN = "REDACTED_TOKEN"  # pragma: allowlist secret
GITEA_URL = "https://git.vish.gg"
GITEA_TOKEN = "REDACTED_TOKEN"  # pragma: allowlist secret

ADGUARD_URL = "http://192.168.0.250:9080"
ADGUARD_USER = "vish"
ADGUARD_PASS = "REDACTED_PASSWORD"


def _adguard_get(path):
    with httpx.Client(timeout=10) as client:
        client.post(f"{ADGUARD_URL}/control/login", json={"name": ADGUARD_USER, "password": ADGUARD_PASS})
        r = client.get(f"{ADGUARD_URL}/control{path}")
        r.raise_for_status()
        return r.json() if r.content else {}


@router.get("/network/headscale")
def headscale_nodes():
    """List Headscale nodes."""
    result = subprocess.run(
        ["ssh", "-o", "ConnectTimeout=3", "calypso",
         "docker exec headscale headscale nodes list -o json"],
        capture_output=True, text=True, timeout=15,
    )
    if result.returncode != 0:
        return {"nodes": [], "error": result.stderr.strip()}
    nodes = json.loads(result.stdout)
    return {"nodes": [
        {"name": n.get("givenName") or n.get("name", "?"),
         "ip": (n.get("ipAddresses") or ["?"])[0],
         "online": n.get("online", False),
         "last_seen": n.get("lastSeen", "")}
        for n in nodes
    ]}


@router.get("/network/adguard")
def adguard_stats():
    """Get AdGuard DNS stats."""
    try:
        stats = _adguard_get("/stats")
        return {
            "total_queries": stats.get("num_dns_queries", 0),
            "blocked": stats.get("num_blocked_filtering", 0),
            "avg_time": stats.get("avg_processing_time", 0),
        }
    except Exception as e:
        return {"error": str(e)}


@router.get("/network/adguard/rewrites")
def adguard_rewrites():
    """List AdGuard DNS rewrites."""
    try:
        data = _adguard_get("/rewrite/list")
        return [{"domain": r.get("domain", ""), "answer": r.get("answer", "")} for r in (data or [])]
    except Exception as e:
        return {"error": str(e)}


@router.get("/network/cloudflare")
def cloudflare_stats():
    """Cloudflare DNS record summary."""
    try:
        with httpx.Client(timeout=10) as client:
            r = client.get(f"https://api.cloudflare.com/client/v4/zones/{CLOUDFLARE_ZONE_ID}/dns_records",
                           headers={"Authorization": f"Bearer {CLOUDFLARE_TOKEN}"},
                           params={"per_page": 100})
            r.raise_for_status()
            records = r.json().get("result", [])
            proxied = sum(1 for rec in records if rec.get("proxied"))
            types = {}
            for rec in records:
                t = rec.get("type", "?")
                types[t] = types.get(t, 0) + 1
            return {"total": len(records), "proxied": proxied, "dns_only": len(records) - proxied, "types": types}
    except Exception as e:
        return {"error": str(e)}


@router.get("/network/authentik")
def authentik_sessions():
    """Authentik active sessions and recent events."""
    try:
        with httpx.Client(timeout=10, verify=False) as client:
            # Sessions
            sr = client.get(f"{AUTHENTIK_URL}/api/v3/core/authenticated_sessions/",
                            headers={"Authorization": f"Bearer {AUTHENTIK_TOKEN}"})
            sessions = sr.json().get("results", []) if sr.status_code == 200 else []
            # Recent events
            er = client.get(f"{AUTHENTIK_URL}/api/v3/events/events/",
                            headers={"Authorization": f"Bearer {AUTHENTIK_TOKEN}"},
                            params={"page_size": 10, "ordering": "-created"})
            events = er.json().get("results", []) if er.status_code == 200 else []
            return {
                "active_sessions": len(sessions),
                "sessions": [{"user": s.get("user", {}).get("username", "?"),
                              "last_used": s.get("last_used_at", "?")} for s in sessions[:10]],
                "recent_events": [{"action": e.get("action", "?"), "user": e.get("user", {}).get("username", "?"),
                                   "created": e.get("created", "?")[:19]} for e in events[:5]],
            }
    except Exception as e:
        return {"error": str(e)}


@router.get("/network/gitea")
def gitea_activity():
    """Recent Gitea commits and open PRs."""
    try:
        with httpx.Client(timeout=10) as client:
            # Recent commits
            cr = client.get(f"{GITEA_URL}/api/v1/repos/vish/homelab/commits",
                            headers={"Authorization": f"token {GITEA_TOKEN}"},
                            params={"limit": 5, "sha": "main"})
            commits = []
            if cr.status_code == 200:
                for c in cr.json()[:5]:
                    commits.append({
                        "sha": c.get("sha", "?")[:7],
                        "message": c.get("commit", {}).get("message", "?").split("\n")[0][:80],
                        "date": c.get("commit", {}).get("committer", {}).get("date", "?")[:10],
                        "author": c.get("commit", {}).get("author", {}).get("name", "?"),
                    })

            # Open PRs
            pr = client.get(f"{GITEA_URL}/api/v1/repos/vish/homelab/pulls",
                            headers={"Authorization": f"token {GITEA_TOKEN}"},
                            params={"state": "open", "limit": 5})
            prs = []
            if pr.status_code == 200:
                for p in pr.json():
                    prs.append({
                        "number": p.get("number"),
                        "title": p.get("title", "?"),
                        "user": p.get("user", {}).get("login", "?"),
                    })

            return {"commits": commits, "open_prs": prs}
    except Exception as e:
        return {"error": str(e)}