wrolpi/controller/api/schemas.py
lrnselfreliance d701846254
Some checks are pending
CodeQL / Analyze (push) Waiting to run
Controller Tests / controller-tests (push) Waiting to run
Jest Tests / jest-tests (push) Waiting to run
Hotspot captive portal: welcome page with the WROLPi's address (#709)
* Hotspot captive portal: welcome page with the WROLPi's address

Devices that join the hotspot probe a fixed plain-HTTP URL to check for
Internet access.  NetworkManager's shared-mode dnsmasq now resolves those
probe hostnames (and `wrolpi`) to the hotspot address via an include the
Controller writes when the hotspot starts, the probes land on the Controller
(port 80), and are redirected to /portal: "This is your WROLPi, its address is
10.42.0.1, open WROLPi / open the Controller".

Only the probe hostnames are hijacked so the hotspot keeps forwarding
everything else when the WROLPi has an uplink.  The include is corrected and
the connection re-activated once if NetworkManager picked a non-default
subnet.  Writing the include is best-effort; the hotspot starts regardless.

`hotspot.captive_portal` (controller.yaml, default true) toggles the probe
redirects and the include; like the other hotspot settings it applies on the
next hotspot start, so saving while the hotspot is up reports
restart_required and both Controller UIs say so.  The status endpoint gains
the hotspot address and portal URL, shown in both UIs.  repair.sh only
creates the include directory so a disabled setting survives upgrades.
Native only: Docker has no hotspot.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: address review

* React: re-read the hotspot address whenever the HotspotToggle starts or
  stops the hotspot, matching the fallback UI's refresh-on-status.
* A WPA3 start that reverts to WPA2 leaves the hotspot up, so it now gets
  the captive portal address correction too.
* The corrective `nmcli connection up` logs a warning on a non-zero exit;
  dnsmasq would otherwise keep the old address silently.
* HotspotStatusResponse.captive_portal defaults to true like everywhere else.
* Trim design narrative from the module docstring and React comments.
* The include-before-activate test now observes the include at the moment
  of the `nmcli ... hotspot` call instead of inferring from call order.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: never forward any record type for probe hosts

`address=` only synthesizes A/AAAA answers.  iOS also asks for the HTTPS
record type; dnsmasq forwarded that upstream, Apple's answer aliased to
captive.g.aaplimg.com, and the phone fetched the genuine "Success" from
Apple instead of the portal.  Each probe host now also gets `local=`, so
dnsmasq answers every type itself, and the CDN alias is a probe host too.

Observed with dnsmasq query logging on a hotspot with an eth0 uplink.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: let acknowledged clients finish so the OS offers "Done"

A phone's sign-in sheet only shows "Done" once the OS re-probes and gets
the genuine success reply.  The Controller now remembers each client (by
hotspot IP) that has loaded the portal page and answers its later probes
with the reply each OS expects (Apple "Success" HTML, Android 204, Windows
"Microsoft Connect Test", ...).  The set is cleared when the hotspot
starts so a new session shows the page again.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: acknowledge on Continue, not on view

iOS fetches the portal page in the background before opening the sheet,
then re-probes after the sheet's own navigation.  Acknowledging on view
made that re-probe return "Success" as the sheet's content.  Phones only
re-check the network (and offer "Done") after a full-page navigation, so
the welcome page now has a Continue button leading to /portal/continue,
which is the only place a client is acknowledged.  The Continue page says
to tap Done and links to WROLPi and the Controller.

Also hijack clients3.google.com (older Android probe host).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* help: bump wrolpi-help to the captive portal and Speed Test docs

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* captive portal: comments describe the Continue design; clear acks on stop

Three comments still described acknowledging on view, which c72a7f7b
removed.  Acknowledged clients are now also cleared when the hotspot
stops, so a lease reused in a later session sees the portal page.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-13 20:10:02 -06:00

584 lines
26 KiB
Python

"""
Pydantic models for Controller API requests and responses.
"""
from typing import Optional
from pydantic import BaseModel, Field
# ============================================================================
# Health
# ============================================================================
class HealthResponse(BaseModel):
"""Response model for /api/health endpoint."""
status: str = Field(description="Health status of the controller")
version: str = Field(description="Controller version")
docker_mode: bool = Field(description="Whether running in Docker mode")
drive_mounted: bool = Field(description="Whether the primary drive is mounted")
# ============================================================================
# Status - CPU
# ============================================================================
class CpuStatusResponse(BaseModel):
"""Response model for /api/stats/cpu endpoint."""
percent: float = Field(description="CPU usage percentage (0-100)")
frequency_mhz: Optional[int] = Field(default=None, description="Current CPU frequency in MHz")
frequency_max_mhz: Optional[int] = Field(default=None, description="Maximum CPU frequency in MHz")
temperature_c: Optional[float] = Field(default=None, description="CPU temperature in Celsius")
cores: int = Field(description="Number of CPU cores")
fan_rpm: Optional[int] = Field(default=None, description="Cooling fan speed in RPM, if a fan is present")
# ============================================================================
# Status - Memory
# ============================================================================
class MemoryStatusResponse(BaseModel):
"""Response model for /api/stats/memory endpoint."""
total_bytes: int = Field(description="Total memory in bytes")
available_bytes: int = Field(description="Available memory in bytes")
used_bytes: int = Field(description="Used memory in bytes")
percent: float = Field(description="Memory usage percentage (0-100)")
total_gb: float = Field(description="Total memory in GB")
used_gb: float = Field(description="Used memory in GB")
available_gb: float = Field(description="Available memory in GB")
# ============================================================================
# Status - Load
# ============================================================================
class LoadStatusResponse(BaseModel):
"""Response model for /api/stats/load endpoint."""
load_1min: float = Field(description="1-minute load average")
load_5min: float = Field(description="5-minute load average")
load_15min: float = Field(description="15-minute load average")
# ============================================================================
# Status - Drives
# ============================================================================
class DriveStatusResponse(BaseModel):
"""Response model for individual drive status."""
device: str = Field(description="Device path (e.g., /dev/sda1)")
mount_point: str = Field(description="Mount point path")
fstype: str = Field(description="Filesystem type")
total_bytes: int = Field(description="Total space in bytes")
used_bytes: int = Field(description="Used space in bytes")
free_bytes: int = Field(description="Free space in bytes")
percent: float = Field(description="Usage percentage (0-100)")
total_gb: float = Field(description="Total space in GB")
used_gb: float = Field(description="Used space in GB")
free_gb: float = Field(description="Free space in GB")
class PrimaryDriveStatusResponse(BaseModel):
"""Response model for /api/stats/drives/primary endpoint."""
mounted: bool = Field(description="Whether the primary drive is mounted")
device: Optional[str] = Field(default=None, description="Device path")
mount_point: Optional[str] = Field(default=None, description="Mount point path")
fstype: Optional[str] = Field(default=None, description="Filesystem type")
total_bytes: Optional[int] = Field(default=None, description="Total space in bytes")
used_bytes: Optional[int] = Field(default=None, description="Used space in bytes")
free_bytes: Optional[int] = Field(default=None, description="Free space in bytes")
percent: Optional[float] = Field(default=None, description="Usage percentage")
total_gb: Optional[float] = Field(default=None, description="Total space in GB")
used_gb: Optional[float] = Field(default=None, description="Used space in GB")
free_gb: Optional[float] = Field(default=None, description="Free space in GB")
# ============================================================================
# Status - Network
# ============================================================================
class NetworkInterfaceResponse(BaseModel):
"""Response model for individual network interface."""
name: str = Field(description="Interface name (e.g., eth0, wlan0)")
ipv4: Optional[str] = Field(default=None, description="IPv4 address")
ipv6: Optional[str] = Field(default=None, description="IPv6 address")
mac: Optional[str] = Field(default=None, description="MAC address")
bytes_sent: int = Field(description="Total bytes sent")
bytes_recv: int = Field(description="Total bytes received")
bytes_sent_mb: float = Field(description="Total MB sent")
bytes_recv_mb: float = Field(description="Total MB received")
# ============================================================================
# Status - Power
# ============================================================================
class PowerStatusResponse(BaseModel):
"""Response model for /api/stats/power endpoint."""
undervoltage_detected: bool = Field(description="Currently experiencing undervoltage")
currently_throttled: bool = Field(description="Currently throttled due to voltage/temp")
undervoltage_occurred: bool = Field(description="Undervoltage has occurred since boot")
throttling_occurred: bool = Field(description="Throttling has occurred since boot")
# ============================================================================
# Status - Full
# ============================================================================
class FullStatusResponse(BaseModel):
"""Response model for /api/stats endpoint."""
cpu: CpuStatusResponse = Field(description="CPU status")
memory: MemoryStatusResponse = Field(description="Memory status")
load: LoadStatusResponse = Field(description="System load")
drives: list[DriveStatusResponse] = Field(description="All mounted drives")
primary_drive: Optional[DriveStatusResponse] = Field(default=None, description="Primary WROLPi drive")
network: list[NetworkInterfaceResponse] = Field(description="Network interfaces")
power: PowerStatusResponse = Field(description="Power status")
# ============================================================================
# Admin - Hotspot
# ============================================================================
class HotspotStatusResponse(BaseModel):
"""Response model for /api/hotspot/status endpoint."""
enabled: bool = Field(description="Whether hotspot is currently enabled")
available: bool = Field(description="Whether hotspot functionality is available")
reason: Optional[str] = Field(default=None, description="Reason if unavailable")
ssid: Optional[str] = Field(default=None, description="Hotspot SSID when enabled")
device: Optional[str] = Field(default=None, description="WiFi device name")
ip: Optional[str] = Field(default=None, description="Hotspot IPv4 address when enabled")
captive_portal: bool = Field(default=True, description="Whether the captive portal is enabled")
portal_url: Optional[str] = Field(default=None, description="Landing page URL when the hotspot is up")
class HotspotDevicesResponse(BaseModel):
"""Response model for /api/hotspot/devices endpoint."""
devices: list[str] = Field(description="Available WiFi device names")
class HotspotProtocolsResponse(BaseModel):
"""Response model for /api/hotspot/protocols endpoint."""
device: str = Field(description="WiFi device name that was probed")
protocols: list[str] = Field(description="Hotspot protocols the device can host (wpa2, wpa3)")
class HotspotSettingsRequest(BaseModel):
"""Request model for updating hotspot settings; only provided values are changed."""
device: Optional[str] = Field(default=None, description="WiFi device name")
ssid: Optional[str] = Field(default=None, description="Hotspot SSID")
password: Optional[str] = Field(default=None, description="Hotspot password (8 to 63 characters)")
protocol: Optional[str] = Field(default=None, description="Hotspot protocol (wpa2 or wpa3)")
captive_portal: Optional[bool] = Field(default=None,
description="Show a landing page to devices that join the hotspot")
class HotspotSettingsResponse(BaseModel):
"""Response model for /api/hotspot/settings endpoint."""
device: str = Field(description="WiFi device name")
ssid: str = Field(description="Hotspot SSID")
password: str = Field(description="Hotspot password")
protocol: str = Field(description="Hotspot protocol (wpa2 or wpa3)")
captive_portal: bool = Field(default=True, description="Show a landing page to devices that join the hotspot")
restart_required: bool = Field(default=False,
description="A saved change only applies after the hotspot is restarted")
class HotspotActionResponse(BaseModel):
"""Response model for hotspot start/stop actions."""
success: bool = Field(description="Whether the action succeeded")
ssid: Optional[str] = Field(default=None, description="Hotspot SSID")
device: Optional[str] = Field(default=None, description="WiFi device name")
error: Optional[str] = Field(default=None, description="Error message if failed")
# ============================================================================
# Admin - Bluetooth
# ============================================================================
class BluetoothStatusResponse(BaseModel):
"""Response model for /api/bluetooth/status endpoint."""
enabled: bool = Field(description="Whether Bluetooth radio is currently on")
available: bool = Field(description="Whether Bluetooth functionality is available")
reason: Optional[str] = Field(default=None, description="Reason if unavailable")
class BluetoothActionResponse(BaseModel):
"""Response model for Bluetooth block/unblock actions."""
success: bool = Field(description="Whether the action succeeded")
error: Optional[str] = Field(default=None, description="Error message if failed")
# ============================================================================
# Admin - Desktop
# ============================================================================
class DesktopStatusResponse(BaseModel):
"""Response model for /api/desktop/status endpoint."""
running: bool = Field(description="Whether the display manager is currently running")
available: bool = Field(description="Whether a display manager is installed")
reason: Optional[str] = Field(default=None, description="Reason if unavailable")
class DesktopActionResponse(BaseModel):
"""Response model for desktop start/stop actions."""
success: bool = Field(description="Whether the action succeeded")
switched_to_console: Optional[bool] = Field(
default=None, description="Whether stopping switched the display to a console terminal")
error: Optional[str] = Field(default=None, description="Error message if failed")
# ============================================================================
# Admin - VNC
# ============================================================================
class VncStatusResponse(BaseModel):
"""Response model for /api/vnc/status endpoint."""
running: bool = Field(description="Whether the VNC server is currently running")
available: bool = Field(description="Whether a usable VNC server is installed")
desktop_running: bool = Field(description="Whether the desktop VNC would serve is running")
can_start: bool = Field(description="Whether VNC can be started right now")
reason: Optional[str] = Field(default=None, description="Why VNC is unavailable or cannot start")
backend: Optional[str] = Field(default=None, description="VNC backend in use (wayvnc or realvnc)")
port: int = Field(description="Port the VNC server listens on")
class VncActionResponse(BaseModel):
"""Response model for VNC start/stop actions."""
success: bool = Field(description="Whether the action succeeded")
error: Optional[str] = Field(default=None, description="Error message if failed")
# ============================================================================
# Admin - Samba
# ============================================================================
class SambaShareInfo(BaseModel):
"""Information about a single Samba share."""
name: str = Field(description="Share name")
path: str = Field(description="Filesystem path being shared")
read_only: bool = Field(description="Whether share is read-only")
comment: Optional[str] = Field(default="", description="Share description")
class SambaStatusResponse(BaseModel):
"""Response model for /api/samba/status endpoint."""
enabled: bool = Field(description="Whether Samba is currently running")
available: bool = Field(description="Whether Samba functionality is available")
reason: Optional[str] = Field(default=None, description="Reason if unavailable")
shares: list[SambaShareInfo] = Field(default_factory=list, description="Configured shares")
class SambaShareAddRequest(BaseModel):
"""Request model for adding a Samba share."""
name: str = Field(description="Share name (alphanumeric, hyphens, underscores)")
path: str = Field(description="Filesystem path to share")
read_only: bool = Field(default=True, description="Whether share is read-only")
comment: Optional[str] = Field(default="", description="Share description")
class SambaShareResponse(BaseModel):
"""Response model for share add/update actions."""
success: bool = Field(description="Whether the action succeeded")
share: Optional[SambaShareInfo] = Field(default=None, description="The share")
error: Optional[str] = Field(default=None, description="Error message if failed")
class SambaShareRemoveResponse(BaseModel):
"""Response model for removing a Samba share."""
success: bool = Field(description="Whether the share was removed")
error: Optional[str] = Field(default=None, description="Error message if failed")
# ============================================================================
# Admin - Throttle
# ============================================================================
class ThrottleStatusResponse(BaseModel):
"""Response model for /api/throttle/status endpoint."""
enabled: bool = Field(description="Whether CPU throttle (powersave) is enabled")
available: bool = Field(description="Whether throttle control is available")
reason: Optional[str] = Field(default=None, description="Reason if unavailable")
governor: Optional[str] = Field(default=None, description="Current CPU governor")
available_governors: Optional[list[str]] = Field(default=None, description="Available governors")
class ThrottleActionResponse(BaseModel):
"""Response model for throttle enable/disable actions."""
success: bool = Field(description="Whether the action succeeded")
governor: Optional[str] = Field(default=None, description="New CPU governor")
error: Optional[str] = Field(default=None, description="Error message if failed")
# ============================================================================
# Admin - System Control
# ============================================================================
class SystemActionResponse(BaseModel):
"""Response model for system shutdown/reboot actions."""
success: bool = Field(description="Whether the action was initiated")
message: Optional[str] = Field(default=None, description="Status message")
error: Optional[str] = Field(default=None, description="Error message if failed")
class ServiceRestartResult(BaseModel):
"""Result of restarting a single service."""
success: bool = Field(description="Whether the service restart succeeded")
error: Optional[str] = Field(default=None, description="Error message if failed")
pending: Optional[bool] = Field(default=None, description="Whether restart is pending")
class RestartServicesResponse(BaseModel):
"""Response model for /api/restart endpoint."""
success: bool = Field(description="Whether the restart was initiated")
services: dict[str, ServiceRestartResult] = Field(description="Results per service")
error: Optional[str] = Field(default=None, description="Error message if failed")
# ============================================================================
# Admin - Timezone
# ============================================================================
class TimezoneStatusResponse(BaseModel):
"""Response model for /api/timezone/status endpoint."""
available: bool = Field(description="Whether system timezone control is available")
timezone: Optional[str] = Field(default=None, description="Current system timezone (IANA)")
reason: Optional[str] = Field(default=None, description="Reason if unavailable")
class TimezoneSetRequest(BaseModel):
"""Request model for setting the system timezone."""
timezone: str = Field(description="IANA timezone string (e.g. America/Denver)")
class TimezoneSetResponse(BaseModel):
"""Response model for setting the system timezone."""
success: bool = Field(description="Whether the timezone was set")
timezone: Optional[str] = Field(default=None, description="New timezone if successful")
error: Optional[str] = Field(default=None, description="Error message if failed")
# ============================================================================
# Services - Status
# ============================================================================
class ServiceStatusResponse(BaseModel):
"""Response model for individual service status."""
name: str = Field(description="Service name")
status: str = Field(description="Service status (running, stopped, failed, unknown)")
systemd_name: Optional[str] = Field(default=None, description="Systemd unit name")
container_name: Optional[str] = Field(default=None, description="Docker container name")
active: Optional[str] = Field(default=None, description="Systemd active state")
docker_status: Optional[str] = Field(default=None, description="Docker container status")
enabled: Optional[bool] = Field(default=None, description="Whether service starts at boot")
port: Optional[int] = Field(default=None, description="Service port")
viewable: Optional[bool] = Field(default=False, description="Whether service has a web UI")
view_path: Optional[str] = Field(default="", description="Path to web UI")
use_https: Optional[bool] = Field(default=False, description="Whether service uses HTTPS")
description: Optional[str] = Field(default="", description="Service description")
kind: Optional[str] = Field(default=None, description="Service kind (persistent or task)")
group: Optional[str] = Field(default=None, description="Display group (core or optional)")
available: Optional[bool] = Field(default=None, description="Whether management is available")
reason: Optional[str] = Field(default=None, description="Reason if unavailable")
error: Optional[str] = Field(default=None, description="Error message if any")
class ServiceActionResponse(BaseModel):
"""Response model for service start/stop/restart/enable/disable actions."""
success: bool = Field(description="Whether the action succeeded")
service: str = Field(description="Service name")
action: str = Field(description="Action performed (start, stop, restart, enable, disable)")
error: Optional[str] = Field(default=None, description="Error message if failed")
class ServiceLogsResponse(BaseModel):
"""Response model for service logs."""
service: str = Field(description="Service name")
lines: int = Field(description="Number of lines requested")
since: Optional[str] = Field(default=None, description="Time filter if specified")
logs: str = Field(description="Log content")
class ServicesListErrorResponse(BaseModel):
"""Response model when services list fails."""
error: str = Field(description="Error message")
reason: Optional[str] = Field(default=None, description="Detailed reason")
# ============================================================================
# Scripts
# ============================================================================
class ScriptParameter(BaseModel):
"""Definition of a script input parameter."""
name: str = Field(description="Parameter name (used as env var name)")
label: str = Field(description="Display label for the input field")
type: str = Field(description="Parameter type: 'branch' or 'text'")
required: bool = Field(default=False, description="Whether the parameter must have a value")
class ScriptInfo(BaseModel):
"""Information about an available script."""
name: str = Field(description="Script identifier")
display_name: str = Field(description="Human-readable script name")
description: str = Field(description="What the script does")
warnings: list[str] = Field(description="Warnings to show before running")
available: bool = Field(description="Whether the script can be run")
parameters: list[ScriptParameter] = Field(default_factory=list, description="Input parameters for the script")
class ScriptStatusResponse(BaseModel):
"""Response model for script running status."""
running: bool = Field(description="Whether a script is currently running")
script_name: Optional[str] = Field(default=None, description="Name of running script")
service_name: Optional[str] = Field(default=None, description="Systemd service name")
started_at: Optional[str] = Field(default=None, description="ISO timestamp when started")
elapsed_seconds: Optional[int] = Field(default=None, description="Seconds since start")
class ScriptStartResponse(BaseModel):
"""Response model for starting a script."""
success: bool = Field(description="Whether the script started successfully")
message: Optional[str] = Field(default=None, description="Success message")
error: Optional[str] = Field(default=None, description="Error message if failed")
class ScriptOutputResponse(BaseModel):
"""Response model for script output."""
output: str = Field(description="Script log output")
lines: int = Field(description="Number of lines requested")
script_name: str = Field(description="Script name")
class ScriptStartRequest(BaseModel):
"""Request model for starting a script."""
params: Optional[dict[str, str]] = Field(default=None, description="Parameter values (e.g., {'branch': 'release'})")
class BranchResponse(BaseModel):
"""Response model for current git branch."""
branch: Optional[str] = Field(description="Current git branch, or null if unavailable")
# ============================================================================
# Readiness
# ============================================================================
class ReadyResponse(BaseModel):
"""Response model for /api/ready endpoint."""
api: bool = Field(description="Whether the main WROLPi API is responding")
app: bool = Field(description="Whether the React app is responding")
# ============================================================================
# Onboarding
# ============================================================================
class OnboardingCandidateResponse(BaseModel):
"""A drive that could be selected as the primary WROLPi drive."""
path: str = Field(description="Device path (e.g., /dev/sda1)")
name: str = Field(description="Device name (e.g., sda1)")
size: str = Field(description="Human-readable size (e.g., 1.8T)")
fstype: Optional[str] = Field(default=None, description="Filesystem type")
label: Optional[str] = Field(default=None, description="Filesystem label")
uuid: Optional[str] = Field(default=None, description="Filesystem UUID")
model: Optional[str] = Field(default=None, description="Drive model")
mountpoint: Optional[str] = Field(default=None, description="Current mount point if auto-mounted (e.g., /media/pi/MyDrive)")
class OnboardingProbeRequest(BaseModel):
"""Request to probe a drive for WROLPi configuration."""
device_path: str = Field(description="Device path to probe (e.g., /dev/sda1)")
fstype: str = Field(description="Filesystem type (e.g., ext4)")
class OnboardingProbeResponse(BaseModel):
"""Result of probing a drive for WROLPi configuration."""
config_found: bool = Field(description="Whether a controller.yaml was found on the drive")
mounts: list = Field(default_factory=list, description="Mount entries from config if found")
device_path: str = Field(description="Device path that was probed")
fstype: str = Field(description="Filesystem type")
class OnboardingCommitRequest(BaseModel):
"""Request to execute onboarding setup."""
device_path: str = Field(description="Device path to use as primary drive")
fstype: str = Field(description="Filesystem type")
force_shadowed: bool = Field(
default=False,
description="Proceed even if existing data at the mount target would be hidden by the new mount",
)
class ShadowedDataInfo(BaseModel):
"""Information about data that would be hidden by a mount."""
size_bytes: int = Field(description="Total size of data that would be shadowed")
entries: list[str] = Field(description="Top-level entries that would be hidden")
class OnboardingCommitResponse(BaseModel):
"""Result of onboarding setup."""
success: bool = Field(description="Whether onboarding completed successfully")
error: Optional[str] = Field(default=None, description="Error message if failed")
mounts: list[str] = Field(default_factory=list, description="Mount points that were mounted")
repair_started: bool = Field(default=False, description="Whether repair script was started")
needs_force: Optional[str] = Field(
default=None,
description="If set, the request was soft-blocked. Re-submit with the matching force flag. "
"Currently: 'shadowed'.",
)
shadowed_data: Optional[ShadowedDataInfo] = Field(
default=None,
description="Populated when needs_force='shadowed'.",
)