Stefano Bertelli
0546b44507
Complete Python/NiceGUI rewrite of the Wirezone (Elixir/Phoenix) VPN management platform. All 10 implementation phases delivered. Core stack: - NiceGUI reactive UI with SQLModel ORM on PostgreSQL (asyncpg) - Alembic migrations, Valkey/Redis cache, pydantic-settings config - WireGuard management via subprocess (wg/ip/nft CLIs) - 164 tests passing, 35% code coverage Features: - User/device/rule CRUD with admin and unprivileged roles - Full device config form with per-device WG overrides - WireGuard client config generation with QR codes - REST API (v0) with Bearer token auth for all resources - TOTP MFA with QR registration and challenge flow - OIDC SSO with authlib (provider registry, auto-create users) - Magic link passwordless sign-in via email - SAML SP-initiated SSO with IdP metadata parsing - WebAuthn/FIDO2 security key registration - nftables firewall with per-user chains and masquerade - Background tasks: WG stats polling, VPN session expiry, OIDC token refresh, WAN connectivity checks - Startup reconciliation (DB ↔ WireGuard state sync) - In-memory notification system with header badge - Admin UI: users, devices, rules, settings (3 tabs), diagnostics - Loguru logging with optional timestamped file output Deployment: - Multi-stage Dockerfile (python:3.13-slim) - Docker Compose prod stack (bridge networking, NET_ADMIN, nftables) - Forgejo CI: tests → semantic versioning → Docker registry push - Health endpoint at /api/health
65 lines
1.7 KiB
Python
65 lines
1.7 KiB
Python
"""In-memory notification queue with severity levels."""
|
|
|
|
from collections import deque
|
|
from datetime import datetime
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from loguru import logger
|
|
|
|
from wiregui.utils.time import utcnow
|
|
|
|
MAX_NOTIFICATIONS = 100
|
|
|
|
|
|
class Notification:
|
|
__slots__ = ("id", "severity", "message", "user", "timestamp")
|
|
|
|
def __init__(self, severity: str, message: str, user: str | None = None):
|
|
self.id = str(uuid4())
|
|
self.severity = severity # "info" | "warning" | "error"
|
|
self.message = message
|
|
self.user = user
|
|
self.timestamp = utcnow()
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"severity": self.severity,
|
|
"message": self.message,
|
|
"user": self.user,
|
|
"timestamp": self.timestamp.isoformat(),
|
|
}
|
|
|
|
|
|
_notifications: deque[Notification] = deque(maxlen=MAX_NOTIFICATIONS)
|
|
|
|
|
|
def add(severity: str, message: str, user: str | None = None) -> Notification:
|
|
"""Add a notification to the queue."""
|
|
n = Notification(severity, message, user)
|
|
_notifications.appendleft(n)
|
|
logger.debug("Notification added: [{}] {}", severity, message)
|
|
return n
|
|
|
|
|
|
def current() -> list[Notification]:
|
|
"""Return all current notifications (newest first)."""
|
|
return list(_notifications)
|
|
|
|
|
|
def clear(notification_id: str) -> None:
|
|
"""Remove a specific notification by ID."""
|
|
for i, n in enumerate(_notifications):
|
|
if n.id == notification_id:
|
|
del _notifications[i]
|
|
return
|
|
|
|
|
|
def clear_all() -> None:
|
|
"""Remove all notifications."""
|
|
_notifications.clear()
|
|
|
|
|
|
def count() -> int:
|
|
return len(_notifications)
|