provvedo/wiregui
2
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages 1 Wiki Activity Actions
wiregui/wiregui/pages/devices.py
Stefano Bertelli 71a5f57105
Some checks failed
Dev / test (push) Failing after 2m43s
Details
Dev / docker (push) Has been skipped
Details
feat: live traffic chart, connection status indicators, 5s refresh 
- Add ECharts live traffic rate chart on device detail page (RX/s + TX/s
  area lines, 60-point rolling window, human-readable byte axis)
- Add traffic rate display (B/s) next to RX/TX totals
- Add connection status column (green/yellow/red dot) to user and admin
  device tables based on handshake age
- Add status badge to device detail page
- Reduce all UI refresh timers from 30s to 5s
- Add row click navigation on admin devices table
- Allow admins to view any device detail (not just their own)
- Fix rowClick event args (list not dict) on both device tables
- Add connection_status() helper in utils/time.py
2026-03-31 19:12:33 -05:00

605 lines
28 KiB
Python
Raw Blame History

"""User-facing device management pages."""
import asyncio
import io
from uuid import UUID
import qrcode
import qrcode.image.svg
from loguru import logger
from nicegui import app, ui
from sqlmodel import select
from wiregui.config import get_settings
from wiregui.db import async_session
from wiregui.models.device import Device
from wiregui.pages.layout import layout
from wiregui.services.events import on_device_created, on_device_deleted, on_device_updated
from wiregui.utils.crypto import generate_keypair, generate_preshared_key
from wiregui.utils.network import allocate_ipv4, allocate_ipv6
from wiregui.utils.server_key import get_server_public_key
from wiregui.utils.wg_conf import build_client_config
def _format_bytes(b: int | None) -> str:
if b is None:
return "-"
for unit in ("B", "KB", "MB", "GB", "TB"):
if b < 1024:
return f"{b:.1f} {unit}"
b /= 1024
return f"{b:.1f} PB"
@ui.page("/devices")
async def devices_page():
if not app.storage.user.get("authenticated"):
return ui.navigate.to("/login")
layout()
user_id = UUID(app.storage.user["user_id"])
settings = get_settings()
# Load client defaults from DB config (falls back to env vars)
async with async_session() as session:
from wiregui.models.configuration import Configuration
_db_cfg = (await session.execute(select(Configuration).limit(1))).scalar_one_or_none()
_defaults = {
"allowed_ips": ", ".join(_db_cfg.default_client_allowed_ips) if _db_cfg and _db_cfg.default_client_allowed_ips else settings.wg_allowed_ips,
"dns": ", ".join(_db_cfg.default_client_dns) if _db_cfg and _db_cfg.default_client_dns else settings.wg_dns,
"endpoint": _db_cfg.default_client_endpoint if _db_cfg and _db_cfg.default_client_endpoint else settings.wg_endpoint_host,
"mtu": str(_db_cfg.default_client_mtu) if _db_cfg else str(settings.wg_mtu),
"keepalive": str(_db_cfg.default_client_persistent_keepalive) if _db_cfg else str(settings.wg_persistent_keepalive),
}
async def load_devices() -> list[Device]:
async with async_session() as session:
result = await session.execute(
select(Device).where(Device.user_id == user_id).order_by(Device.inserted_at.desc())
)
return list(result.scalars().all())
async def refresh_table():
from wiregui.utils.time import connection_status
devices = await load_devices()
table.rows = [
{
"id": str(d.id),
"name": d.name,
"description": d.description or "",
"status_color": connection_status(d.latest_handshake)[0],
"status_label": connection_status(d.latest_handshake)[1],
"ipv4": d.ipv4 or "-",
"ipv6": d.ipv6 or "-",
"public_key": d.public_key[:16] + "...",
"rx": _format_bytes(d.rx_bytes),
"tx": _format_bytes(d.tx_bytes),
"handshake": str(d.latest_handshake)[:19] if d.latest_handshake else "-",
}
for d in devices
]
table.update()
# --- Create device ---
async def create_device():
name = create_name.value.strip()
if not name:
ui.notify("Device name is required", type="negative")
return
try:
settings = get_settings()
private_key, public_key = generate_keypair()
psk = generate_preshared_key()
async with async_session() as session:
ipv4 = await allocate_ipv4(session, settings.wg_ipv4_network)
ipv6 = await allocate_ipv6(session, settings.wg_ipv6_network)
device = Device(
name=name,
description=create_desc.value.strip() or None,
public_key=public_key,
preshared_key=psk,
ipv4=ipv4,
ipv6=ipv6,
user_id=user_id,
use_default_allowed_ips=create_use_default_ips.value,
use_default_dns=create_use_default_dns.value,
use_default_endpoint=create_use_default_endpoint.value,
use_default_mtu=create_use_default_mtu.value,
use_default_persistent_keepalive=create_use_default_keepalive.value,
endpoint=(create_endpoint.value.strip() or None
if not create_use_default_endpoint.value else None),
dns=([s.strip() for s in create_dns.value.split(",") if s.strip()]
if not create_use_default_dns.value and create_dns.value else []),
mtu=(int(create_mtu.value)
if not create_use_default_mtu.value and create_mtu.value else None),
persistent_keepalive=(int(create_keepalive.value)
if not create_use_default_keepalive.value and create_keepalive.value else None),
allowed_ips=([s.strip() for s in create_allowed_ips.value.split(",") if s.strip()]
if not create_use_default_ips.value and create_allowed_ips.value else []),
)
session.add(device)
await session.commit()
await session.refresh(device)
logger.info("Device created: {} ({})", device.name, device.ipv4)
# Build config and show dialog immediately — don't wait for WG/firewall
server_pubkey = await get_server_public_key()
async with async_session() as session:
from sqlmodel import select as sel
from wiregui.models.configuration import Configuration
db_config = (await session.execute(sel(Configuration).limit(1))).scalar_one_or_none()
config_text = build_client_config(device, private_key, server_pubkey, db_config)
create_dialog.close()
_reset_create_form()
await refresh_table()
_show_config_dialog(device.name, config_text)
# Configure WG peer and firewall in background (don't block the UI)
asyncio.create_task(on_device_created(device))
except Exception as e:
logger.error("Failed to create device: {}", e)
try:
ui.notify(f"Error: {e}", type="negative")
except RuntimeError:
pass
def _reset_create_form():
create_name.value = ""
create_desc.value = ""
create_use_default_ips.value = True
create_use_default_dns.value = True
create_use_default_endpoint.value = True
create_use_default_mtu.value = True
create_use_default_keepalive.value = True
create_allowed_ips.value = _defaults["allowed_ips"]
create_dns.value = _defaults["dns"]
create_endpoint.value = _defaults["endpoint"]
create_mtu.value = _defaults["mtu"]
create_keepalive.value = _defaults["keepalive"]
# --- Delete device ---
async def delete_device(device_id: str):
async with async_session() as session:
device = await session.get(Device, UUID(device_id))
if device and device.user_id == user_id:
await session.delete(device)
await session.commit()
logger.info("Device deleted: {}", device.name)
await on_device_deleted(device)
ui.notify(f"Deleted {device.name}")
await refresh_table()
def on_row_click(e):
row = e.args[1] if isinstance(e.args, list) else e.args
ui.navigate.to(f"/devices/{row['id']}")
# --- Page content ---
with ui.column().classes("w-full p-4"):
with ui.row().classes("w-full items-center justify-between"):
ui.label("My Devices").classes("text-h5")
ui.button("Add Device", icon="add", on_click=lambda: create_dialog.open()).props("color=primary")
columns = [
{"name": "status", "label": "", "field": "status_label", "align": "center"},
{"name": "name", "label": "Name", "field": "name", "align": "left", "sortable": True},
{"name": "ipv4", "label": "IPv4", "field": "ipv4", "align": "left"},
{"name": "ipv6", "label": "IPv6", "field": "ipv6", "align": "left"},
{"name": "public_key", "label": "Public Key", "field": "public_key", "align": "left"},
{"name": "rx", "label": "RX", "field": "rx", "align": "right"},
{"name": "tx", "label": "TX", "field": "tx", "align": "right"},
{"name": "handshake", "label": "Last Handshake", "field": "handshake", "align": "left"},
{"name": "actions", "label": "", "field": "id", "align": "center"},
]
table = ui.table(columns=columns, rows=[], row_key="id").classes("w-full")
table.on("rowClick", on_row_click)
table.add_slot(
"body-cell-status",
'''
<q-td :props="props">
<q-badge :color="props.row.status_color" rounded class="q-mr-sm" />
<span class="text-caption">{{ props.row.status_label }}</span>
</q-td>
''',
)
table.add_slot(
"body-cell-actions",
'''
<q-td :props="props">
<q-btn flat dense icon="delete" color="negative"
@click.stop="() => $parent.$emit('delete', props.row.id)" />
</q-td>
''',
)
table.on("delete", lambda e: delete_device(e.args))
# --- Create device dialog (full form) ---
with ui.dialog() as create_dialog:
with ui.card().classes("w-[600px]"):
ui.label("New Device").classes("text-h6")
create_name = ui.input("Device Name").props("outlined dense").classes("w-full")
create_desc = ui.input("Description (optional)").props("outlined dense").classes("w-full")
ui.separator().classes("q-my-sm")
ui.label("Configuration Overrides").classes("text-subtitle2")
ui.label("Toggle off to set custom values instead of server defaults.").classes("text-caption text-grey")
with ui.grid(columns=2).classes("w-full gap-2"):
create_use_default_ips = ui.switch("Use default Allowed IPs", value=True)
create_allowed_ips = ui.input("Allowed IPs", value=_defaults["allowed_ips"]).props(
"outlined dense"
).classes("w-full").bind_enabled_from(create_use_default_ips, "value", backward=lambda v: not v)
create_use_default_dns = ui.switch("Use default DNS", value=True)
create_dns = ui.input("DNS Servers", value=_defaults["dns"]).props(
"outlined dense"
).classes("w-full").bind_enabled_from(create_use_default_dns, "value", backward=lambda v: not v)
create_use_default_endpoint = ui.switch("Use default Endpoint", value=True)
create_endpoint = ui.input("Endpoint", value=_defaults["endpoint"]).props(
"outlined dense"
).classes("w-full").bind_enabled_from(create_use_default_endpoint, "value", backward=lambda v: not v)
create_use_default_mtu = ui.switch("Use default MTU", value=True)
create_mtu = ui.input("MTU", value=_defaults["mtu"]).props(
"outlined dense"
).classes("w-full").bind_enabled_from(create_use_default_mtu, "value", backward=lambda v: not v)
create_use_default_keepalive = ui.switch("Use default Keepalive", value=True)
create_keepalive = ui.input("Persistent Keepalive", value=_defaults["keepalive"]).props(
"outlined dense"
).classes("w-full").bind_enabled_from(create_use_default_keepalive, "value", backward=lambda v: not v)
with ui.row().classes("w-full justify-end q-mt-md"):
ui.button("Cancel", on_click=create_dialog.close).props("flat")
ui.button("Create", on_click=create_device).props("color=primary")
await refresh_table()
ui.timer(5, refresh_table)
@ui.page("/devices/{device_id}")
async def device_detail_page(device_id: str):
if not app.storage.user.get("authenticated"):
return ui.navigate.to("/login")
layout()
user_id = UUID(app.storage.user["user_id"])
role = app.storage.user.get("role", "")
async with async_session() as sess:
device = await sess.get(Device, UUID(device_id))
if not device or (device.user_id != user_id and role != "admin"):
ui.label("Device not found").classes("text-h5 text-negative p-4")
return
# --- Edit handlers ---
async def save_edit():
async with async_session() as session:
d = await session.get(Device, UUID(device_id))
if not d:
return
d.name = edit_name.value.strip()
d.description = edit_desc.value.strip() or None
d.use_default_allowed_ips = edit_use_default_ips.value
d.use_default_dns = edit_use_default_dns.value
d.use_default_endpoint = edit_use_default_endpoint.value
d.use_default_mtu = edit_use_default_mtu.value
d.use_default_persistent_keepalive = edit_use_default_keepalive.value
if not d.use_default_endpoint:
d.endpoint = edit_endpoint.value.strip() or None
if not d.use_default_dns:
d.dns = [s.strip() for s in edit_dns.value.split(",") if s.strip()]
if not d.use_default_mtu:
d.mtu = int(edit_mtu.value) if edit_mtu.value else None
if not d.use_default_persistent_keepalive:
d.persistent_keepalive = int(edit_keepalive.value) if edit_keepalive.value else None
if not d.use_default_allowed_ips:
d.allowed_ips = [s.strip() for s in edit_allowed_ips.value.split(",") if s.strip()]
session.add(d)
await session.commit()
await session.refresh(d)
await on_device_updated(d)
logger.info("Device updated: {}", edit_name.value)
ui.notify("Device updated", type="positive")
ui.navigate.to(f"/devices/{device_id}")
async def delete_and_redirect():
async with async_session() as session:
d = await session.get(Device, UUID(device_id))
if d:
await session.delete(d)
await session.commit()
logger.info("Device deleted: {}", d.name)
await on_device_deleted(d)
ui.navigate.to("/devices")
settings = get_settings()
# --- Page content ---
with ui.column().classes("w-full p-4"):
with ui.row().classes("items-center gap-2"):
ui.button(icon="arrow_back", on_click=lambda: ui.navigate.to("/devices")).props("flat")
ui.label(device.name).classes("text-h5")
if device.description:
ui.label(f"— {device.description}").classes("text-subtitle1 text-grey")
# Device info card
with ui.card().classes("w-full q-mt-md"):
ui.label("Device Details").classes("text-subtitle1 text-bold")
ui.separator()
with ui.grid(columns=2).classes("w-full gap-2 q-pa-sm"):
ui.label("Public Key:").classes("text-bold")
ui.label(device.public_key).classes("font-mono text-sm")
ui.label("IPv4:").classes("text-bold")
ui.label(device.ipv4 or "-")
ui.label("IPv6:").classes("text-bold")
ui.label(device.ipv6 or "-")
ui.label("Created:").classes("text-bold")
ui.label(str(device.inserted_at)[:19])
# Traffic stats (live-updating)
with ui.card().classes("w-full q-mt-md"):
ui.label("Traffic Stats").classes("text-subtitle1 text-bold")
ui.separator()
from wiregui.utils.time import connection_status
_color, _label = connection_status(device.latest_handshake)
with ui.row().classes("items-center gap-2 q-pa-sm"):
stat_badge = ui.badge("", color=_color).props("rounded")
stat_status = ui.label(_label).classes("text-caption")
with ui.grid(columns=3).classes("w-full gap-2 q-pa-sm"):
ui.label("RX:").classes("text-bold")
stat_rx = ui.label(_format_bytes(device.rx_bytes))
stat_rx_rate = ui.label("").classes("text-caption text-grey")
ui.label("TX:").classes("text-bold")
stat_tx = ui.label(_format_bytes(device.tx_bytes))
stat_tx_rate = ui.label("").classes("text-caption text-grey")
ui.label("Last Handshake:").classes("text-bold")
stat_handshake = ui.label(str(device.latest_handshake)[:19] if device.latest_handshake else "-")
ui.label("") # spacer
ui.label("Remote IP:").classes("text-bold")
stat_remote = ui.label(device.remote_ip or "-")
ui.label("") # spacer
# Traffic chart
MAX_CHART_POINTS = 60
_chart_times: list[str] = []
_chart_rx: list[float] = []
_chart_tx: list[float] = []
with ui.card().classes("w-full q-mt-md"):
ui.label("Traffic Rate").classes("text-subtitle1 text-bold")
ui.separator()
traffic_chart = ui.echart({
"tooltip": {
"trigger": "axis",
":valueFormatter": """(v) => {
if (v >= 1048576) return (v / 1048576).toFixed(1) + ' MB/s';
if (v >= 1024) return (v / 1024).toFixed(1) + ' KB/s';
return v.toFixed(0) + ' B/s';
}""",
},
"legend": {"data": ["RX/s", "TX/s"], "right": 20, "top": 5},
"xAxis": {"type": "category", "data": [], "boundaryGap": False},
"yAxis": {
"type": "value",
"axisLabel": {
":formatter": """(v) => {
if (v >= 1073741824) return (v / 1073741824).toFixed(1) + ' GB/s';
if (v >= 1048576) return (v / 1048576).toFixed(1) + ' MB/s';
if (v >= 1024) return (v / 1024).toFixed(1) + ' KB/s';
return v.toFixed(0) + ' B/s';
}""",
},
},
"series": [
{
"name": "RX/s",
"type": "line",
"smooth": True,
"symbol": "none",
"areaStyle": {"opacity": 0.15},
"lineStyle": {"width": 2},
"itemStyle": {"color": "#3598C3"},
"data": [],
},
{
"name": "TX/s",
"type": "line",
"smooth": True,
"symbol": "none",
"areaStyle": {"opacity": 0.15},
"lineStyle": {"width": 2},
"itemStyle": {"color": "#5AA6B9"},
"data": [],
},
],
"grid": {"left": 60, "right": 20, "top": 40, "bottom": 30},
}).classes("w-full").style("height: 250px")
_prev_rx = device.rx_bytes or 0
_prev_tx = device.tx_bytes or 0
_prev = {"rx": _prev_rx, "tx": _prev_tx}
async def refresh_stats():
from wiregui.utils.time import connection_status
from datetime import datetime
async with async_session() as session:
d = await session.get(Device, UUID(device_id))
if not d:
return
# Compute rates
cur_rx = d.rx_bytes or 0
cur_tx = d.tx_bytes or 0
rx_rate = max(0, (cur_rx - _prev["rx"]) / 5)
tx_rate = max(0, (cur_tx - _prev["tx"]) / 5)
_prev["rx"] = cur_rx
_prev["tx"] = cur_tx
# Update labels
stat_rx.text = _format_bytes(d.rx_bytes)
stat_tx.text = _format_bytes(d.tx_bytes)
stat_rx_rate.text = f"({_format_bytes(int(rx_rate))}/s)"
stat_tx_rate.text = f"({_format_bytes(int(tx_rate))}/s)"
stat_handshake.text = str(d.latest_handshake)[:19] if d.latest_handshake else "-"
stat_remote.text = d.remote_ip or "-"
color, label = connection_status(d.latest_handshake)
stat_badge.props(f'color={color}')
stat_status.text = label
# Update chart
now = datetime.now().strftime("%H:%M:%S")
_chart_times.append(now)
_chart_rx.append(round(rx_rate, 1))
_chart_tx.append(round(tx_rate, 1))
if len(_chart_times) > MAX_CHART_POINTS:
_chart_times.pop(0)
_chart_rx.pop(0)
_chart_tx.pop(0)
traffic_chart.options["xAxis"]["data"] = _chart_times
traffic_chart.options["series"][0]["data"] = _chart_rx
traffic_chart.options["series"][1]["data"] = _chart_tx
traffic_chart.update()
ui.timer(5, refresh_stats)
# Active configuration
with ui.card().classes("w-full q-mt-md"):
ui.label("Active Configuration").classes("text-subtitle1 text-bold")
ui.separator()
with ui.grid(columns=2).classes("w-full gap-2 q-pa-sm"):
_ips = device.allowed_ips if not device.use_default_allowed_ips else settings.wg_allowed_ips
ui.label("Allowed IPs:").classes("text-bold")
ui.label(str(_ips) if isinstance(_ips, str) else ", ".join(_ips) if _ips else "-")
_dns = device.dns if not device.use_default_dns else settings.wg_dns
ui.label("DNS:").classes("text-bold")
ui.label(str(_dns) if isinstance(_dns, str) else ", ".join(_dns) if _dns else "-")
_ep = device.endpoint if not device.use_default_endpoint else settings.wg_endpoint_host
ui.label("Endpoint:").classes("text-bold")
ui.label(f"{_ep}:{settings.wg_endpoint_port}" if _ep else "-")
_mtu = device.mtu if not device.use_default_mtu else settings.wg_mtu
ui.label("MTU:").classes("text-bold")
ui.label(str(_mtu) if _mtu else "-")
_ka = device.persistent_keepalive if not device.use_default_persistent_keepalive else settings.wg_persistent_keepalive
ui.label("Persistent Keepalive:").classes("text-bold")
ui.label(str(_ka) if _ka else "-")
# Edit form
with ui.card().classes("w-full q-mt-md"):
ui.label("Edit Device").classes("text-subtitle1 text-bold")
ui.separator()
edit_name = ui.input("Device Name", value=device.name).props("outlined dense").classes("w-full")
edit_desc = ui.input("Description", value=device.description or "").props("outlined dense").classes("w-full")
ui.separator().classes("q-my-sm")
ui.label("Configuration Overrides").classes("text-subtitle2")
with ui.grid(columns=2).classes("w-full gap-2"):
edit_use_default_ips = ui.switch("Use default Allowed IPs", value=device.use_default_allowed_ips)
edit_allowed_ips = ui.input(
"Allowed IPs", value=", ".join(device.allowed_ips) if device.allowed_ips else "",
).props("outlined dense").classes("w-full").bind_enabled_from(
edit_use_default_ips, "value", backward=lambda v: not v
)
edit_use_default_dns = ui.switch("Use default DNS", value=device.use_default_dns)
edit_dns = ui.input(
"DNS Servers", value=", ".join(device.dns) if device.dns else "",
).props("outlined dense").classes("w-full").bind_enabled_from(
edit_use_default_dns, "value", backward=lambda v: not v
)
edit_use_default_endpoint = ui.switch("Use default Endpoint", value=device.use_default_endpoint)
edit_endpoint = ui.input(
"Endpoint", value=device.endpoint or "",
).props("outlined dense").classes("w-full").bind_enabled_from(
edit_use_default_endpoint, "value", backward=lambda v: not v
)
edit_use_default_mtu = ui.switch("Use default MTU", value=device.use_default_mtu)
edit_mtu = ui.input(
"MTU", value=str(device.mtu) if device.mtu else "",
).props("outlined dense").classes("w-full").bind_enabled_from(
edit_use_default_mtu, "value", backward=lambda v: not v
)
edit_use_default_keepalive = ui.switch("Use default Keepalive", value=device.use_default_persistent_keepalive)
edit_keepalive = ui.input(
"Persistent Keepalive", value=str(device.persistent_keepalive) if device.persistent_keepalive else "",
).props("outlined dense").classes("w-full").bind_enabled_from(
edit_use_default_keepalive, "value", backward=lambda v: not v
)
ui.button("Save Changes", on_click=save_edit).props("color=primary").classes("q-mt-md")
# Danger zone
with ui.card().classes("w-full q-mt-md"):
ui.label("Danger Zone").classes("text-subtitle1 text-bold text-negative")
ui.separator()
ui.button("Delete Device", icon="delete", on_click=lambda: confirm_dialog.open()).props(
"color=negative unelevated"
)
# Confirm delete dialog
with ui.dialog() as confirm_dialog:
with ui.card().classes("w-80"):
ui.label("Delete Device?").classes("text-h6")
ui.label(f"This will permanently remove '{device.name}' and its WireGuard peer.").classes("text-body2")
with ui.row().classes("w-full justify-end q-mt-sm"):
ui.button("Cancel", on_click=confirm_dialog.close).props("flat")
ui.button("Delete", on_click=delete_and_redirect).props("color=negative")
def _show_config_dialog(device_name: str, config_text: str):
"""Show a dialog with the WireGuard client configuration and QR code."""
with ui.dialog(value=True) as dialog:
with ui.card().classes("w-[700px] max-w-[90vw]"):
ui.label(f"Config for {device_name}").classes("text-h6")
ui.label("Save this — the private key won't be shown again.").classes("text-caption text-negative q-mb-sm")
ui.code(config_text, language="ini").classes("w-full")
try:
import base64
qr = qrcode.make(config_text)
buf = io.BytesIO()
qr.save(buf, format="PNG")
b64 = base64.b64encode(buf.getvalue()).decode()
with ui.row().classes("w-full justify-center q-mt-md"):
ui.image(f"data:image/png;base64,{b64}").style(
"width: 200px; height: 200px; border-radius: 8px"
)
except Exception:
ui.label("QR code generation failed").classes("text-caption text-grey")
with ui.row().classes("w-full gap-2 q-mt-md"):
ui.button(
"Download .conf",
on_click=lambda: ui.download(config_text.encode(), f"{device_name}.conf"),
).props("color=primary unelevated").classes("flex-grow")
ui.button("Close", on_click=dialog.close).props("flat")
Reference in a new issue View git blame Copy permalink