253 lines
7.5 KiB
Python
253 lines
7.5 KiB
Python
"""
|
|
nebula-ui — FastAPI web UI for managing a Nebula overlay network.
|
|
|
|
Reads configuration from environment variables (set by the NixOS module):
|
|
NEBULA_UI_CA_CERT_PATH — path to the decrypted CA cert file (from sops-nix)
|
|
NEBULA_UI_CA_KEY_PATH — path to the decrypted CA key file (from sops-nix)
|
|
NEBULA_UI_STATS_URL — nebula stats HTTP endpoint, e.g. http://127.0.0.1:8472
|
|
NEBULA_UI_NETWORK_NAME — nebula network name, e.g. "jallen-nebula"
|
|
NEBULA_UI_LISTEN_HOST — host to bind to (default: 127.0.0.1)
|
|
NEBULA_UI_LISTEN_PORT — port to listen on (default: 8472)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
|
|
import httpx
|
|
import qrcode
|
|
import qrcode.image.svg
|
|
from fastapi import FastAPI, Form, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config from environment
|
|
# ---------------------------------------------------------------------------
|
|
CA_CERT_PATH = os.environ.get("NEBULA_UI_CA_CERT_PATH", "")
|
|
CA_KEY_PATH = os.environ.get("NEBULA_UI_CA_KEY_PATH", "")
|
|
STATS_URL = os.environ.get("NEBULA_UI_STATS_URL", "http://127.0.0.1:8472")
|
|
NETWORK_NAME = os.environ.get("NEBULA_UI_NETWORK_NAME", "nebula")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# App
|
|
# ---------------------------------------------------------------------------
|
|
app = FastAPI(title="Nebula UI", docs_url=None, redoc_url=None)
|
|
|
|
_templates_dir = Path(__file__).parent / "templates"
|
|
# When installed via Nix the templates are copied into the same package dir
|
|
# as __init__.py (i.e. nebula_ui/templates/). __file__ resolves correctly.
|
|
templates = Jinja2Templates(directory=str(_templates_dir))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _qr_svg(data: str) -> str:
|
|
"""Return an inline SVG string for a QR code encoding *data*."""
|
|
factory = qrcode.image.svg.SvgPathImage
|
|
img = qrcode.make(
|
|
data, image_factory=factory, error_correction=qrcode.constants.ERROR_CORRECT_L
|
|
)
|
|
buf = io.BytesIO()
|
|
img.save(buf)
|
|
return buf.getvalue().decode()
|
|
|
|
|
|
def _nebula_cert_bin() -> str:
|
|
"""Return the path to nebula-cert (must be on PATH)."""
|
|
path = shutil.which("nebula-cert")
|
|
if path is None:
|
|
raise RuntimeError("nebula-cert not found on PATH")
|
|
return path
|
|
|
|
|
|
def _sign_cert(name: str, ip: str, groups: str, duration: str) -> tuple[str, str]:
|
|
"""
|
|
Sign a new Nebula host certificate using the CA files pointed to by
|
|
NEBULA_UI_CA_CERT_PATH / NEBULA_UI_CA_KEY_PATH.
|
|
|
|
Returns (cert_pem, key_pem) as strings.
|
|
|
|
Raises RuntimeError on failure.
|
|
"""
|
|
if not CA_CERT_PATH or not CA_KEY_PATH:
|
|
raise RuntimeError(
|
|
"CA cert/key paths are not configured "
|
|
"(NEBULA_UI_CA_CERT_PATH / NEBULA_UI_CA_KEY_PATH)"
|
|
)
|
|
ca_cert = Path(CA_CERT_PATH)
|
|
ca_key = Path(CA_KEY_PATH)
|
|
if not ca_cert.exists():
|
|
raise RuntimeError(f"CA cert not found: {ca_cert}")
|
|
if not ca_key.exists():
|
|
raise RuntimeError(f"CA key not found: {ca_key}")
|
|
|
|
with tempfile.TemporaryDirectory(prefix="nebula-ui-") as tmp:
|
|
tmp_path = Path(tmp)
|
|
out_crt = tmp_path / "host.crt"
|
|
out_key = tmp_path / "host.key"
|
|
|
|
cmd = [
|
|
_nebula_cert_bin(),
|
|
"sign",
|
|
"-name",
|
|
name,
|
|
"-ip",
|
|
ip,
|
|
"-ca-crt",
|
|
str(ca_cert),
|
|
"-ca-key",
|
|
str(ca_key),
|
|
"-out-crt",
|
|
str(out_crt),
|
|
"-out-key",
|
|
str(out_key),
|
|
]
|
|
if groups.strip():
|
|
cmd += ["-groups", groups.strip()]
|
|
if duration.strip():
|
|
cmd += ["-duration", duration.strip()]
|
|
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"nebula-cert sign failed:\n{result.stderr}")
|
|
|
|
cert_pem = out_crt.read_text()
|
|
key_pem = out_key.read_text()
|
|
|
|
return cert_pem, key_pem
|
|
|
|
|
|
def _cert_info(cert_pem: str) -> dict:
|
|
"""Return parsed cert info by running nebula-cert print."""
|
|
with tempfile.NamedTemporaryFile(suffix=".crt", mode="w", delete=False) as f:
|
|
f.write(cert_pem)
|
|
f.flush()
|
|
result = subprocess.run(
|
|
[_nebula_cert_bin(), "print", "-json", "-path", f.name],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
Path(f.name).unlink(missing_ok=True)
|
|
if result.returncode != 0:
|
|
return {}
|
|
try:
|
|
return json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
|
|
|
|
def _fetch_stats() -> dict | None:
|
|
"""Fetch the nebula stats JSON. Returns None on error."""
|
|
try:
|
|
resp = httpx.get(f"{STATS_URL}/v1/stats", timeout=3.0)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception:
|
|
pass
|
|
# Some nebula versions use /stats
|
|
try:
|
|
resp = httpx.get(f"{STATS_URL}/stats", timeout=3.0)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _fetch_hostmap() -> dict | None:
|
|
"""Fetch the nebula hostmap JSON."""
|
|
try:
|
|
resp = httpx.get(f"{STATS_URL}/v1/hostmap", timeout=3.0)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
resp = httpx.get(f"{STATS_URL}/hostmap", timeout=3.0)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def index(request: Request):
|
|
stats = _fetch_stats()
|
|
hostmap = _fetch_hostmap()
|
|
return templates.TemplateResponse(
|
|
"index.html",
|
|
{
|
|
"request": request,
|
|
"network_name": NETWORK_NAME,
|
|
"stats": stats,
|
|
"hostmap": hostmap,
|
|
},
|
|
)
|
|
|
|
|
|
@app.get("/sign", response_class=HTMLResponse)
|
|
async def sign_form(request: Request):
|
|
return templates.TemplateResponse(
|
|
"sign.html",
|
|
{
|
|
"request": request,
|
|
"network_name": NETWORK_NAME,
|
|
"error": None,
|
|
"result": None,
|
|
},
|
|
)
|
|
|
|
|
|
@app.post("/sign", response_class=HTMLResponse)
|
|
async def sign_submit(
|
|
request: Request,
|
|
name: Annotated[str, Form()],
|
|
ip: Annotated[str, Form()],
|
|
groups: Annotated[str, Form()] = "",
|
|
duration: Annotated[str, Form()] = "",
|
|
):
|
|
error = None
|
|
result = None
|
|
try:
|
|
cert_pem, key_pem = _sign_cert(name, ip, groups, duration)
|
|
cert_info = _cert_info(cert_pem)
|
|
cert_qr = _qr_svg(cert_pem)
|
|
key_qr = _qr_svg(key_pem)
|
|
result = {
|
|
"name": name,
|
|
"ip": ip,
|
|
"cert_pem": cert_pem,
|
|
"key_pem": key_pem,
|
|
"cert_qr": cert_qr,
|
|
"key_qr": key_qr,
|
|
"cert_info": cert_info,
|
|
}
|
|
except Exception as exc:
|
|
error = str(exc)
|
|
|
|
return templates.TemplateResponse(
|
|
"sign.html",
|
|
{
|
|
"request": request,
|
|
"network_name": NETWORK_NAME,
|
|
"error": error,
|
|
"result": result,
|
|
},
|
|
)
|