Add PiVPN status page

docs/pivpn.md

@@ -0,0 +1,37 @@
+# PiVPN
+
+SimpleSaferServer has a read-only PiVPN page at `/pivpn`.
+
+The page is meant to help an admin see whether PiVPN looks present on the server and whether the common WireGuard or OpenVPN systemd services are running. It also shows common PiVPN commands that can be copied into a root terminal.
+
+## What The Page Shows
+
+- Whether the `pivpn` command is available to the web service.
+- The status of `wg-quick@wg0.service`, which is the common WireGuard unit used by PiVPN.
+- The status of common OpenVPN units: `openvpn@server.service` and `openvpn.service`.
+- Copyable commands for listing, adding, removing, and debugging PiVPN clients.
+
+## What The Page Does Not Do
+
+The page does not add, remove, or edit VPN clients from the browser.
+
+That is intentional. PiVPN already provides terminal tools for those jobs, and adding browser-side VPN management would need more careful handling of profile files, QR codes, and client secrets.
+
+## Useful Terminal Commands
+
+Run these on the server:
+
+```bash
+pivpn -c
+pivpn -a
+pivpn -r
+pivpn -d
+```
+
+Use `sudo` first if your shell is not already running as root.
+
+## PiVPN Documentation
+
+Use the official PiVPN docs for installation and VPN-specific setup:
+
+`https://docs.pivpn.io/`

index.html

@@ -162,6 +162,7 @@ <h2 id="documentation" class="mb-3"><i class="fa-solid fa-book"></i> Documentati
             <li><i class="fa-solid fa-right-to-bracket"></i> <a href="https://github.com/chrismin13/SimpleSaferServer/blob/main/docs/login.md" target="_blank">Login &amp; User Management</a></li>
             <li><i class="fa-solid fa-gauge-high"></i> <a href="https://github.com/chrismin13/SimpleSaferServer/blob/main/docs/dashboard.md" target="_blank">Dashboard</a></li>
             <li><i class="fa-solid fa-hard-drive"></i> <a href="https://github.com/chrismin13/SimpleSaferServer/blob/main/docs/drive_health.md" target="_blank">Drive Health</a></li>
+            <li><i class="fa-solid fa-shield-halved"></i> <a href="https://github.com/chrismin13/SimpleSaferServer/blob/main/docs/pivpn.md" target="_blank">PiVPN</a></li>
             <li><i class="fa-solid fa-cloud-arrow-up"></i> <a href="https://github.com/chrismin13/SimpleSaferServer/blob/main/docs/cloud_backup.md" target="_blank">Cloud Backup</a></li>
             <li><i class="fa-solid fa-download"></i> <a href="https://github.com/chrismin13/SimpleSaferServer/blob/main/docs/system_updates.md" target="_blank">System Updates</a></li>
             <li><i class="fa-solid fa-globe"></i> <a href="https://github.com/chrismin13/SimpleSaferServer/blob/main/docs/ddns.md" target="_blank">Dynamic DNS</a></li>

simple_safer_server/app_factory.py

@@ -21,6 +21,7 @@
 from simple_safer_server.routes.cloud_backup import cloud_backup as cloud_backup_routes
 from simple_safer_server.routes.ddns import ddns as ddns_routes
 from simple_safer_server.routes.drive_health import drive_health as drive_health_routes
+from simple_safer_server.routes.pivpn import pivpn as pivpn_routes
 from simple_safer_server.routes.server_identity import server_identity as server_identity_routes
 from simple_safer_server.routes.setup_wizard import setup
 from simple_safer_server.routes.smb import smb as smb_routes
@@ -37,6 +38,7 @@
 from simple_safer_server.services.ddns_service import DdnsService
 from simple_safer_server.services.disabled_timers import DisabledTimerService
 from simple_safer_server.services.drive_health import DriveHealthSummaryService
+from simple_safer_server.services.pivpn_service import PiVpnService
 from simple_safer_server.services.runtime import get_fake_state, get_flask_secret_key, get_runtime
 from simple_safer_server.services.server_identity import ServerIdentityService
 from simple_safer_server.services.smb_manager import SMB_DOCS_URL, SMBManager
@@ -156,6 +158,7 @@ def create_app() -> Flask:
         command_adapter=storage_command_adapter,
     )
     drive_health_summary_service = DriveHealthSummaryService()
+    pivpn_service = PiVpnService(runtime=runtime, systemd_adapter=systemd_adapter)
     app.extensions["simple_safer_server"] = AppServices(
         runtime=runtime,
         fake_state=fake_state,
@@ -173,6 +176,7 @@ def create_app() -> Flask:
         server_identity_service=server_identity_service,
         storage_service=storage_service,
         drive_health_summary_service=drive_health_summary_service,
+        pivpn_service=pivpn_service,
     )
 
     app.register_blueprint(setup)
@@ -186,6 +190,7 @@ def create_app() -> Flask:
     app.register_blueprint(users_routes)
     app.register_blueprint(storage_routes)
     app.register_blueprint(drive_health_routes)
+    app.register_blueprint(pivpn_routes)
 
     @app.route("/")
     def index():

simple_safer_server/routes/pivpn.py

@@ -0,0 +1,22 @@
+from typing import Any
+
+from flask import Blueprint, current_app, render_template, session
+
+from simple_safer_server.services.user_manager import admin_required
+
+pivpn = Blueprint("pivpn_routes", __name__)
+
+
+def _get_services() -> Any:
+    """Return app-level services registered during Flask startup."""
+    return current_app.extensions["simple_safer_server"]
+
+
+@pivpn.route("/pivpn")
+@admin_required
+def pivpn_page():
+    return render_template(
+        "pivpn.html",
+        username=session.get("username"),
+        pivpn_status=_get_services().pivpn_service.get_page_status(),
+    )

simple_safer_server/services/container.py

@@ -7,6 +7,7 @@
 from simple_safer_server.services.cloud_backup_service import CloudBackupService
 from simple_safer_server.services.ddns_service import DdnsService
 from simple_safer_server.services.drive_health import DriveHealthSummaryService
+from simple_safer_server.services.pivpn_service import PiVpnService
 from simple_safer_server.services.server_identity import ServerIdentityService
 from simple_safer_server.services.storage_service import StorageService
 from simple_safer_server.services.task_service import TaskService
@@ -32,3 +33,4 @@ class AppServices:
     server_identity_service: ServerIdentityService
     storage_service: StorageService
     drive_health_summary_service: DriveHealthSummaryService
+    pivpn_service: PiVpnService

simple_safer_server/services/pivpn_service.py

@@ -0,0 +1,181 @@
+import shutil
+from dataclasses import dataclass
+from typing import Any
+
+PIVPN_DOCS_URL = "https://docs.pivpn.io/"
+PIVPN_SSS_DOCS_URL = "https://github.com/chrismin13/SimpleSaferServer/blob/main/docs/pivpn.md"
+
+WIREGUARD_UNITS = ("wg-quick@wg0.service",)
+OPENVPN_UNITS = ("openvpn@server.service", "openvpn.service")
+
+
+@dataclass(frozen=True)
+class PiVpnServiceStatus:
+    """A small service status value that templates can render directly."""
+
+    label: str
+    badge_class: str
+    detail: str
+    unit_name: str | None = None
+
+
+@dataclass(frozen=True)
+class PiVpnCommand:
+    """A terminal command worth showing without running it from the web page."""
+
+    label: str
+    command: str
+    description: str
+
+
+@dataclass(frozen=True)
+class PiVpnPageStatus:
+    installed: bool
+    install_status: PiVpnServiceStatus
+    wireguard_status: PiVpnServiceStatus
+    openvpn_status: PiVpnServiceStatus
+    command_path: str | None
+    commands: tuple[PiVpnCommand, ...]
+    docs_url: str = PIVPN_DOCS_URL
+    app_docs_url: str = PIVPN_SSS_DOCS_URL
+
+
+class PiVpnService:
+    """Builds the read-only PiVPN page state."""
+
+    def __init__(self, runtime: Any, systemd_adapter: Any) -> None:
+        self._runtime = runtime
+        self._systemd_adapter = systemd_adapter
+
+    def get_page_status(self) -> PiVpnPageStatus:
+        if self._runtime.is_fake:
+            return self._fake_status()
+
+        command_path = shutil.which("pivpn")
+        installed = command_path is not None
+        install_status = (
+            PiVpnServiceStatus("Installed", "badge-success", command_path or "pivpn found")
+            if installed
+            else PiVpnServiceStatus(
+                "Not found",
+                "badge-warning",
+                "Install PiVPN first, then come back here for status and commands.",
+            )
+        )
+
+        return PiVpnPageStatus(
+            installed=installed,
+            install_status=install_status,
+            wireguard_status=self._first_loaded_unit_status("WireGuard", WIREGUARD_UNITS),
+            openvpn_status=self._first_loaded_unit_status("OpenVPN", OPENVPN_UNITS),
+            command_path=command_path,
+            commands=PiVpnService.default_commands(),
+        )
+
+    @staticmethod
+    def default_commands() -> tuple[PiVpnCommand, ...]:
+        return (
+            PiVpnCommand("List clients", "pivpn -c", "Show configured VPN clients."),
+            PiVpnCommand("Add client", "pivpn -a", "Create a new VPN client profile."),
+            PiVpnCommand("Remove client", "pivpn -r", "Revoke a VPN client profile."),
+            PiVpnCommand("Debug", "pivpn -d", "Run PiVPN's built-in diagnostic check."),
+        )
+
+    def _fake_status(self) -> PiVpnPageStatus:
+        return PiVpnPageStatus(
+            installed=True,
+            install_status=PiVpnServiceStatus(
+                "Demo mode",
+                "badge-info",
+                "Fake mode shows what the PiVPN page looks like without reading this system.",
+            ),
+            wireguard_status=PiVpnServiceStatus(
+                "Active",
+                "badge-success",
+                "Fake WireGuard status",
+                "wg-quick@wg0.service",
+            ),
+            openvpn_status=PiVpnServiceStatus(
+                "Not found",
+                "badge-neutral",
+                "Fake mode does not include an OpenVPN unit.",
+            ),
+            command_path="/usr/local/bin/pivpn",
+            commands=PiVpnService.default_commands(),
+        )
+
+    def _first_loaded_unit_status(
+        self, service_name: str, unit_names: tuple[str, ...]
+    ) -> PiVpnServiceStatus:
+        for unit_name in unit_names:
+            status = self._unit_status(service_name, unit_name)
+            if status.label != "Not found":
+                return status
+
+        checked = ", ".join(unit_names)
+        return PiVpnServiceStatus(
+            "Not found",
+            "badge-neutral",
+            f"Checked common units: {checked}",
+        )
+
+    def _unit_status(self, service_name: str, unit_name: str) -> PiVpnServiceStatus:
+        try:
+            properties = self._parse_systemd_properties(
+                self._systemd_adapter.show_properties(
+                    unit_name,
+                    "LoadState",
+                    "ActiveState",
+                    "SubState",
+                )
+            )
+        except Exception:
+            return PiVpnServiceStatus(
+                "Unknown",
+                "badge-warning",
+                f"Could not read {unit_name}.",
+                unit_name,
+            )
+
+        load_state = properties.get("LoadState", "")
+        active_state = properties.get("ActiveState", "")
+        sub_state = properties.get("SubState", "")
+        if load_state == "not-found":
+            return PiVpnServiceStatus("Not found", "badge-neutral", f"{unit_name} is not loaded.")
+        if active_state == "active":
+            return PiVpnServiceStatus(
+                "Active",
+                "badge-success",
+                f"{service_name} is running through {unit_name}.",
+                unit_name,
+            )
+        if active_state == "inactive":
+            return PiVpnServiceStatus(
+                "Inactive",
+                "badge-warning",
+                f"{unit_name} is installed but not running.",
+                unit_name,
+            )
+        if active_state == "failed":
+            return PiVpnServiceStatus(
+                "Failed",
+                "badge-danger",
+                f"{unit_name} is in failed state.",
+                unit_name,
+            )
+        state_text = " / ".join(part for part in (load_state, active_state, sub_state) if part)
+        return PiVpnServiceStatus(
+            "Unknown",
+            "badge-warning",
+            state_text or f"{unit_name} returned no status.",
+            unit_name,
+        )
+
+    @staticmethod
+    def _parse_systemd_properties(output: str) -> dict[str, str]:
+        properties: dict[str, str] = {}
+        for line in output.splitlines():
+            key, separator, value = line.partition("=")
+            if separator:
+                properties[key] = value
+        return properties

templates/base.html

@@ -59,6 +59,9 @@
                     <i class="fas fa-hard-drive fa-fw"></i> Drive Health
                 </a>
                 {% if is_admin %}
+                <a href="{{ url_for('pivpn_routes.pivpn_page') }}" class="nav-item {% if request.endpoint == 'pivpn_routes.pivpn_page' %}active{% endif %}">
+                    <i class="fas fa-shield-halved fa-fw"></i> PiVPN
+                </a>
                 <a href="{{ url_for('ddns_routes.ddns_page') }}" class="nav-item {% if request.endpoint == 'ddns_routes.ddns_page' %}active{% endif %}">
                     <i class="fas fa-globe fa-fw"></i> DDNS
                 </a>