hmm
This commit is contained in:
mjallen18
252
packages/system/nebula-ui/app.py
Normal file
252
packages/system/nebula-ui/app.py
Normal file
@@ -0,0 +1,252 @@
|
||||
"""
|
||||
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,
|
||||
},
|
||||
)
|
||||
Reference in New Issue
Block a user