"""Hermes Command Center - FastAPI backend with Network Status Dashboard."""
import asyncio
import base64
import os
import secrets
import ssl
import subprocess
import time
from pathlib import Path
import yaml
from fastapi import FastAPI, Request, Response, Query
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, JSONResponse
# ── Auth ──────────────────────────────────────────────────────────────
DASHBOARD_USER = os.environ.get("DASHBOARD_USER", "shawn")
DASHBOARD_PASS = os.environ.get("DASHBOARD_PASS", "Brett85!@")
def _check_auth(auth_header: str | None) -> bool:
if not auth_header:
return False
try:
decoded = base64.b64decode(auth_header.removeprefix("Basic ")).decode()
user, _, passwd = decoded.partition(":")
return secrets.compare_digest(user, DASHBOARD_USER) and secrets.compare_digest(passwd, DASHBOARD_PASS)
except Exception:
return False
_unauthorized = Response(
content="401 Unauthorized
",
status_code=401,
headers={"WWW-Authenticate": 'Basic realm="Command Center"'},
)
app = FastAPI(title="Hermes Command Center")
STATIC_DIR = Path(__file__).parent / "static"
OBSIDIAN_VAULT = Path(os.path.expanduser("~/obsidian")).resolve()
# ── Network Health Cache ──────────────────────────────────────────────
INVENTORY_PATH = os.path.expanduser("~/.hermes/network-inventory.yaml")
_network_cache: dict = {}
CACHE_TTL = 30 # seconds
def _load_inventory() -> dict:
"""Load network inventory from YAML."""
path = Path(INVENTORY_PATH)
if not path.exists():
return {"hosts": [], "services": []}
with open(path) as f:
return yaml.safe_load(f) or {}
def _get_host_ip(host: dict) -> str | None:
"""Return the best IP to check for a given host definition."""
if host.get("local_ip"):
return host["local_ip"]
if host.get("tailscale_ip"):
return host["tailscale_ip"]
return None
async def _ping_host(ip: str, timeout: int = 2) -> float | None:
"""Ping a host and return latency in ms, or None if unreachable."""
try:
start = time.monotonic()
proc = await asyncio.create_subprocess_exec(
"ping", "-c", "1", "-W", str(timeout), ip,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await proc.wait()
elapsed = (time.monotonic() - start) * 1000
if proc.returncode == 0:
return round(elapsed, 1)
return None
except Exception:
return None
async def _check_port(ip: str, port: int, timeout: int = 2) -> bool:
"""Check if a TCP port is open on the given host."""
try:
_, writer = await asyncio.wait_for(
asyncio.open_connection(ip, port),
timeout=timeout,
)
writer.close()
await writer.wait_closed()
return True
except (OSError, asyncio.TimeoutError):
return False
async def _check_http(url: str, timeout: int = 3) -> dict:
"""Check an HTTP(S) endpoint and return status info."""
import urllib.request
try:
req = urllib.request.Request(url, method="HEAD" if url.startswith("http") else "GET")
start = time.monotonic()
ctx = ssl._create_unverified_context()
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
elapsed = (time.monotonic() - start) * 1000
return {"status": "online", "http_status": resp.status, "latency_ms": round(elapsed, 1)}
except Exception as e:
return {"status": "offline", "error": str(e)[:80]}
def _get_local_services() -> list[dict]:
"""Parse ss -tlnp output for locally listening services."""
try:
result = subprocess.run(["ss", "-tlnp"], capture_output=True, text=True, timeout=5)
services = []
lines = result.stdout.strip().split("\n")
for line in lines[1:]:
if not line.strip():
continue
parts = line.split()
if len(parts) < 4:
continue
local = parts[3]
addr, port_raw = local.rsplit(":", 1)
proc = parts[-1] if len(parts) > 5 else ""
addr = addr.strip("[]")
services.append({
"local_address": addr,
"port": port_raw,
"process": proc,
})
return services
except (subprocess.TimeoutExpired, FileNotFoundError, IndexError) as e:
return [{"error": str(e)}]
async def _check_host(host: dict) -> dict:
"""Perform health checks on a single host and return enriched status."""
ip = _get_host_ip(host)
if not ip:
safe_host = {k: v for k, v in host.items() if k != "notes"}
return {**safe_host, "ip": None, "status": "unknown", "reason": "no IP configured", "latency_ms": None, "ports": []}
latency = await _ping_host(ip)
known_ports = []
if host.get("ssh_port"):
known_ports.append(("SSH", int(host["ssh_port"])))
checked_ports = []
for name, port in known_ports:
open_port = await _check_port(ip, port)
checked_ports.append({"name": name, "port": port, "open": open_port})
if latency is not None:
safe_host = {k: v for k, v in host.items() if k != "notes"}
return {**safe_host, "ip": ip, "status": "online", "latency_ms": latency, "ports": checked_ports}
else:
safe_host = {k: v for k, v in host.items() if k != "notes"}
return {**safe_host, "ip": ip, "status": "offline", "latency_ms": None, "ports": checked_ports}
async def _check_service(service: dict) -> dict:
"""Perform health check on a single service definition."""
result = {k: v for k, v in service.items() if k not in ("notes",)}
result.setdefault("status", "unknown")
result["latency_ms"] = None
result["http_status"] = None
# Try HTTP/S URL first
if service.get("url"):
check = await _check_http(service["url"])
result["status"] = check["status"]
result["latency_ms"] = check.get("latency_ms")
result["http_status"] = check.get("http_status")
result["error"] = check.get("error")
return result
# Try port check on 127.0.0.1 (services defined as running locally)
port = service.get("port")
if port and service.get("category") != "infrastructure":
is_open = await _check_port("127.0.0.1", int(port))
if is_open:
result["status"] = "online"
result["latency_ms"] = 0.0
else:
result["status"] = "offline"
return result
async def _run_health_checks() -> dict:
"""Run all health checks and return combined results with caching."""
global _network_cache
now = time.monotonic()
if _network_cache and (now - _network_cache.get("_ts", 0)) < CACHE_TTL:
return _network_cache
inventory = _load_inventory()
hosts = inventory.get("hosts", [])
services = inventory.get("services", [])
host_tasks = [_check_host(h) for h in hosts]
host_results = await asyncio.gather(*host_tasks)
service_tasks = [_check_service(s) for s in services]
service_results = await asyncio.gather(*service_tasks)
local_services = _get_local_services()
result = {
"hosts": host_results,
"services": service_results,
"local_services": local_services,
"cached_at": time.time(),
"cache_ttl": CACHE_TTL,
}
_network_cache = result
_network_cache["_ts"] = now
return result
# ── Path security ─────────────────────────────────────────────────────
def _resolve_vault_path(user_path: str) -> Path | None:
"""Resolve a user-supplied path relative to the vault root.
Returns the resolved absolute Path if it's under OBSIDIAN_VAULT,
or None if the path is invalid/outside the vault.
"""
safe = user_path.lstrip("/")
candidate = (OBSIDIAN_VAULT / safe).resolve()
try:
candidate.relative_to(OBSIDIAN_VAULT)
except ValueError:
return None
if not candidate.exists():
return None
return candidate
# ── Auth middleware ───────────────────────────────────────────────────
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
if request.url.path in ("/health", "/api/health"):
return await call_next(request)
if request.url.path.startswith("/api/network/"):
return await call_next(request)
if not _check_auth(request.headers.get("Authorization")):
return _unauthorized
return await call_next(request)
@app.get("/health")
async def health():
return {"status": "ok", "service": "hermes-command-center"}
# ── Network API Endpoints ────────────────────────────────────────────
@app.get("/api/network/hosts")
async def network_hosts():
"""List all hosts from inventory with live status."""
data = await _run_health_checks()
return {"hosts": data["hosts"], "cached_at": data["cached_at"], "cache_ttl": data["cache_ttl"]}
@app.get("/api/network/services")
async def network_services():
"""List all services with live health check results."""
data = await _run_health_checks()
return {"services": data["services"], "cached_at": data["cached_at"], "cache_ttl": data["cache_ttl"]}
@app.get("/api/network/services/local")
async def network_services_local():
"""List locally running services (from ss -tlnp)."""
return {"local_services": _get_local_services()}
@app.get("/api/network/refresh")
async def network_refresh():
"""Force a fresh health check of all hosts/services."""
global _network_cache
_network_cache = {}
data = await _run_health_checks()
return {
"hosts": data["hosts"],
"services": data["services"],
"local_services": data["local_services"],
"cached_at": data["cached_at"],
"cache_ttl": data["cache_ttl"],
}
# ── Vault API ─────────────────────────────────────────────────────────
@app.get("/api/vault/tree")
async def vault_tree(path: str = Query("", description="Relative path under vault")):
resolved = _resolve_vault_path(path)
if resolved is None:
return JSONResponse({"error": "Path not found or outside vault"}, status_code=404)
if not resolved.is_dir():
return JSONResponse({"error": "Not a directory"}, status_code=400)
items = []
try:
for entry in sorted(resolved.iterdir(), key=lambda e: (not e.is_dir(), e.name.lower())):
if entry.name.startswith("."):
continue
info = {
"name": entry.name,
"type": "folder" if entry.is_dir() else "file",
"size": entry.stat().st_size if entry.is_file() else 0,
"modified": int(entry.stat().st_mtime),
}
items.append(info)
except PermissionError:
return JSONResponse({"error": "Permission denied"}, status_code=403)
rel = resolved.relative_to(OBSIDIAN_VAULT)
return JSONResponse({
"path": str(rel) if str(rel) != "." else "",
"items": items,
})
@app.get("/api/vault/read")
async def vault_read(path: str = Query(..., description="Relative path to markdown file")):
resolved = _resolve_vault_path(path)
if resolved is None:
return JSONResponse({"error": "File not found or outside vault"}, status_code=404)
if not resolved.is_file():
return JSONResponse({"error": "Not a file"}, status_code=400)
if resolved.suffix.lower() not in (".md", ".txt", ".markdown"):
return JSONResponse({"error": "Cannot read non-text files"}, status_code=400)
try:
content = resolved.read_text(encoding="utf-8")
except Exception as e:
return JSONResponse({"error": f"Failed to read file: {e}"}, status_code=500)
rel = resolved.relative_to(OBSIDIAN_VAULT)
return JSONResponse({
"path": str(rel),
"name": resolved.name,
"content": content,
"size": resolved.stat().st_size,
"modified": int(resolved.stat().st_mtime),
})
@app.get("/api/vault/search")
async def vault_search(q: str = Query(..., min_length=1, description="Search query")):
"""Full-text grep-style search across all .md files under the vault root."""
results = []
try:
cmd = [
"grep", "-rn",
"--include=*.md",
"-i",
"-E", q,
str(OBSIDIAN_VAULT),
]
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
if proc.returncode not in (0, 1):
return JSONResponse({"error": f"Search failed: {proc.stderr.strip()}"}, status_code=500)
for line in proc.stdout.splitlines():
parts = line.split(":", 2)
if len(parts) < 3:
continue
filepath, lineno, content = parts
fp = Path(filepath).resolve()
try:
fp.relative_to(OBSIDIAN_VAULT)
except ValueError:
continue
rel_path = str(fp.relative_to(OBSIDIAN_VAULT))
idx = content.lower().find(q.lower())
start = max(0, idx - 60)
end = min(len(content), idx + len(q) + 60)
snippet = content[start:end].strip()
if start > 0:
snippet = "..." + snippet
if end < len(content):
snippet = snippet + "..."
results.append({
"path": rel_path,
"line": int(lineno),
"snippet": snippet,
})
except subprocess.TimeoutExpired:
return JSONResponse({"error": "Search timed out"}, status_code=504)
except FileNotFoundError:
return JSONResponse({"error": "grep not available on this system"}, status_code=500)
return JSONResponse({
"query": q,
"count": len(results),
"results": results,
})
# ── Serve frontend ────────────────────────────────────────────────────
if STATIC_DIR.exists():
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
@app.get("/", response_class=HTMLResponse)
async def index():
index_path = STATIC_DIR / "index.html"
if index_path.exists():
return HTMLResponse(content=index_path.read_text())
return HTMLResponse(content="Command Center - Loading...
")