a9acdb85ce
Replace PM3 compile-from-source in pi-gen with pre-built tarball extraction (saves 43-58 min). Merge stagePM3 into stageDangerousPi as 02-pm3-install substage, renumber all subsequent substages. Switch CI PM3 build to native ARM64 runner (ubuntu-24.04-arm64) eliminating QEMU overhead. Add weekly base-image workflow for pre-baking stages 0-2. Support PM3_TARBALL, BASE_IMAGE, and APT_PROXY env vars in build-image.sh. Also includes prior Phase 5 work: theme system, design system integration, component update system, OS updates, CI build pipeline, and test results. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1205 lines
36 KiB
Python
1205 lines
36 KiB
Python
"""System API endpoints.
|
|
|
|
Refactored to use services for business logic.
|
|
Session management uses PM3Service, system operations use SystemService.
|
|
"""
|
|
import json
|
|
from pathlib import Path
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel
|
|
from typing import Optional, Dict, List
|
|
|
|
from .. import config
|
|
from ..services.container import container
|
|
from ..managers.ups_manager import get_ups_manager
|
|
from ..managers.ble_manager import get_ble_manager
|
|
from ..managers.os_update_manager import get_os_update_manager
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class CreateSessionRequest(BaseModel):
|
|
force_takeover: bool = False
|
|
device_id: Optional[str] = None # Device to create session for
|
|
|
|
|
|
class CreateSessionResponse(BaseModel):
|
|
success: bool
|
|
session_id: Optional[str] = None
|
|
device_id: Optional[str] = None
|
|
error: Optional[str] = None
|
|
|
|
|
|
class SessionInfo(BaseModel):
|
|
session_id: str
|
|
device_id: Optional[str]
|
|
client_ip: str
|
|
created_at: float
|
|
last_activity: float
|
|
time_remaining: float
|
|
|
|
|
|
class CPUCoreInfo(BaseModel):
|
|
"""Per-core CPU information."""
|
|
core_id: int
|
|
percent: float
|
|
online: bool = True # Whether this core is currently enabled
|
|
|
|
|
|
class CPUInfo(BaseModel):
|
|
"""CPU information including per-core data."""
|
|
count: int
|
|
percent: float
|
|
temperature: Optional[float] = None
|
|
per_core: list[CPUCoreInfo] = []
|
|
load_average: Optional[list[float]] = None
|
|
|
|
|
|
class SystemInfo(BaseModel):
|
|
"""System information response."""
|
|
hostname: str
|
|
uptime: float
|
|
cpu_temp: Optional[float]
|
|
cpu: Optional[CPUInfo] = None
|
|
memory_used: float
|
|
memory_total: float
|
|
disk_used: float
|
|
disk_total: float
|
|
|
|
|
|
@router.post("/session/create", response_model=CreateSessionResponse)
|
|
async def create_session(request: Request, body: CreateSessionRequest):
|
|
"""Create a new session for PM3 access.
|
|
|
|
Uses PM3Service for session management.
|
|
Optionally specify device_id for multi-device support.
|
|
"""
|
|
# Get client IP from request
|
|
client_ip = request.client.host if request.client else "unknown"
|
|
user_agent = request.headers.get("user-agent")
|
|
|
|
result = await container.pm3_service.create_session(
|
|
client_ip=client_ip,
|
|
user_agent=user_agent,
|
|
force_takeover=body.force_takeover,
|
|
device_id=body.device_id
|
|
)
|
|
|
|
if result.success:
|
|
return CreateSessionResponse(
|
|
success=True,
|
|
session_id=result.data["session_id"],
|
|
device_id=result.data.get("device_id"),
|
|
error=None
|
|
)
|
|
else:
|
|
return CreateSessionResponse(
|
|
success=False,
|
|
session_id=None,
|
|
device_id=None,
|
|
error=result.error.message
|
|
)
|
|
|
|
|
|
@router.post("/session/{session_id}/release")
|
|
async def release_session(session_id: str, device_id: Optional[str] = None):
|
|
"""Release an active session.
|
|
|
|
Uses PM3Service for session management.
|
|
|
|
Args:
|
|
session_id: Session ID to release
|
|
device_id: Optional device ID for faster lookup
|
|
"""
|
|
result = await container.pm3_service.release_session(session_id, device_id)
|
|
|
|
if not result.success:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=result.error.message
|
|
)
|
|
|
|
return {"success": True, "message": result.data["message"]}
|
|
|
|
|
|
@router.get("/session/active", response_model=Optional[SessionInfo])
|
|
async def get_active_session(device_id: Optional[str] = None):
|
|
"""Get information about the active session for a device.
|
|
|
|
Uses PM3Service (via SessionManager) for session info.
|
|
|
|
Args:
|
|
device_id: Optional device ID. If None, returns any active session.
|
|
"""
|
|
# Access session manager through container for consistency
|
|
session = container.session_manager.get_active_session(device_id)
|
|
|
|
if not session:
|
|
return None
|
|
|
|
import time
|
|
from .. import config
|
|
|
|
time_remaining = config.SESSION_TIMEOUT - (time.time() - session.last_activity)
|
|
|
|
return SessionInfo(
|
|
session_id=session.session_id,
|
|
device_id=session.device_id,
|
|
client_ip=session.client_ip,
|
|
created_at=session.created_at,
|
|
last_activity=session.last_activity,
|
|
time_remaining=max(0, time_remaining)
|
|
)
|
|
|
|
|
|
@router.get("/sessions/all")
|
|
async def get_all_sessions():
|
|
"""Get all active sessions across all devices.
|
|
|
|
Returns a list of all active sessions with their device IDs.
|
|
"""
|
|
result = container.pm3_service.get_all_sessions()
|
|
return result.data
|
|
|
|
|
|
@router.get("/info", response_model=SystemInfo)
|
|
async def get_system_info():
|
|
"""Get system information.
|
|
|
|
Refactored to use SystemService for business logic.
|
|
"""
|
|
from ..services.container import container
|
|
|
|
result = await container.system_service.get_info()
|
|
|
|
if not result.success:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=result.error.message
|
|
)
|
|
|
|
cpu_data = result.data["cpu"]
|
|
memory_data = result.data["memory"]
|
|
disk_data = result.data["disk"]
|
|
|
|
# Build per-core CPU info
|
|
per_core = [
|
|
CPUCoreInfo(
|
|
core_id=c["core_id"],
|
|
percent=c["percent"],
|
|
online=c.get("online", True)
|
|
)
|
|
for c in cpu_data.get("per_core", [])
|
|
]
|
|
|
|
cpu_info = CPUInfo(
|
|
count=cpu_data.get("count", 0),
|
|
percent=cpu_data.get("percent", 0.0),
|
|
temperature=cpu_data.get("temperature"),
|
|
per_core=per_core,
|
|
load_average=cpu_data.get("load_average")
|
|
)
|
|
|
|
return SystemInfo(
|
|
hostname=result.data["hostname"],
|
|
uptime=result.data["uptime"],
|
|
cpu_temp=cpu_data.get("temperature"),
|
|
cpu=cpu_info,
|
|
memory_used=memory_data["used"],
|
|
memory_total=memory_data["total"],
|
|
disk_used=disk_data["used"],
|
|
disk_total=disk_data["total"]
|
|
)
|
|
|
|
|
|
@router.get("/config")
|
|
async def get_config():
|
|
"""Get system configuration (non-sensitive values)."""
|
|
from .. import config
|
|
from ..services.container import container
|
|
|
|
# Get WiFi mode from manager, default to AP mode
|
|
wifi_mode = container.wifi_manager._current_mode.value if container.wifi_manager else "ap"
|
|
|
|
return {
|
|
"pm3_device": config.PM3_DEVICE,
|
|
"session_timeout": config.SESSION_TIMEOUT,
|
|
"wifi_mode": wifi_mode,
|
|
"ble_enabled": config.BLE_ENABLED,
|
|
"auth_enabled": config.AUTH_ENABLED,
|
|
"https_enabled": config.HTTPS_ENABLED
|
|
}
|
|
|
|
|
|
@router.post("/restart")
|
|
async def restart_system(delay: int = 0):
|
|
"""Restart the system.
|
|
|
|
Refactored to use SystemService for business logic.
|
|
|
|
Args:
|
|
delay: Delay in seconds before restart (default: 0)
|
|
"""
|
|
from ..services.container import container
|
|
|
|
result = await container.system_service.restart(delay=delay)
|
|
|
|
if not result.success:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=result.error.message
|
|
)
|
|
|
|
return {"success": True, "message": result.data["message"]}
|
|
|
|
|
|
@router.post("/shutdown")
|
|
async def shutdown_system(delay: int = 0):
|
|
"""Initiate system shutdown.
|
|
|
|
Refactored to use SystemService for business logic.
|
|
|
|
Args:
|
|
delay: Delay in seconds before shutdown (default: 0)
|
|
"""
|
|
from ..services.container import container
|
|
|
|
result = await container.system_service.shutdown(delay=delay)
|
|
|
|
if not result.success:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=result.error.message
|
|
)
|
|
|
|
return {"success": True, "message": result.data["message"]}
|
|
|
|
|
|
class UPSStatusResponse(BaseModel):
|
|
"""UPS status response model."""
|
|
battery_percentage: float
|
|
voltage: float
|
|
current: float
|
|
power_source: str
|
|
battery_status: str
|
|
time_remaining: Optional[int] = None
|
|
temperature: Optional[float] = None
|
|
last_updated: Optional[str] = None
|
|
is_available: bool
|
|
error_message: Optional[str] = None
|
|
|
|
|
|
class UPSThresholdsRequest(BaseModel):
|
|
"""Request to set UPS battery thresholds."""
|
|
shutdown_threshold: Optional[float] = None
|
|
warning_threshold: Optional[float] = None
|
|
|
|
|
|
class PowerRestrictionsResponse(BaseModel):
|
|
"""Power restrictions response model."""
|
|
restricted: bool
|
|
reason: Optional[str] = None
|
|
power_source: str
|
|
ups_available: bool
|
|
battery_percentage: Optional[float] = None
|
|
allow_firmware_flash: bool
|
|
allow_bootloader_flash: bool
|
|
allow_intensive_operations: bool
|
|
message: Optional[str] = None
|
|
warning: Optional[str] = None
|
|
shutdown_imminent: Optional[bool] = None
|
|
|
|
|
|
@router.get("/ups/status", response_model=UPSStatusResponse)
|
|
async def get_ups_status():
|
|
"""Get current UPS battery status."""
|
|
ups_manager = get_ups_manager()
|
|
status = await ups_manager.get_status()
|
|
|
|
return UPSStatusResponse(
|
|
battery_percentage=status.battery_percentage,
|
|
voltage=status.voltage,
|
|
current=status.current,
|
|
power_source=status.power_source.value,
|
|
battery_status=status.battery_status.value,
|
|
time_remaining=status.time_remaining,
|
|
temperature=status.temperature,
|
|
last_updated=status.last_updated,
|
|
is_available=status.is_available,
|
|
error_message=status.error_message
|
|
)
|
|
|
|
|
|
@router.post("/ups/thresholds")
|
|
async def set_ups_thresholds(request: UPSThresholdsRequest):
|
|
"""Set UPS battery thresholds for warnings and shutdown."""
|
|
ups_manager = get_ups_manager()
|
|
|
|
if request.shutdown_threshold is not None:
|
|
ups_manager.set_shutdown_threshold(request.shutdown_threshold)
|
|
|
|
if request.warning_threshold is not None:
|
|
ups_manager.set_warning_threshold(request.warning_threshold)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "UPS thresholds updated"
|
|
}
|
|
|
|
|
|
@router.post("/ups/shutdown")
|
|
async def trigger_ups_shutdown(delay: int = 30):
|
|
"""Trigger safe shutdown via UPS manager.
|
|
|
|
Args:
|
|
delay: Delay in seconds before shutdown (default: 30)
|
|
"""
|
|
ups_manager = get_ups_manager()
|
|
await ups_manager.shutdown_device(delay=delay)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": f"Shutdown initiated with {delay}s delay"
|
|
}
|
|
|
|
|
|
@router.get("/power/restrictions", response_model=PowerRestrictionsResponse)
|
|
async def get_power_restrictions():
|
|
"""Get current power restrictions based on UPS/battery state.
|
|
|
|
Returns power policy information including:
|
|
- Whether operations are restricted
|
|
- Current power source (AC, battery, or assumed AC if no UPS)
|
|
- Battery level (if UPS present)
|
|
- Which operations are allowed (firmware flash, bootloader flash, etc.)
|
|
- User-friendly messages and warnings
|
|
|
|
This endpoint is critical for determining whether power-intensive
|
|
operations (like firmware flashing) should be allowed.
|
|
|
|
If UPS hardware is not detected, assumes stable AC power and allows
|
|
all operations (user responsibility to ensure power stability).
|
|
"""
|
|
ups_manager = get_ups_manager()
|
|
restrictions = ups_manager.get_power_restrictions()
|
|
|
|
return PowerRestrictionsResponse(**restrictions)
|
|
|
|
|
|
class PiModelResponse(BaseModel):
|
|
"""Pi model information response."""
|
|
model_config = {"protected_namespaces": ()} # Allow model_* field names
|
|
|
|
model: str
|
|
model_short: str
|
|
total_cores: int
|
|
default_active_cores: int
|
|
min_cores: int
|
|
max_cores: int
|
|
|
|
|
|
class CPUCoresConfigResponse(BaseModel):
|
|
"""CPU cores configuration response."""
|
|
total_cores: int
|
|
online_cores: int
|
|
configured_cores: Optional[int] = None # From cmdline.txt maxcpus parameter
|
|
configurable_cores: list[int]
|
|
pi_model: Optional[PiModelResponse] = None
|
|
|
|
|
|
class SetCPUCoresRequest(BaseModel):
|
|
"""Request to set CPU cores."""
|
|
num_cores: int
|
|
persist: bool = True # Save to config for boot persistence
|
|
reboot: bool = True # Reboot after saving
|
|
|
|
|
|
@router.get("/cpu/cores", response_model=CPUCoresConfigResponse)
|
|
async def get_cpu_cores():
|
|
"""Get CPU cores configuration.
|
|
|
|
Returns information about:
|
|
- Total physical cores
|
|
- Currently online cores
|
|
- Which cores can be toggled (core 0 is always on)
|
|
- Pi model information with recommended defaults
|
|
"""
|
|
result = await container.system_service.get_cpu_cores_config()
|
|
|
|
if not result.success:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=result.error.message
|
|
)
|
|
|
|
# Convert pi_model dict to response model if present
|
|
pi_model_data = result.data.get("pi_model")
|
|
pi_model = None
|
|
if pi_model_data:
|
|
pi_model = PiModelResponse(**pi_model_data)
|
|
|
|
return CPUCoresConfigResponse(
|
|
total_cores=result.data["total_cores"],
|
|
online_cores=result.data["online_cores"],
|
|
configured_cores=result.data.get("configured_cores"),
|
|
configurable_cores=result.data["configurable_cores"],
|
|
pi_model=pi_model
|
|
)
|
|
|
|
|
|
@router.post("/cpu/cores")
|
|
async def set_cpu_cores(request: SetCPUCoresRequest):
|
|
"""Set the number of active CPU cores.
|
|
|
|
Core 0 is always active. This endpoint enables/disables cores 1 through N.
|
|
|
|
For Pi Zero 2 W, the recommended default is 2 cores (out of 4) for
|
|
better thermal management and power efficiency.
|
|
|
|
Args:
|
|
num_cores: Number of cores to keep active (1 to max_cores)
|
|
persist: Save setting to config file (default: True)
|
|
reboot: Reboot system after saving (default: True)
|
|
"""
|
|
result = await container.system_service.set_cpu_cores(
|
|
num_cores=request.num_cores,
|
|
persist=request.persist,
|
|
reboot=request.reboot
|
|
)
|
|
|
|
if not result.success:
|
|
raise HTTPException(
|
|
status_code=400 if result.error.code == "invalid_cores" else 500,
|
|
detail=result.error.message
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": result.data["message"],
|
|
"requested": result.data.get("requested"),
|
|
"actual": result.data.get("actual"),
|
|
"rebooting": result.data.get("rebooting", False)
|
|
}
|
|
|
|
|
|
@router.get("/pi/model", response_model=PiModelResponse)
|
|
async def get_pi_model():
|
|
"""Get Raspberry Pi model information.
|
|
|
|
Returns the detected Pi model with recommended CPU core settings.
|
|
"""
|
|
result = await container.system_service.get_pi_model()
|
|
|
|
if not result.success:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=result.error.message
|
|
)
|
|
|
|
return PiModelResponse(**result.data)
|
|
|
|
|
|
class BLEStatusResponse(BaseModel):
|
|
"""BLE status response model."""
|
|
enabled: bool
|
|
available: bool
|
|
advertising: bool
|
|
connected_devices: int
|
|
device_name: str
|
|
queued_notifications: int
|
|
|
|
|
|
class SendNotificationRequest(BaseModel):
|
|
"""Request to send a BLE notification."""
|
|
type: str
|
|
message: str
|
|
data: Optional[Dict] = None
|
|
|
|
|
|
@router.get("/ble/status", response_model=BLEStatusResponse)
|
|
async def get_ble_status():
|
|
"""Get BLE manager status."""
|
|
ble_manager = get_ble_manager()
|
|
status = await ble_manager.get_status()
|
|
|
|
return BLEStatusResponse(**status)
|
|
|
|
|
|
@router.post("/ble/advertising/start")
|
|
async def start_ble_advertising():
|
|
"""Start BLE advertising."""
|
|
ble_manager = get_ble_manager()
|
|
|
|
if not ble_manager.is_available():
|
|
raise HTTPException(status_code=503, detail="BLE not available")
|
|
|
|
await ble_manager.start_advertising()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "BLE advertising started"
|
|
}
|
|
|
|
|
|
@router.post("/ble/advertising/stop")
|
|
async def stop_ble_advertising():
|
|
"""Stop BLE advertising."""
|
|
ble_manager = get_ble_manager()
|
|
await ble_manager.stop_advertising()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "BLE advertising stopped"
|
|
}
|
|
|
|
|
|
@router.post("/ble/notify")
|
|
async def send_ble_notification(request: SendNotificationRequest):
|
|
"""Send a BLE notification to connected devices."""
|
|
ble_manager = get_ble_manager()
|
|
|
|
if not ble_manager.is_available():
|
|
raise HTTPException(status_code=503, detail="BLE not available")
|
|
|
|
from ..managers.ble_manager import NotificationType
|
|
|
|
# Validate notification type
|
|
try:
|
|
notification_type = NotificationType(request.type)
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail=f"Invalid notification type: {request.type}")
|
|
|
|
await ble_manager.send_notification(
|
|
notification_type=notification_type,
|
|
message=request.message,
|
|
data=request.data
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Notification sent"
|
|
}
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# SSL/HTTPS API
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class SSLCertificateInfo(BaseModel):
|
|
"""SSL certificate information."""
|
|
enabled: bool
|
|
certificate_exists: bool
|
|
certificate_path: Optional[str] = None
|
|
key_path: Optional[str] = None
|
|
subject: Optional[str] = None
|
|
issuer: Optional[str] = None
|
|
not_before: Optional[str] = None
|
|
not_after: Optional[str] = None
|
|
san: Optional[list[str]] = None # Subject Alternative Names
|
|
fingerprint_sha256: Optional[str] = None
|
|
|
|
|
|
class SSLRegenerateRequest(BaseModel):
|
|
"""Request to regenerate SSL certificate."""
|
|
reload_nginx: bool = True
|
|
|
|
|
|
@router.get("/ssl/info", response_model=SSLCertificateInfo)
|
|
async def get_ssl_info():
|
|
"""Get SSL certificate information.
|
|
|
|
Returns details about the current SSL configuration including:
|
|
- Whether HTTPS is enabled
|
|
- Certificate existence and paths
|
|
- Certificate subject, issuer, validity dates
|
|
- Subject Alternative Names (SANs)
|
|
- SHA256 fingerprint
|
|
"""
|
|
import os
|
|
import subprocess
|
|
from .. import config
|
|
|
|
cert_path = "/opt/dangerous-pi/ssl/dangerous-pi.crt"
|
|
key_path = "/opt/dangerous-pi/ssl/dangerous-pi.key"
|
|
|
|
# Check if HTTPS is enabled and certificate exists
|
|
https_enabled = config.HTTPS_ENABLED
|
|
cert_exists = os.path.exists(cert_path)
|
|
key_exists = os.path.exists(key_path)
|
|
|
|
if not cert_exists:
|
|
return SSLCertificateInfo(
|
|
enabled=https_enabled,
|
|
certificate_exists=False,
|
|
certificate_path=cert_path,
|
|
key_path=key_path
|
|
)
|
|
|
|
# Parse certificate details using openssl
|
|
cert_info = {
|
|
"subject": None,
|
|
"issuer": None,
|
|
"not_before": None,
|
|
"not_after": None,
|
|
"san": [],
|
|
"fingerprint": None
|
|
}
|
|
|
|
try:
|
|
# Get subject
|
|
result = subprocess.run(
|
|
["openssl", "x509", "-in", cert_path, "-noout", "-subject"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
if result.returncode == 0:
|
|
cert_info["subject"] = result.stdout.strip().replace("subject=", "")
|
|
|
|
# Get issuer
|
|
result = subprocess.run(
|
|
["openssl", "x509", "-in", cert_path, "-noout", "-issuer"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
if result.returncode == 0:
|
|
cert_info["issuer"] = result.stdout.strip().replace("issuer=", "")
|
|
|
|
# Get validity dates
|
|
result = subprocess.run(
|
|
["openssl", "x509", "-in", cert_path, "-noout", "-dates"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
if result.returncode == 0:
|
|
for line in result.stdout.strip().split("\n"):
|
|
if line.startswith("notBefore="):
|
|
cert_info["not_before"] = line.replace("notBefore=", "")
|
|
elif line.startswith("notAfter="):
|
|
cert_info["not_after"] = line.replace("notAfter=", "")
|
|
|
|
# Get SANs
|
|
result = subprocess.run(
|
|
["openssl", "x509", "-in", cert_path, "-noout", "-ext", "subjectAltName"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
if result.returncode == 0 and ("subjectAltName" in result.stdout or "Subject Alternative Name" in result.stdout):
|
|
# Parse SANs from output like "DNS:localhost, IP:192.168.4.1"
|
|
san_line = result.stdout.strip()
|
|
for line in san_line.split("\n"):
|
|
if "DNS:" in line or "IP" in line:
|
|
# Split by comma and clean up
|
|
sans = [s.strip() for s in line.split(",")]
|
|
cert_info["san"] = [s for s in sans if s.startswith(("DNS:", "IP"))]
|
|
break
|
|
|
|
# Get SHA256 fingerprint
|
|
result = subprocess.run(
|
|
["openssl", "x509", "-in", cert_path, "-noout", "-fingerprint", "-sha256"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
if result.returncode == 0:
|
|
# Format: "sha256 Fingerprint=XX:XX:XX..."
|
|
fingerprint = result.stdout.strip()
|
|
if "=" in fingerprint:
|
|
cert_info["fingerprint"] = fingerprint.split("=", 1)[1]
|
|
|
|
except subprocess.TimeoutExpired:
|
|
pass # Return what we have
|
|
except Exception:
|
|
pass # Return what we have
|
|
|
|
return SSLCertificateInfo(
|
|
enabled=https_enabled,
|
|
certificate_exists=cert_exists and key_exists,
|
|
certificate_path=cert_path,
|
|
key_path=key_path,
|
|
subject=cert_info["subject"],
|
|
issuer=cert_info["issuer"],
|
|
not_before=cert_info["not_before"],
|
|
not_after=cert_info["not_after"],
|
|
san=cert_info["san"] if cert_info["san"] else None,
|
|
fingerprint_sha256=cert_info["fingerprint"]
|
|
)
|
|
|
|
|
|
@router.post("/ssl/regenerate")
|
|
async def regenerate_ssl_certificate(request: SSLRegenerateRequest):
|
|
"""Regenerate the self-signed SSL certificate.
|
|
|
|
This will:
|
|
1. Generate a new EC P-256 key and self-signed certificate
|
|
2. Optionally reload nginx to pick up the new certificate
|
|
|
|
The new certificate will be valid for 10 years and include SANs for:
|
|
- 192.168.4.1 (AP gateway IP)
|
|
- dangerous-pi.local (mDNS hostname)
|
|
- localhost
|
|
|
|
Args:
|
|
reload_nginx: Whether to reload nginx after regeneration (default: True)
|
|
|
|
Returns:
|
|
Success status and certificate info
|
|
"""
|
|
import subprocess
|
|
import os
|
|
|
|
# Check multiple possible locations for the script
|
|
script_candidates = [
|
|
"/opt/dangerous-pi/scripts/generate-ssl-cert.sh",
|
|
os.path.expanduser("~/dangerous-pi/scripts/generate-ssl-cert.sh"),
|
|
os.path.join(os.path.dirname(__file__), "../../../scripts/generate-ssl-cert.sh"),
|
|
]
|
|
script_path = None
|
|
for candidate in script_candidates:
|
|
if os.path.exists(candidate):
|
|
script_path = candidate
|
|
break
|
|
|
|
if not script_path:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="SSL certificate generation script not found"
|
|
)
|
|
|
|
try:
|
|
# Run certificate generation with --force flag
|
|
result = subprocess.run(
|
|
[script_path, "--force"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Certificate generation failed: {result.stderr}"
|
|
)
|
|
|
|
# Reload nginx if requested
|
|
nginx_reloaded = False
|
|
if request.reload_nginx:
|
|
try:
|
|
nginx_result = subprocess.run(
|
|
["systemctl", "reload", "nginx"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10
|
|
)
|
|
nginx_reloaded = nginx_result.returncode == 0
|
|
except Exception:
|
|
pass # Non-fatal, certificate was still generated
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "SSL certificate regenerated successfully",
|
|
"nginx_reloaded": nginx_reloaded,
|
|
"output": result.stdout
|
|
}
|
|
|
|
except subprocess.TimeoutExpired:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="Certificate generation timed out"
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Certificate generation error: {str(e)}"
|
|
)
|
|
|
|
|
|
class SSLToggleRequest(BaseModel):
|
|
"""Request to enable or disable HTTPS."""
|
|
enabled: bool
|
|
|
|
|
|
@router.post("/ssl/toggle")
|
|
async def toggle_https(request: SSLToggleRequest):
|
|
"""Enable or disable HTTPS.
|
|
|
|
Updates the HTTPS_ENABLED setting in the .env file and runs
|
|
configure-nginx.sh to switch nginx between HTTP and HTTPS configs.
|
|
Reloads nginx to apply the change immediately.
|
|
|
|
Args:
|
|
enabled: True to enable HTTPS, False to disable
|
|
|
|
Returns:
|
|
Success status and current HTTPS state
|
|
"""
|
|
import subprocess
|
|
import os
|
|
|
|
env_file = os.getenv("ENV_FILE", "/opt/dangerous-pi/.env")
|
|
configure_script = "/opt/dangerous-pi/scripts/configure-nginx.sh"
|
|
|
|
# Update .env file
|
|
try:
|
|
lines = []
|
|
found = False
|
|
if os.path.exists(env_file):
|
|
with open(env_file, "r") as f:
|
|
lines = f.readlines()
|
|
|
|
new_lines = []
|
|
for line in lines:
|
|
if line.strip().startswith("HTTPS_ENABLED="):
|
|
new_lines.append(f"HTTPS_ENABLED={'true' if request.enabled else 'false'}\n")
|
|
found = True
|
|
else:
|
|
new_lines.append(line)
|
|
|
|
if not found:
|
|
new_lines.append(f"HTTPS_ENABLED={'true' if request.enabled else 'false'}\n")
|
|
|
|
with open(env_file, "w") as f:
|
|
f.writelines(new_lines)
|
|
|
|
except PermissionError:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="Cannot write to .env file - check permissions"
|
|
)
|
|
|
|
# Update runtime config
|
|
from .. import config as app_config
|
|
app_config.HTTPS_ENABLED = request.enabled
|
|
|
|
# Run configure-nginx.sh to switch nginx config
|
|
nginx_configured = False
|
|
if os.path.exists(configure_script):
|
|
try:
|
|
result = subprocess.run(
|
|
[configure_script],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
env={**os.environ, "HTTPS_ENABLED": "true" if request.enabled else "false"}
|
|
)
|
|
if result.returncode == 0:
|
|
# Reload nginx
|
|
subprocess.run(
|
|
["systemctl", "reload", "nginx"],
|
|
capture_output=True,
|
|
timeout=10
|
|
)
|
|
nginx_configured = True
|
|
except Exception:
|
|
pass # Non-fatal, .env was still updated
|
|
|
|
return {
|
|
"success": True,
|
|
"enabled": request.enabled,
|
|
"nginx_configured": nginx_configured,
|
|
"message": f"HTTPS {'enabled' if request.enabled else 'disabled'}"
|
|
}
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Header Widgets API
|
|
# -----------------------------------------------------------------------------
|
|
|
|
class WidgetResponse(BaseModel):
|
|
"""Header widget response model."""
|
|
id: str
|
|
source: str
|
|
severity: str
|
|
message: str
|
|
dismissible: bool
|
|
icon: Optional[str] = None
|
|
action_label: Optional[str] = None
|
|
action_url: Optional[str] = None
|
|
created_at: str
|
|
expires_at: Optional[str] = None
|
|
|
|
|
|
@router.get("/widgets", response_model=list[WidgetResponse])
|
|
async def get_header_widgets():
|
|
"""Get all active header widgets.
|
|
|
|
Returns widgets from:
|
|
- System managers (UPS, PM3, Updates)
|
|
- Enabled plugins
|
|
|
|
Widgets are sorted by severity (error > warning > info > success).
|
|
"""
|
|
from ..managers.plugin_manager import get_plugin_manager
|
|
|
|
plugin_manager = get_plugin_manager()
|
|
widgets = plugin_manager.get_active_widgets()
|
|
|
|
# Sort by severity (error first, then warning, info, success)
|
|
severity_order = {"error": 0, "warning": 1, "info": 2, "success": 3}
|
|
widgets.sort(key=lambda w: severity_order.get(w.severity.value, 99))
|
|
|
|
return [
|
|
WidgetResponse(
|
|
id=w.id,
|
|
source=w.source,
|
|
severity=w.severity.value,
|
|
message=w.message,
|
|
dismissible=w.dismissible,
|
|
icon=w.icon,
|
|
action_label=w.action_label,
|
|
action_url=w.action_url,
|
|
created_at=w.created_at or "",
|
|
expires_at=w.expires_at
|
|
)
|
|
for w in widgets
|
|
]
|
|
|
|
|
|
@router.post("/widgets/{widget_id}/dismiss")
|
|
async def dismiss_widget(widget_id: str):
|
|
"""Dismiss a header widget.
|
|
|
|
Dismissed widgets will not reappear until the server restarts
|
|
or the widget is explicitly re-registered.
|
|
|
|
Args:
|
|
widget_id: Full widget ID (e.g., "ups.hardware_missing", "plugin.hello_world.status")
|
|
|
|
Returns:
|
|
Success status
|
|
"""
|
|
from ..managers.plugin_manager import get_plugin_manager
|
|
|
|
plugin_manager = get_plugin_manager()
|
|
success = plugin_manager.dismiss_widget(widget_id)
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Widget '{widget_id}' not found or not dismissible"
|
|
)
|
|
|
|
return {"success": True, "widget_id": widget_id}
|
|
|
|
|
|
@router.post("/widgets/clear-dismissed")
|
|
async def clear_dismissed_widgets():
|
|
"""Clear all dismissed widgets, allowing them to reappear.
|
|
|
|
This is useful if a user wants to see previously dismissed
|
|
notifications again.
|
|
"""
|
|
from ..managers.plugin_manager import get_plugin_manager
|
|
|
|
plugin_manager = get_plugin_manager()
|
|
plugin_manager.clear_dismissed()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OS Updates API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/os/info")
|
|
async def get_os_info():
|
|
"""Get OS-level system information.
|
|
|
|
Returns Debian version, kernel, architecture, uptime, hostname,
|
|
last apt update timestamp, auto-update setting, and reboot-required status.
|
|
"""
|
|
manager = get_os_update_manager()
|
|
return await manager.get_os_info()
|
|
|
|
|
|
@router.get("/os/updates")
|
|
async def get_os_updates(refresh: bool = False):
|
|
"""Get available OS package updates.
|
|
|
|
Returns a list of upgradable packages with current and available versions.
|
|
Results are cached for 1 hour unless refresh=True.
|
|
"""
|
|
manager = get_os_update_manager()
|
|
packages = await manager.check_available_updates(force_refresh=refresh)
|
|
return {
|
|
"count": len(packages),
|
|
"packages": packages,
|
|
"upgrading": manager.is_upgrading,
|
|
}
|
|
|
|
|
|
class OsUpgradeRequest(BaseModel):
|
|
"""Request to trigger an OS package upgrade."""
|
|
security_only: bool = False
|
|
|
|
|
|
@router.post("/os/update")
|
|
async def run_os_update(request: OsUpgradeRequest):
|
|
"""Trigger an OS package upgrade.
|
|
|
|
Runs `apt-get upgrade -y` (or unattended-upgrade --verbose for security-only).
|
|
Only one upgrade can run at a time.
|
|
|
|
Args:
|
|
security_only: If True, only install security updates
|
|
"""
|
|
manager = get_os_update_manager()
|
|
result = await manager.run_upgrade(security_only=request.security_only)
|
|
|
|
if not result.get("success"):
|
|
raise HTTPException(status_code=409 if "already in progress" in result.get("error", "") else 500,
|
|
detail=result.get("error", "Upgrade failed"))
|
|
|
|
return result
|
|
|
|
|
|
class AutoUpdatesRequest(BaseModel):
|
|
"""Request to toggle automatic security updates."""
|
|
enabled: bool
|
|
|
|
|
|
@router.post("/os/auto-updates")
|
|
async def toggle_auto_updates(request: AutoUpdatesRequest):
|
|
"""Toggle automatic security updates.
|
|
|
|
Writes AUTO_SECURITY_UPDATES to the .env file and restarts the
|
|
systemd timer that controls unattended-upgrades.
|
|
|
|
Args:
|
|
enabled: True to enable, False to disable automatic security updates
|
|
"""
|
|
manager = get_os_update_manager()
|
|
result = await manager.toggle_auto_updates(request.enabled)
|
|
|
|
if not result.get("success"):
|
|
raise HTTPException(status_code=500, detail=result.get("error", "Failed to toggle auto-updates"))
|
|
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Theme registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class ThemeDefinitionResponse(BaseModel):
|
|
"""A single theme definition."""
|
|
id: str
|
|
name: str
|
|
description: str = ""
|
|
supportsModes: List[str] = ["dark", "light", "auto"]
|
|
defaultMode: str = "dark"
|
|
author: str = ""
|
|
css_url: str = ""
|
|
source: str = "builtin" # "builtin" or "plugin"
|
|
|
|
|
|
class ThemeRegistryResponse(BaseModel):
|
|
"""Response for GET /api/system/themes."""
|
|
themes: List[ThemeDefinitionResponse]
|
|
|
|
|
|
def _scan_theme_dirs() -> List[ThemeDefinitionResponse]:
|
|
"""Scan themes directories for installed theme packages."""
|
|
themes_found: List[ThemeDefinitionResponse] = []
|
|
|
|
# Search paths: dev + production
|
|
themes_dirs = [
|
|
Path(__file__).parent.parent.parent / "frontend" / "themes", # dev
|
|
Path("/opt/dangerous-pi/app/frontend/themes"), # prod
|
|
]
|
|
|
|
seen_ids: set = set()
|
|
for themes_dir in themes_dirs:
|
|
if not themes_dir.is_dir():
|
|
continue
|
|
for entry in sorted(themes_dir.iterdir()):
|
|
if not entry.is_dir() or entry.name.startswith("."):
|
|
continue
|
|
theme_json = entry / "theme.json"
|
|
if not theme_json.exists():
|
|
continue
|
|
try:
|
|
meta = json.loads(theme_json.read_text())
|
|
theme_id = meta.get("id", entry.name)
|
|
if theme_id in seen_ids:
|
|
continue
|
|
seen_ids.add(theme_id)
|
|
themes_found.append(ThemeDefinitionResponse(
|
|
id=theme_id,
|
|
name=meta.get("name", theme_id),
|
|
description=meta.get("description", ""),
|
|
supportsModes=meta.get("supportsModes", ["dark", "light", "auto"]),
|
|
defaultMode=meta.get("defaultMode", "dark"),
|
|
author=meta.get("author", ""),
|
|
css_url=f"/themes/{theme_id}/tokens.css",
|
|
source="builtin",
|
|
))
|
|
except (json.JSONDecodeError, OSError) as exc:
|
|
print(f"Warning: bad theme.json in {entry}: {exc}")
|
|
|
|
return themes_found
|
|
|
|
|
|
def _scan_plugin_themes() -> List[ThemeDefinitionResponse]:
|
|
"""Collect themes registered by plugins via the theme_register hook."""
|
|
from ..managers.plugin_manager import get_plugin_manager
|
|
pm = get_plugin_manager()
|
|
|
|
themes: List[ThemeDefinitionResponse] = []
|
|
if "theme_register" not in pm._hooks:
|
|
return themes
|
|
|
|
import asyncio
|
|
results = []
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
loop = None
|
|
|
|
# Hooks are called synchronously here since they are simple data returns
|
|
for callback in pm._hooks.get("theme_register", []):
|
|
try:
|
|
if asyncio.iscoroutinefunction(callback):
|
|
# Schedule in running loop if available
|
|
if loop:
|
|
import concurrent.futures
|
|
# Can't await in sync context; skip async hooks
|
|
continue
|
|
else:
|
|
continue
|
|
result = callback()
|
|
if result:
|
|
results.append(result)
|
|
except Exception as exc:
|
|
print(f"Warning: theme_register hook error: {exc}")
|
|
|
|
for r in results:
|
|
themes.append(ThemeDefinitionResponse(
|
|
id=r.get("id", "unknown"),
|
|
name=r.get("name", "Unknown Theme"),
|
|
description=r.get("description", ""),
|
|
supportsModes=r.get("supportsModes", ["dark", "light", "auto"]),
|
|
defaultMode=r.get("defaultMode", "dark"),
|
|
author=r.get("author", ""),
|
|
css_url=r.get("css_url", f"/themes/{r.get('id', 'unknown')}/tokens.css"),
|
|
source="plugin",
|
|
))
|
|
|
|
return themes
|
|
|
|
|
|
@router.get("/themes", response_model=ThemeRegistryResponse)
|
|
async def get_available_themes():
|
|
"""Get available themes from disk and plugin registry.
|
|
|
|
Scans the themes/ directory for installed theme packages
|
|
and collects themes registered by plugins via the theme_register hook.
|
|
"""
|
|
builtin = _scan_theme_dirs()
|
|
plugin_themes = _scan_plugin_themes()
|
|
|
|
# Merge, preferring builtin for duplicate IDs
|
|
seen = {t.id for t in builtin}
|
|
all_themes = list(builtin)
|
|
for pt in plugin_themes:
|
|
if pt.id not in seen:
|
|
all_themes.append(pt)
|
|
seen.add(pt.id)
|
|
|
|
return ThemeRegistryResponse(themes=all_themes)
|
|
|
|
return {"success": True, "message": "Dismissed widgets cleared"}
|