mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-05-16 02:13:32 +00:00
Compare commits
18 Commits
update-rgb
...
fix/post-i
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1a3e6f0685 | ||
|
|
5b6137f5f4 | ||
|
|
f97573c368 | ||
|
|
9a74db6de3 | ||
|
|
b7295129b5 | ||
|
|
3e94bb9664 | ||
|
|
44316d3bae | ||
|
|
baebe4f5f7 | ||
|
|
fccd6e70be | ||
|
|
2a74db3a59 | ||
|
|
4b39fbcfd1 | ||
|
|
7ba66e541c | ||
|
|
3f66d15af7 | ||
|
|
9490cf6023 | ||
|
|
15fc9003ac | ||
|
|
55a6a53fca | ||
|
|
c54718af2d | ||
|
|
e8afd23c98 |
@@ -1086,6 +1086,7 @@ SYSTEMCTL_PATH=$(which systemctl)
|
|||||||
REBOOT_PATH=$(which reboot)
|
REBOOT_PATH=$(which reboot)
|
||||||
POWEROFF_PATH=$(which poweroff)
|
POWEROFF_PATH=$(which poweroff)
|
||||||
BASH_PATH=$(which bash)
|
BASH_PATH=$(which bash)
|
||||||
|
JOURNALCTL_PATH=$(which journalctl 2>/dev/null || true)
|
||||||
|
|
||||||
# Create sudoers content
|
# Create sudoers content
|
||||||
cat > /tmp/ledmatrix_web_sudoers << EOF
|
cat > /tmp/ledmatrix_web_sudoers << EOF
|
||||||
@@ -1101,10 +1102,22 @@ $ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix.service
|
|||||||
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH enable ledmatrix.service
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH enable ledmatrix.service
|
||||||
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH disable ledmatrix.service
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH disable ledmatrix.service
|
||||||
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH status ledmatrix.service
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH status ledmatrix.service
|
||||||
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix
|
||||||
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix.service
|
||||||
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start ledmatrix-web.service
|
||||||
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop ledmatrix-web.service
|
||||||
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix-web.service
|
||||||
$ACTUAL_USER ALL=(ALL) NOPASSWD: $PYTHON_PATH $PROJECT_ROOT_DIR/display_controller.py
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $PYTHON_PATH $PROJECT_ROOT_DIR/display_controller.py
|
||||||
$ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/start_display.sh
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/start_display.sh
|
||||||
$ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/stop_display.sh
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/stop_display.sh
|
||||||
EOF
|
EOF
|
||||||
|
if [ -n "$JOURNALCTL_PATH" ]; then
|
||||||
|
cat >> /tmp/ledmatrix_web_sudoers << EOF
|
||||||
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -u ledmatrix.service *
|
||||||
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -u ledmatrix *
|
||||||
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -t ledmatrix *
|
||||||
|
EOF
|
||||||
|
fi
|
||||||
|
|
||||||
if [ -f "$SUDOERS_FILE" ] && cmp -s /tmp/ledmatrix_web_sudoers "$SUDOERS_FILE"; then
|
if [ -f "$SUDOERS_FILE" ] && cmp -s /tmp/ledmatrix_web_sudoers "$SUDOERS_FILE"; then
|
||||||
echo "Sudoers configuration already up to date"
|
echo "Sudoers configuration already up to date"
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
requests>=2.28.0
|
requests>=2.28.0
|
||||||
Pillow>=9.1.0
|
Pillow>=10.3.0
|
||||||
pytz>=2022.1
|
pytz>=2022.1
|
||||||
numpy>=1.24.0
|
numpy>=1.24.0
|
||||||
|
|||||||
@@ -89,9 +89,9 @@ TEMP_SUDOERS="/tmp/ledmatrix_web_sudoers_$$"
|
|||||||
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH status ledmatrix.service"
|
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH status ledmatrix.service"
|
||||||
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix"
|
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix"
|
||||||
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix.service"
|
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix.service"
|
||||||
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start ledmatrix-web"
|
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start ledmatrix-web.service"
|
||||||
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop ledmatrix-web"
|
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop ledmatrix-web.service"
|
||||||
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix-web"
|
echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix-web.service"
|
||||||
|
|
||||||
# Optional: journalctl (non-critical — skip if not found)
|
# Optional: journalctl (non-critical — skip if not found)
|
||||||
if [ -n "$JOURNALCTL_PATH" ]; then
|
if [ -n "$JOURNALCTL_PATH" ]; then
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import sys
|
|||||||
import time
|
import time
|
||||||
import logging
|
import logging
|
||||||
import signal
|
import signal
|
||||||
|
import subprocess
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
# Add project root to path (parent of scripts/utils/)
|
# Add project root to path (parent of scripts/utils/)
|
||||||
@@ -43,6 +44,10 @@ class WiFiMonitorDaemon:
|
|||||||
self.wifi_manager = WiFiManager()
|
self.wifi_manager = WiFiManager()
|
||||||
self.running = True
|
self.running = True
|
||||||
self.last_state = None
|
self.last_state = None
|
||||||
|
# Counts consecutive checks where nmcli says "connected" but internet is unreachable.
|
||||||
|
# After _nm_restart_threshold failures, NetworkManager is restarted as a recovery step.
|
||||||
|
self._consecutive_internet_failures = 0
|
||||||
|
self._nm_restart_threshold = 5 # ~2.5 min at 30s interval
|
||||||
|
|
||||||
# Register signal handlers for graceful shutdown
|
# Register signal handlers for graceful shutdown
|
||||||
signal.signal(signal.SIGINT, self._signal_handler)
|
signal.signal(signal.SIGINT, self._signal_handler)
|
||||||
@@ -122,6 +127,43 @@ class WiFiMonitorDaemon:
|
|||||||
else:
|
else:
|
||||||
logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
|
logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
|
||||||
|
|
||||||
|
# Escalating recovery: if nmcli reports connected but actual internet
|
||||||
|
# is unreachable for several consecutive checks, restart NetworkManager.
|
||||||
|
# This is done HERE (not inside check_and_manage_ap_mode) to keep the
|
||||||
|
# AP-enable trigger clean and avoid false-positive AP enables from
|
||||||
|
# transient packet loss on otherwise working WiFi.
|
||||||
|
if updated_status.connected and not updated_status.ap_mode_active:
|
||||||
|
if not self.wifi_manager.check_internet_connectivity():
|
||||||
|
self._consecutive_internet_failures += 1
|
||||||
|
logger.warning(
|
||||||
|
f"Internet unreachable despite nmcli connection "
|
||||||
|
f"({self._consecutive_internet_failures}/{self._nm_restart_threshold})"
|
||||||
|
)
|
||||||
|
if self._consecutive_internet_failures >= self._nm_restart_threshold:
|
||||||
|
logger.warning("Restarting NetworkManager to recover internet connectivity")
|
||||||
|
try:
|
||||||
|
subprocess.run(
|
||||||
|
["/usr/bin/systemctl", "restart", "NetworkManager"],
|
||||||
|
capture_output=True, timeout=20, check=True
|
||||||
|
)
|
||||||
|
self._consecutive_internet_failures = 0
|
||||||
|
# NM restart causes a brief WiFi drop; reset the AP-mode grace
|
||||||
|
# counter so that transient disconnect doesn't count toward
|
||||||
|
# triggering AP mode.
|
||||||
|
self.wifi_manager._disconnected_checks = 0
|
||||||
|
except subprocess.CalledProcessError as e:
|
||||||
|
logger.error(f"NetworkManager restart failed (rc={e.returncode}); "
|
||||||
|
"resetting failure counter to avoid tight retry loop")
|
||||||
|
self._consecutive_internet_failures = 0
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"NetworkManager restart error: {e}; "
|
||||||
|
"resetting failure counter to avoid tight retry loop")
|
||||||
|
self._consecutive_internet_failures = 0
|
||||||
|
else:
|
||||||
|
self._consecutive_internet_failures = 0
|
||||||
|
else:
|
||||||
|
self._consecutive_internet_failures = 0
|
||||||
|
|
||||||
# Sleep until next check
|
# Sleep until next check
|
||||||
time.sleep(self.check_interval)
|
time.sleep(self.check_interval)
|
||||||
|
|
||||||
|
|||||||
@@ -60,12 +60,16 @@ def get_wifi_config_path():
|
|||||||
|
|
||||||
HOSTAPD_CONFIG_PATH = Path("/etc/hostapd/hostapd.conf")
|
HOSTAPD_CONFIG_PATH = Path("/etc/hostapd/hostapd.conf")
|
||||||
DNSMASQ_CONFIG_PATH = Path("/etc/dnsmasq.d/ledmatrix-captive.conf")
|
DNSMASQ_CONFIG_PATH = Path("/etc/dnsmasq.d/ledmatrix-captive.conf")
|
||||||
|
# Drop-in config for NetworkManager's built-in dnsmasq (ipv4.method=shared).
|
||||||
|
# Writing address=/#/<ap_ip> here causes NM to resolve every hostname to the AP,
|
||||||
|
# triggering the OS captive-portal popup automatically on iOS/Android/Windows/macOS.
|
||||||
|
NM_DNSMASQ_SHARED_DIR = Path("/etc/NetworkManager/dnsmasq-shared.d")
|
||||||
|
NM_DNSMASQ_SHARED_CONF = NM_DNSMASQ_SHARED_DIR / "ledmatrix-captive.conf"
|
||||||
HOSTAPD_SERVICE = "hostapd"
|
HOSTAPD_SERVICE = "hostapd"
|
||||||
DNSMASQ_SERVICE = "dnsmasq"
|
DNSMASQ_SERVICE = "dnsmasq"
|
||||||
|
|
||||||
# Default AP settings
|
# Default AP settings
|
||||||
DEFAULT_AP_SSID = "LEDMatrix-Setup"
|
DEFAULT_AP_SSID = "LEDMatrix-Setup"
|
||||||
DEFAULT_AP_PASSWORD = "ledmatrix123"
|
|
||||||
DEFAULT_AP_CHANNEL = 7
|
DEFAULT_AP_CHANNEL = 7
|
||||||
|
|
||||||
# LED status message file (for display_controller integration)
|
# LED status message file (for display_controller integration)
|
||||||
@@ -138,6 +142,11 @@ class WiFiManager:
|
|||||||
self._disconnected_checks = 0
|
self._disconnected_checks = 0
|
||||||
self._disconnected_checks_required = 3 # Require 3 consecutive disconnected checks (90 seconds at 30s interval)
|
self._disconnected_checks_required = 3 # Require 3 consecutive disconnected checks (90 seconds at 30s interval)
|
||||||
|
|
||||||
|
# Timestamp set when AP mode is enabled; used for the idle-timeout check
|
||||||
|
self._ap_enabled_at: Optional[float] = None
|
||||||
|
# Which redirect backend was used (iptables/nftables/None); set per-instance
|
||||||
|
self._redirect_backend: Optional[str] = None
|
||||||
|
|
||||||
logger.info(f"WiFi Manager initialized - nmcli: {self.has_nmcli}, iwlist: {self.has_iwlist}, "
|
logger.info(f"WiFi Manager initialized - nmcli: {self.has_nmcli}, iwlist: {self.has_iwlist}, "
|
||||||
f"hostapd: {self.has_hostapd}, dnsmasq: {self.has_dnsmasq}, "
|
f"hostapd: {self.has_hostapd}, dnsmasq: {self.has_dnsmasq}, "
|
||||||
f"interface: {self._wifi_interface}, trixie: {self._is_trixie}")
|
f"interface: {self._wifi_interface}, trixie: {self._is_trixie}")
|
||||||
@@ -201,6 +210,24 @@ class WiFiManager:
|
|||||||
except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError):
|
except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError):
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
def _find_command_path(self, command: str) -> Optional[str]:
|
||||||
|
"""
|
||||||
|
Return the absolute path of a command, checking sbin locations that may not
|
||||||
|
be on PATH in restricted service environments. Returns None if not found.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
result = subprocess.run(["which", command], capture_output=True,
|
||||||
|
text=True, timeout=2)
|
||||||
|
if result.returncode == 0 and result.stdout.strip():
|
||||||
|
return result.stdout.strip()
|
||||||
|
except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError):
|
||||||
|
pass
|
||||||
|
for path in [f"/usr/sbin/{command}", f"/sbin/{command}",
|
||||||
|
f"/usr/local/sbin/{command}"]:
|
||||||
|
if os.path.isfile(path) and os.access(path, os.X_OK):
|
||||||
|
return path
|
||||||
|
return None
|
||||||
|
|
||||||
def _discover_wifi_interface(self) -> str:
|
def _discover_wifi_interface(self) -> str:
|
||||||
"""
|
"""
|
||||||
Discover the primary WiFi interface name dynamically.
|
Discover the primary WiFi interface name dynamically.
|
||||||
@@ -303,7 +330,6 @@ class WiFiManager:
|
|||||||
else:
|
else:
|
||||||
self.config = {
|
self.config = {
|
||||||
"ap_ssid": DEFAULT_AP_SSID,
|
"ap_ssid": DEFAULT_AP_SSID,
|
||||||
"ap_password": DEFAULT_AP_PASSWORD,
|
|
||||||
"ap_channel": DEFAULT_AP_CHANNEL,
|
"ap_channel": DEFAULT_AP_CHANNEL,
|
||||||
"auto_enable_ap_mode": True, # Default: auto-enable when no network (safe due to grace period)
|
"auto_enable_ap_mode": True, # Default: auto-enable when no network (safe due to grace period)
|
||||||
"saved_networks": []
|
"saved_networks": []
|
||||||
@@ -659,6 +685,285 @@ class WiFiManager:
|
|||||||
except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError):
|
except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError):
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Helpers
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
_IP_FORWARD_SAVE_PATH = Path("/tmp/ledmatrix_ip_forward_saved")
|
||||||
|
|
||||||
|
def _validate_ap_config(self) -> Tuple[str, int]:
|
||||||
|
"""Return a sanitized (ssid, channel) pair from config, falling back to defaults."""
|
||||||
|
ssid = str(self.config.get("ap_ssid", DEFAULT_AP_SSID))
|
||||||
|
if not ssid or len(ssid) > 32 or not re.match(r'^[\x20-\x7E]+$', ssid):
|
||||||
|
logger.warning(f"AP SSID '{ssid}' is invalid, falling back to default")
|
||||||
|
ssid = DEFAULT_AP_SSID
|
||||||
|
try:
|
||||||
|
channel = int(self.config.get("ap_channel", DEFAULT_AP_CHANNEL))
|
||||||
|
if channel < 1 or channel > 14:
|
||||||
|
raise ValueError
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
logger.warning("AP channel out of range, falling back to default")
|
||||||
|
channel = DEFAULT_AP_CHANNEL
|
||||||
|
return ssid, channel
|
||||||
|
|
||||||
|
def _setup_iptables_redirect(self) -> bool:
|
||||||
|
"""
|
||||||
|
Add port 80 → 5000 redirect rules for the captive portal.
|
||||||
|
|
||||||
|
Tries iptables first, falls back to nftables (used by Debian Trixie).
|
||||||
|
When neither tool is available, logs a warning and returns True — the AP
|
||||||
|
still works and DNS spoofing still triggers the OS popup; users just land
|
||||||
|
on port 5000 directly rather than being redirected from port 80.
|
||||||
|
|
||||||
|
Only returns False when a tool was found but the rule addition itself failed.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
iptables = self._find_command_path("iptables")
|
||||||
|
nft = self._find_command_path("nft")
|
||||||
|
|
||||||
|
if not iptables and not nft:
|
||||||
|
logger.warning(
|
||||||
|
"Neither iptables nor nft found; captive portal port-80 redirect unavailable. "
|
||||||
|
"DNS spoofing will still trigger the OS popup but HTTP on port 80 won't reach Flask."
|
||||||
|
)
|
||||||
|
self._redirect_backend = None
|
||||||
|
return True # AP works; redirect is best-effort
|
||||||
|
|
||||||
|
if iptables:
|
||||||
|
return self._setup_iptables_redirect_iptables(iptables)
|
||||||
|
else:
|
||||||
|
return self._setup_iptables_redirect_nftables(nft)
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"Could not set up port redirect: {e}")
|
||||||
|
try:
|
||||||
|
self._teardown_iptables_redirect()
|
||||||
|
except Exception as cleanup_e:
|
||||||
|
logger.warning(f"Cleanup after redirect exception also failed: {cleanup_e}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
def _setup_iptables_redirect_iptables(self, iptables: str) -> bool:
|
||||||
|
"""Set up port 80→5000 redirect using iptables."""
|
||||||
|
# Save ip_forward state before enabling
|
||||||
|
try:
|
||||||
|
current_fwd = Path("/proc/sys/net/ipv4/ip_forward").read_text().strip()
|
||||||
|
except OSError:
|
||||||
|
current_fwd = None
|
||||||
|
if current_fwd is not None:
|
||||||
|
try:
|
||||||
|
self._IP_FORWARD_SAVE_PATH.write_text(current_fwd)
|
||||||
|
except OSError:
|
||||||
|
current_fwd = None
|
||||||
|
logger.warning("Could not write ip_forward save file; state will not be restored")
|
||||||
|
|
||||||
|
if current_fwd != "1":
|
||||||
|
sysctl = self._find_command_path("sysctl")
|
||||||
|
sysctl_bin = sysctl if sysctl else "sysctl"
|
||||||
|
r = subprocess.run(["sudo", sysctl_bin, "-w", "net.ipv4.ip_forward=1"],
|
||||||
|
capture_output=True, text=True, timeout=5)
|
||||||
|
if r.returncode != 0:
|
||||||
|
logger.error(f"Failed to enable ip_forward: {r.stderr.strip()}")
|
||||||
|
self._teardown_iptables_redirect()
|
||||||
|
return False
|
||||||
|
|
||||||
|
if subprocess.run(
|
||||||
|
["sudo", iptables, "-t", "nat", "-C", "PREROUTING",
|
||||||
|
"-i", self._wifi_interface, "-p", "tcp", "--dport", "80",
|
||||||
|
"-j", "REDIRECT", "--to-port", "5000"],
|
||||||
|
capture_output=True, timeout=5
|
||||||
|
).returncode != 0:
|
||||||
|
r = subprocess.run(
|
||||||
|
["sudo", iptables, "-t", "nat", "-A", "PREROUTING",
|
||||||
|
"-i", self._wifi_interface, "-p", "tcp", "--dport", "80",
|
||||||
|
"-j", "REDIRECT", "--to-port", "5000"],
|
||||||
|
capture_output=True, text=True, timeout=5
|
||||||
|
)
|
||||||
|
if r.returncode != 0:
|
||||||
|
logger.error(f"Failed to add PREROUTING rule: {r.stderr.strip()}")
|
||||||
|
self._teardown_iptables_redirect()
|
||||||
|
return False
|
||||||
|
|
||||||
|
if subprocess.run(
|
||||||
|
["sudo", iptables, "-C", "INPUT",
|
||||||
|
"-i", self._wifi_interface, "-p", "tcp", "--dport", "5000", "-j", "ACCEPT"],
|
||||||
|
capture_output=True, timeout=5
|
||||||
|
).returncode != 0:
|
||||||
|
r = subprocess.run(
|
||||||
|
["sudo", iptables, "-A", "INPUT",
|
||||||
|
"-i", self._wifi_interface, "-p", "tcp", "--dport", "5000", "-j", "ACCEPT"],
|
||||||
|
capture_output=True, text=True, timeout=5
|
||||||
|
)
|
||||||
|
if r.returncode != 0:
|
||||||
|
logger.error(f"Failed to add INPUT rule: {r.stderr.strip()}")
|
||||||
|
self._teardown_iptables_redirect()
|
||||||
|
return False
|
||||||
|
|
||||||
|
self._redirect_backend = "iptables"
|
||||||
|
logger.info("iptables: port 80→5000 redirect rules added")
|
||||||
|
return True
|
||||||
|
|
||||||
|
def _setup_iptables_redirect_nftables(self, nft: str) -> bool:
|
||||||
|
"""Set up port 80→5000 redirect using nftables (Debian Trixie / modern systems)."""
|
||||||
|
# NM's ipv4.method=shared already enables ip_forward; no sysctl needed.
|
||||||
|
cmds = [
|
||||||
|
["sudo", nft, "add", "table", "ip", "ledmatrix"],
|
||||||
|
["sudo", nft, "add", "chain", "ip", "ledmatrix", "prerouting",
|
||||||
|
"{", "type", "nat", "hook", "prerouting", "priority", "-100", ";", "}"],
|
||||||
|
["sudo", nft, "add", "rule", "ip", "ledmatrix", "prerouting",
|
||||||
|
"iif", self._wifi_interface, "tcp", "dport", "80", "redirect", "to", ":5000"],
|
||||||
|
]
|
||||||
|
for cmd in cmds:
|
||||||
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
|
||||||
|
if r.returncode != 0:
|
||||||
|
# Table/chain may already exist — only fail on rule add
|
||||||
|
if "add rule" in " ".join(cmd):
|
||||||
|
logger.error(f"Failed to add nftables redirect rule: {r.stderr.strip()}")
|
||||||
|
self._teardown_iptables_redirect()
|
||||||
|
return False
|
||||||
|
logger.debug(f"nft cmd non-zero (may already exist): {r.stderr.strip()}")
|
||||||
|
|
||||||
|
self._redirect_backend = "nftables"
|
||||||
|
logger.info("nftables: port 80→5000 redirect rule added")
|
||||||
|
return True
|
||||||
|
|
||||||
|
def _teardown_iptables_redirect(self) -> None:
|
||||||
|
"""Remove the port 80→5000 redirect rules and restore ip_forward if saved."""
|
||||||
|
try:
|
||||||
|
backend = self._redirect_backend
|
||||||
|
self._redirect_backend = None
|
||||||
|
|
||||||
|
if backend == "iptables":
|
||||||
|
iptables = self._find_command_path("iptables")
|
||||||
|
if iptables:
|
||||||
|
subprocess.run(
|
||||||
|
["sudo", iptables, "-t", "nat", "-D", "PREROUTING",
|
||||||
|
"-i", self._wifi_interface, "-p", "tcp", "--dport", "80",
|
||||||
|
"-j", "REDIRECT", "--to-port", "5000"],
|
||||||
|
capture_output=True, timeout=5
|
||||||
|
)
|
||||||
|
subprocess.run(
|
||||||
|
["sudo", iptables, "-D", "INPUT",
|
||||||
|
"-i", self._wifi_interface, "-p", "tcp", "--dport", "5000",
|
||||||
|
"-j", "ACCEPT"],
|
||||||
|
capture_output=True, timeout=5
|
||||||
|
)
|
||||||
|
# Restore ip_forward only when we saved it
|
||||||
|
if self._IP_FORWARD_SAVE_PATH.exists():
|
||||||
|
try:
|
||||||
|
saved = self._IP_FORWARD_SAVE_PATH.read_text().strip()
|
||||||
|
self._IP_FORWARD_SAVE_PATH.unlink(missing_ok=True)
|
||||||
|
sysctl = self._find_command_path("sysctl")
|
||||||
|
sysctl_bin = sysctl if sysctl else "sysctl"
|
||||||
|
subprocess.run(["sudo", sysctl_bin, "-w", f"net.ipv4.ip_forward={saved}"],
|
||||||
|
capture_output=True, timeout=5)
|
||||||
|
logger.info(f"ip_forward restored to {saved}")
|
||||||
|
except OSError as e:
|
||||||
|
logger.warning(f"Could not restore ip_forward: {e}")
|
||||||
|
else:
|
||||||
|
logger.debug("ip_forward not modified by setup; leaving unchanged")
|
||||||
|
|
||||||
|
elif backend == "nftables":
|
||||||
|
nft = self._find_command_path("nft")
|
||||||
|
if nft:
|
||||||
|
subprocess.run(
|
||||||
|
["sudo", nft, "delete", "table", "ip", "ledmatrix"],
|
||||||
|
capture_output=True, timeout=5
|
||||||
|
)
|
||||||
|
logger.info("nftables ledmatrix table removed")
|
||||||
|
|
||||||
|
else:
|
||||||
|
# No redirect was set up (neither tool available); nothing to tear down
|
||||||
|
self._IP_FORWARD_SAVE_PATH.unlink(missing_ok=True)
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"Could not tear down port redirect: {e}")
|
||||||
|
|
||||||
|
def _write_nm_dnsmasq_captive_conf(self, ap_ip: str = "192.168.4.1") -> None:
|
||||||
|
"""
|
||||||
|
Write the NM dnsmasq-shared.d drop-in that makes NM's built-in dnsmasq
|
||||||
|
resolve every hostname to the AP IP. This triggers the OS captive-portal
|
||||||
|
popup automatically on iOS / Android / Windows / macOS as soon as the
|
||||||
|
device connects — no manual navigation required.
|
||||||
|
|
||||||
|
NetworkManager reads /etc/NetworkManager/dnsmasq-shared.d/*.conf when it
|
||||||
|
starts the dnsmasq instance for ipv4.method=shared connections.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
content = f"# LEDMatrix captive portal: resolve all hostnames to AP\naddress=/#/{ap_ip}\n"
|
||||||
|
with open("/tmp/ledmatrix-nm-dnsmasq.conf", "w") as f:
|
||||||
|
f.write(content)
|
||||||
|
subprocess.run(
|
||||||
|
["sudo", "mkdir", "-p", str(NM_DNSMASQ_SHARED_DIR)],
|
||||||
|
capture_output=True, timeout=5
|
||||||
|
)
|
||||||
|
subprocess.run(
|
||||||
|
["sudo", "cp", "/tmp/ledmatrix-nm-dnsmasq.conf", str(NM_DNSMASQ_SHARED_CONF)],
|
||||||
|
capture_output=True, timeout=5
|
||||||
|
)
|
||||||
|
logger.info(f"Wrote NM dnsmasq captive-portal config: {NM_DNSMASQ_SHARED_CONF}")
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"Could not write NM dnsmasq captive config: {e}")
|
||||||
|
|
||||||
|
def _remove_nm_dnsmasq_captive_conf(self) -> None:
|
||||||
|
"""Remove the NM dnsmasq-shared.d drop-in written by _write_nm_dnsmasq_captive_conf."""
|
||||||
|
try:
|
||||||
|
subprocess.run(
|
||||||
|
["sudo", "rm", "-f", str(NM_DNSMASQ_SHARED_CONF)],
|
||||||
|
capture_output=True, timeout=5
|
||||||
|
)
|
||||||
|
logger.info("Removed NM dnsmasq captive-portal config")
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"Could not remove NM dnsmasq captive config: {e}")
|
||||||
|
|
||||||
|
def _check_internet_connectivity(self, timeout: int = 5) -> bool:
|
||||||
|
"""
|
||||||
|
Test actual internet reachability — not just nmcli association state.
|
||||||
|
|
||||||
|
A device can be 'connected' in nmcli (associated with an AP) while the
|
||||||
|
router has no WAN link. This check catches that case so the daemon can
|
||||||
|
auto-enable AP mode even when nmcli reports a connection.
|
||||||
|
|
||||||
|
Returns True if at least one reachability method succeeds.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
r = subprocess.run(
|
||||||
|
["ping", "-c", "1", "-W", str(timeout), "8.8.8.8"],
|
||||||
|
capture_output=True, timeout=timeout + 1
|
||||||
|
)
|
||||||
|
if r.returncode == 0:
|
||||||
|
logger.debug("Internet connectivity confirmed via ping 8.8.8.8")
|
||||||
|
return True
|
||||||
|
except (subprocess.SubprocessError, OSError):
|
||||||
|
pass
|
||||||
|
try:
|
||||||
|
import urllib.request as _ureq
|
||||||
|
_ureq.urlopen("http://connectivity-check.ubuntu.com/", timeout=timeout)
|
||||||
|
logger.debug("Internet connectivity confirmed via HTTP check")
|
||||||
|
return True
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
logger.debug("Internet connectivity check failed (both ping and HTTP)")
|
||||||
|
return False
|
||||||
|
|
||||||
|
def check_internet_connectivity(self, timeout: int = 5) -> bool:
|
||||||
|
"""Public wrapper around _check_internet_connectivity for use by the daemon."""
|
||||||
|
return self._check_internet_connectivity(timeout=timeout)
|
||||||
|
|
||||||
|
def _has_ap_clients(self) -> bool:
|
||||||
|
"""
|
||||||
|
Return True if at least one client is associated with the AP.
|
||||||
|
Uses 'iw dev <iface> station dump' which works for both hostapd and
|
||||||
|
nmcli AP modes.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
["iw", "dev", self._wifi_interface, "station", "dump"],
|
||||||
|
capture_output=True, text=True, timeout=5
|
||||||
|
)
|
||||||
|
return bool(result.stdout.strip())
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
|
||||||
def scan_networks(self, allow_cached: bool = True) -> Tuple[List[WiFiNetwork], bool]:
|
def scan_networks(self, allow_cached: bool = True) -> Tuple[List[WiFiNetwork], bool]:
|
||||||
"""
|
"""
|
||||||
Scan for available WiFi networks.
|
Scan for available WiFi networks.
|
||||||
@@ -1293,12 +1598,27 @@ class WiFiManager:
|
|||||||
error_msg = result.stderr.strip() or result.stdout.strip()
|
error_msg = result.stderr.strip() or result.stdout.strip()
|
||||||
logger.error(f"Failed to connect to {ssid}: {error_msg}")
|
logger.error(f"Failed to connect to {ssid}: {error_msg}")
|
||||||
self._show_led_message("Connection failed", duration=5)
|
self._show_led_message("Connection failed", duration=5)
|
||||||
|
if self._is_wrong_password_error(error_msg):
|
||||||
|
return False, f"wrong_password: {error_msg}"
|
||||||
return False, error_msg
|
return False, error_msg
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error connecting with nmcli: {e}")
|
logger.error(f"Error connecting with nmcli: {e}")
|
||||||
self._show_led_message("Connection error", duration=5)
|
self._show_led_message("Connection error", duration=5)
|
||||||
return False, str(e)
|
return False, str(e)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _is_wrong_password_error(error_msg: str) -> bool:
|
||||||
|
"""Return True when nmcli's error output indicates an authentication failure."""
|
||||||
|
indicators = [
|
||||||
|
"secrets were required",
|
||||||
|
"no secret agent",
|
||||||
|
"802-11-wireless-security.psk",
|
||||||
|
"authentication rejected",
|
||||||
|
"association rejected",
|
||||||
|
]
|
||||||
|
lower = error_msg.lower()
|
||||||
|
return any(ind in lower for ind in indicators)
|
||||||
|
|
||||||
def _connect_wpa_supplicant(self, ssid: str, password: str) -> Tuple[bool, str]:
|
def _connect_wpa_supplicant(self, ssid: str, password: str) -> Tuple[bool, str]:
|
||||||
"""Connect using wpa_supplicant (fallback)"""
|
"""Connect using wpa_supplicant (fallback)"""
|
||||||
try:
|
try:
|
||||||
@@ -1570,13 +1890,17 @@ class WiFiManager:
|
|||||||
if self.has_hostapd and self.has_dnsmasq:
|
if self.has_hostapd and self.has_dnsmasq:
|
||||||
result = self._enable_ap_mode_hostapd()
|
result = self._enable_ap_mode_hostapd()
|
||||||
if result[0]:
|
if result[0]:
|
||||||
|
self._ap_enabled_at = time.time()
|
||||||
return result
|
return result
|
||||||
|
|
||||||
# Fallback to nmcli hotspot (simpler, no captive portal)
|
# Fallback to nmcli hotspot (simpler, no captive portal)
|
||||||
if self.has_nmcli:
|
if self.has_nmcli:
|
||||||
logger.info("hostapd/dnsmasq failed or unavailable, trying nmcli hotspot fallback...")
|
logger.info("hostapd/dnsmasq failed or unavailable, trying nmcli hotspot fallback...")
|
||||||
self._show_led_message("Setup Mode", duration=5)
|
self._show_led_message("Setup Mode", duration=5)
|
||||||
return self._enable_ap_mode_nmcli_hotspot()
|
result = self._enable_ap_mode_nmcli_hotspot()
|
||||||
|
if result[0]:
|
||||||
|
self._ap_enabled_at = time.time()
|
||||||
|
return result
|
||||||
|
|
||||||
return False, "No WiFi tools available (nmcli, hostapd, or dnsmasq required)"
|
return False, "No WiFi tools available (nmcli, hostapd, or dnsmasq required)"
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@@ -1649,63 +1973,21 @@ class WiFiManager:
|
|||||||
subprocess.run(["sudo", "systemctl", "stop", HOSTAPD_SERVICE], timeout=5)
|
subprocess.run(["sudo", "systemctl", "stop", HOSTAPD_SERVICE], timeout=5)
|
||||||
return False, f"Failed to start dnsmasq: {result.stderr}"
|
return False, f"Failed to start dnsmasq: {result.stderr}"
|
||||||
|
|
||||||
# Set up iptables port forwarding: redirect port 80 to 5000
|
# Set up iptables port forwarding (port 80 → 5000) and save ip_forward state
|
||||||
# This makes the captive portal work on standard HTTP port
|
if not self._setup_iptables_redirect():
|
||||||
try:
|
logger.error("Captive-portal redirect setup failed; stopping AP services")
|
||||||
# Check if iptables is available
|
subprocess.run(["sudo", "systemctl", "stop", HOSTAPD_SERVICE],
|
||||||
iptables_check = subprocess.run(
|
capture_output=True, timeout=10)
|
||||||
["which", "iptables"],
|
subprocess.run(["sudo", "systemctl", "stop", DNSMASQ_SERVICE],
|
||||||
capture_output=True,
|
capture_output=True, timeout=10)
|
||||||
timeout=2
|
return False, "AP started but captive-portal redirect setup failed"
|
||||||
)
|
|
||||||
|
|
||||||
if iptables_check.returncode == 0:
|
|
||||||
# Enable IP forwarding (needed for NAT)
|
|
||||||
subprocess.run(
|
|
||||||
["sudo", "sysctl", "-w", "net.ipv4.ip_forward=1"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
|
|
||||||
# Add NAT rule to redirect port 80 to 5000 on WiFi interface
|
|
||||||
# First check if rule already exists
|
|
||||||
check_result = subprocess.run(
|
|
||||||
["sudo", "iptables", "-t", "nat", "-C", "PREROUTING", "-i", self._wifi_interface, "-p", "tcp", "--dport", "80", "-j", "REDIRECT", "--to-port", "5000"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
|
|
||||||
if check_result.returncode != 0:
|
|
||||||
# Rule doesn't exist, add it
|
|
||||||
subprocess.run(
|
|
||||||
["sudo", "iptables", "-t", "nat", "-A", "PREROUTING", "-i", self._wifi_interface, "-p", "tcp", "--dport", "80", "-j", "REDIRECT", "--to-port", "5000"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
logger.info("Added iptables rule to redirect port 80 to 5000")
|
|
||||||
|
|
||||||
# Also allow incoming connections on port 80
|
|
||||||
check_input = subprocess.run(
|
|
||||||
["sudo", "iptables", "-C", "INPUT", "-i", self._wifi_interface, "-p", "tcp", "--dport", "80", "-j", "ACCEPT"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
|
|
||||||
if check_input.returncode != 0:
|
|
||||||
subprocess.run(
|
|
||||||
["sudo", "iptables", "-A", "INPUT", "-i", self._wifi_interface, "-p", "tcp", "--dport", "80", "-j", "ACCEPT"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
logger.debug("iptables not available, port forwarding not set up")
|
|
||||||
logger.info("Note: Port 80 forwarding requires iptables. Users will need to access port 5000 directly.")
|
|
||||||
except Exception as e:
|
|
||||||
logger.warning(f"Could not set up iptables port forwarding: {e}")
|
|
||||||
# Continue anyway - port 5000 will still work
|
|
||||||
|
|
||||||
logger.info("AP mode enabled successfully")
|
logger.info("AP mode enabled successfully")
|
||||||
self._show_led_message("Setup Mode Active", duration=5)
|
# Use the validated SSID so the displayed name matches what hostapd broadcast
|
||||||
|
ap_ssid, _ = self._validate_ap_config()
|
||||||
|
self._show_led_message(
|
||||||
|
f"WiFi Setup\n{ap_ssid}\nNo password\n192.168.4.1:5000", duration=10
|
||||||
|
)
|
||||||
return True, "AP mode enabled"
|
return True, "AP mode enabled"
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error starting AP services: {e}")
|
logger.error(f"Error starting AP services: {e}")
|
||||||
@@ -1716,245 +1998,120 @@ class WiFiManager:
|
|||||||
|
|
||||||
def _enable_ap_mode_nmcli_hotspot(self) -> Tuple[bool, str]:
|
def _enable_ap_mode_nmcli_hotspot(self) -> Tuple[bool, str]:
|
||||||
"""
|
"""
|
||||||
Enable AP mode using nmcli hotspot.
|
Enable AP mode using nmcli as an open (passwordless) access point.
|
||||||
|
|
||||||
This method is optimized for both Bookworm and Trixie:
|
Uses 'nmcli connection add type wifi 802-11-wireless.mode ap' instead of
|
||||||
- Trixie: Uses Netplan, connections stored in /run/NetworkManager/system-connections
|
'nmcli device wifi hotspot' because the hotspot subcommand always creates a
|
||||||
- Bookworm: Traditional NetworkManager, connections in /etc/NetworkManager/system-connections
|
WPA2-protected network on Bookworm/Trixie and silently ignores attempts to
|
||||||
|
strip security after creation.
|
||||||
|
|
||||||
On Trixie, we also disable PMF (Protected Management Frames) which can cause
|
Tested for both Bookworm and Trixie (Netplan-based NetworkManager).
|
||||||
connection issues with certain WiFi adapters and clients.
|
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
# Stop any existing connection
|
# Stop any existing connection
|
||||||
self.disconnect_from_network()
|
self.disconnect_from_network()
|
||||||
time.sleep(1)
|
time.sleep(1)
|
||||||
|
|
||||||
# Delete any existing hotspot connections (more thorough cleanup)
|
ap_ssid, ap_channel = self._validate_ap_config()
|
||||||
# First, list all connections to find any with the same SSID or hotspot-related ones
|
|
||||||
ap_ssid = self.config.get("ap_ssid", DEFAULT_AP_SSID)
|
|
||||||
result = subprocess.run(
|
|
||||||
["nmcli", "-t", "-f", "NAME,TYPE,802-11-wireless.ssid", "connection", "show"],
|
|
||||||
capture_output=True,
|
|
||||||
text=True,
|
|
||||||
timeout=10
|
|
||||||
)
|
|
||||||
if result.returncode == 0:
|
|
||||||
for line in result.stdout.strip().split('\n'):
|
|
||||||
if ':' in line:
|
|
||||||
parts = line.split(':')
|
|
||||||
if len(parts) >= 2:
|
|
||||||
conn_name = parts[0].strip()
|
|
||||||
conn_type = parts[1].strip().lower() if len(parts) > 1 else ""
|
|
||||||
conn_ssid = parts[2].strip() if len(parts) > 2 else ""
|
|
||||||
|
|
||||||
# Delete if:
|
# Delete only the specific application-managed AP profiles by name.
|
||||||
# 1. It's a hotspot type
|
# Never delete by SSID — that would destroy a user's saved home network.
|
||||||
# 2. It has the same SSID as our AP
|
|
||||||
# 3. It matches our known connection names
|
|
||||||
should_delete = (
|
|
||||||
'hotspot' in conn_type or
|
|
||||||
conn_ssid == ap_ssid or
|
|
||||||
'hotspot' in conn_name.lower() or
|
|
||||||
conn_name in ["Hotspot", "LEDMatrix-Setup-AP", "TickerSetup-AP"]
|
|
||||||
)
|
|
||||||
|
|
||||||
if should_delete:
|
|
||||||
logger.info(f"Deleting existing connection: {conn_name} (type: {conn_type}, SSID: {conn_ssid})")
|
|
||||||
# First disconnect it if active
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "down", conn_name],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
# Then delete it
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "delete", conn_name],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=10
|
|
||||||
)
|
|
||||||
|
|
||||||
# Also explicitly delete known connection names (in case they weren't caught above)
|
|
||||||
for conn_name in ["Hotspot", "LEDMatrix-Setup-AP", "TickerSetup-AP"]:
|
for conn_name in ["Hotspot", "LEDMatrix-Setup-AP", "TickerSetup-AP"]:
|
||||||
subprocess.run(
|
subprocess.run(["nmcli", "connection", "down", conn_name],
|
||||||
["nmcli", "connection", "down", conn_name],
|
capture_output=True, timeout=5)
|
||||||
capture_output=True,
|
subprocess.run(["nmcli", "connection", "delete", conn_name],
|
||||||
timeout=5
|
capture_output=True, timeout=10)
|
||||||
)
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "delete", conn_name],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=10
|
|
||||||
)
|
|
||||||
|
|
||||||
# Wait a moment for deletions to complete
|
|
||||||
time.sleep(1)
|
time.sleep(1)
|
||||||
|
|
||||||
# Get AP settings from config
|
# Create an open AP connection profile from scratch.
|
||||||
ap_ssid = self.config.get("ap_ssid", DEFAULT_AP_SSID)
|
# Using 'connection add' instead of 'device wifi hotspot' because the
|
||||||
ap_channel = self.config.get("ap_channel", DEFAULT_AP_CHANNEL)
|
# hotspot subcommand always attaches a WPA2 PSK on Bookworm/Trixie and
|
||||||
|
# ignores post-creation security modifications.
|
||||||
# Use nmcli hotspot command (simpler, works with Broadcom chips)
|
logger.info(f"Creating open AP with nmcli connection add: {ap_ssid} on "
|
||||||
# Open network (no password) for easy setup access
|
f"{self._wifi_interface} (no password)")
|
||||||
logger.info(f"Creating open hotspot with nmcli: {ap_ssid} on {self._wifi_interface} (no password)")
|
|
||||||
|
|
||||||
# Note: Some NetworkManager versions add a default password to hotspots
|
|
||||||
# We'll create it and then immediately remove all security settings
|
|
||||||
cmd = [
|
cmd = [
|
||||||
"nmcli", "device", "wifi", "hotspot",
|
"nmcli", "connection", "add",
|
||||||
"ifname", self._wifi_interface,
|
"type", "wifi",
|
||||||
"con-name", "LEDMatrix-Setup-AP",
|
"con-name", "LEDMatrix-Setup-AP",
|
||||||
|
"ifname", self._wifi_interface,
|
||||||
"ssid", ap_ssid,
|
"ssid", ap_ssid,
|
||||||
"band", "bg", # 2.4GHz for maximum compatibility
|
"802-11-wireless.mode", "ap",
|
||||||
"channel", str(ap_channel),
|
"802-11-wireless.band", "bg", # 2.4 GHz for maximum compatibility
|
||||||
# Don't pass password parameter - we'll remove security after creation
|
"802-11-wireless.channel", str(ap_channel),
|
||||||
|
"ipv4.method", "shared",
|
||||||
|
"ipv4.addresses", "192.168.4.1/24",
|
||||||
|
# No 802-11-wireless-security section → open network
|
||||||
]
|
]
|
||||||
|
|
||||||
result = subprocess.run(
|
# PMF (Protected Management Frames) is only meaningful for WPA2/WPA3.
|
||||||
cmd,
|
# An open AP has no security section, so adding 802-11-wireless-security.pmf
|
||||||
capture_output=True,
|
# would cause NM to require key-mgmt too, breaking the connection add on
|
||||||
text=True,
|
# Trixie NM 1.52+. Leave PMF untouched — open APs have no frame protection.
|
||||||
timeout=30
|
|
||||||
)
|
|
||||||
|
|
||||||
if result.returncode == 0:
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
||||||
# Always explicitly remove all security settings to ensure open network
|
|
||||||
# NetworkManager sometimes adds default security even when not specified
|
|
||||||
logger.info("Ensuring hotspot is open (no password)...")
|
|
||||||
time.sleep(2) # Give it a moment to create
|
|
||||||
|
|
||||||
# Remove all possible security settings
|
if result.returncode != 0:
|
||||||
security_settings = [
|
|
||||||
("802-11-wireless-security.key-mgmt", "none"),
|
|
||||||
("802-11-wireless-security.psk", ""),
|
|
||||||
("802-11-wireless-security.wep-key", ""),
|
|
||||||
("802-11-wireless-security.wep-key-type", ""),
|
|
||||||
("802-11-wireless-security.auth-alg", "open"),
|
|
||||||
]
|
|
||||||
|
|
||||||
# On Trixie, also disable PMF (Protected Management Frames)
|
|
||||||
# This can cause connection issues with certain WiFi adapters and clients
|
|
||||||
if self._is_trixie:
|
|
||||||
security_settings.append(("802-11-wireless-security.pmf", "disable"))
|
|
||||||
logger.info("Trixie detected: disabling PMF for better client compatibility")
|
|
||||||
|
|
||||||
for setting, value in security_settings:
|
|
||||||
result_modify = subprocess.run(
|
|
||||||
["nmcli", "connection", "modify", "LEDMatrix-Setup-AP", setting, str(value)],
|
|
||||||
capture_output=True,
|
|
||||||
text=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
if result_modify.returncode != 0:
|
|
||||||
logger.debug(f"Could not set {setting} to {value}: {result_modify.stderr}")
|
|
||||||
|
|
||||||
# On Trixie, set static IP address for the hotspot (default is 10.42.0.1)
|
|
||||||
# We want 192.168.4.1 for consistency
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "modify", "LEDMatrix-Setup-AP",
|
|
||||||
"ipv4.addresses", "192.168.4.1/24",
|
|
||||||
"ipv4.method", "shared"],
|
|
||||||
capture_output=True,
|
|
||||||
text=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
|
|
||||||
# Verify it's open
|
|
||||||
verify_result = subprocess.run(
|
|
||||||
["nmcli", "-t", "-f", "802-11-wireless-security.key-mgmt,802-11-wireless-security.psk", "connection", "show", "LEDMatrix-Setup-AP"],
|
|
||||||
capture_output=True,
|
|
||||||
text=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
|
|
||||||
if verify_result.returncode == 0:
|
|
||||||
output = verify_result.stdout.strip()
|
|
||||||
key_mgmt = ""
|
|
||||||
psk = ""
|
|
||||||
for line in output.split('\n'):
|
|
||||||
if 'key-mgmt:' in line:
|
|
||||||
key_mgmt = line.split(':', 1)[1].strip() if ':' in line else ""
|
|
||||||
elif 'psk:' in line:
|
|
||||||
psk = line.split(':', 1)[1].strip() if ':' in line else ""
|
|
||||||
|
|
||||||
if key_mgmt != "none" or (psk and psk != ""):
|
|
||||||
logger.warning(f"Hotspot still has security (key-mgmt={key_mgmt}, psk={'set' if psk else 'empty'}), deleting and recreating...")
|
|
||||||
# Delete and recreate as last resort
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "down", "LEDMatrix-Setup-AP"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "delete", "LEDMatrix-Setup-AP"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
time.sleep(1)
|
|
||||||
# Recreate without any password parameters
|
|
||||||
cmd_recreate = [
|
|
||||||
"nmcli", "device", "wifi", "hotspot",
|
|
||||||
"ifname", self._wifi_interface,
|
|
||||||
"con-name", "LEDMatrix-Setup-AP",
|
|
||||||
"ssid", ap_ssid,
|
|
||||||
"band", "bg",
|
|
||||||
"channel", str(ap_channel),
|
|
||||||
]
|
|
||||||
subprocess.run(cmd_recreate, capture_output=True, timeout=30)
|
|
||||||
# Set IP address for consistency
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "modify", "LEDMatrix-Setup-AP",
|
|
||||||
"ipv4.addresses", "192.168.4.1/24",
|
|
||||||
"ipv4.method", "shared"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
# Disable PMF on Trixie
|
|
||||||
if self._is_trixie:
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "modify", "LEDMatrix-Setup-AP",
|
|
||||||
"802-11-wireless-security.pmf", "disable"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
logger.info("Recreated hotspot as open network")
|
|
||||||
else:
|
|
||||||
logger.info("Hotspot verified as open (no password)")
|
|
||||||
|
|
||||||
# Restart the connection to apply all changes
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "down", "LEDMatrix-Setup-AP"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
time.sleep(1)
|
|
||||||
subprocess.run(
|
|
||||||
["nmcli", "connection", "up", "LEDMatrix-Setup-AP"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=10
|
|
||||||
)
|
|
||||||
logger.info("Hotspot restarted with open network settings")
|
|
||||||
logger.info(f"AP mode started via nmcli hotspot: {ap_ssid}")
|
|
||||||
time.sleep(2)
|
|
||||||
|
|
||||||
# Verify hotspot is running
|
|
||||||
status = self._get_ap_status_nmcli()
|
|
||||||
if status.get('active'):
|
|
||||||
ip = status.get('ip', '192.168.4.1')
|
|
||||||
logger.info(f"AP mode confirmed active at {ip}")
|
|
||||||
self._show_led_message(f"Setup: {ip}", duration=5)
|
|
||||||
return True, f"AP mode enabled (hotspot mode) - Access at {ip}:5000"
|
|
||||||
else:
|
|
||||||
logger.error("AP mode started but not verified")
|
|
||||||
return False, "AP mode started but verification failed"
|
|
||||||
else:
|
|
||||||
error_msg = result.stderr.strip() or result.stdout.strip()
|
error_msg = result.stderr.strip() or result.stdout.strip()
|
||||||
logger.error(f"Failed to start AP mode via nmcli: {error_msg}")
|
logger.error(f"Failed to create AP connection profile: {error_msg}")
|
||||||
self._show_led_message("AP mode failed", duration=5)
|
self._show_led_message("AP mode failed", duration=5)
|
||||||
return False, f"Failed to start AP mode: {error_msg}"
|
return False, f"Failed to create AP profile: {error_msg}"
|
||||||
|
|
||||||
|
# Write the NM dnsmasq-shared.d captive-portal config BEFORE bringing up
|
||||||
|
# the connection so NM's dnsmasq picks it up at start time.
|
||||||
|
# This causes every hostname DNS query from a connected device to resolve
|
||||||
|
# to 192.168.4.1, automatically triggering the OS captive-portal popup.
|
||||||
|
self._write_nm_dnsmasq_captive_conf()
|
||||||
|
|
||||||
|
logger.info("AP connection profile created, bringing it up...")
|
||||||
|
up_result = subprocess.run(
|
||||||
|
["nmcli", "connection", "up", "LEDMatrix-Setup-AP"],
|
||||||
|
capture_output=True, text=True, timeout=20
|
||||||
|
)
|
||||||
|
if up_result.returncode != 0:
|
||||||
|
error_msg = up_result.stderr.strip() or up_result.stdout.strip()
|
||||||
|
logger.error(f"Failed to bring up AP connection: {error_msg}")
|
||||||
|
self._remove_nm_dnsmasq_captive_conf()
|
||||||
|
subprocess.run(["nmcli", "connection", "delete", "LEDMatrix-Setup-AP"],
|
||||||
|
capture_output=True, timeout=10)
|
||||||
|
self._show_led_message("AP mode failed", duration=5)
|
||||||
|
return False, f"Failed to start AP: {error_msg}"
|
||||||
|
|
||||||
|
time.sleep(2)
|
||||||
|
|
||||||
|
# NM's ipv4.method=shared manages ip_forward automatically, so we only
|
||||||
|
# need to add the iptables port-redirect rules for the captive portal.
|
||||||
|
if not self._setup_iptables_redirect():
|
||||||
|
logger.error("Captive-portal redirect setup failed; rolling back AP profile")
|
||||||
|
self._remove_nm_dnsmasq_captive_conf()
|
||||||
|
subprocess.run(["nmcli", "connection", "down", "LEDMatrix-Setup-AP"],
|
||||||
|
capture_output=True, timeout=10)
|
||||||
|
subprocess.run(["nmcli", "connection", "delete", "LEDMatrix-Setup-AP"],
|
||||||
|
capture_output=True, timeout=10)
|
||||||
|
self._clear_led_message()
|
||||||
|
return False, "AP started but captive-portal redirect setup failed"
|
||||||
|
|
||||||
|
# Verify the AP is actually running
|
||||||
|
status = self._get_ap_status_nmcli()
|
||||||
|
if status.get('active'):
|
||||||
|
ip = status.get('ip', '192.168.4.1')
|
||||||
|
logger.info(f"AP mode confirmed active at {ip} (open network, no password)")
|
||||||
|
self._show_led_message(f"WiFi Setup\n{ap_ssid}\nNo password\n{ip}:5000", duration=10)
|
||||||
|
return True, f"AP mode enabled (open network) - Access at {ip}:5000"
|
||||||
|
else:
|
||||||
|
logger.error("AP mode started but not verified by status check — rolling back")
|
||||||
|
self._teardown_iptables_redirect()
|
||||||
|
self._remove_nm_dnsmasq_captive_conf()
|
||||||
|
subprocess.run(["nmcli", "connection", "down", "LEDMatrix-Setup-AP"],
|
||||||
|
capture_output=True, timeout=10)
|
||||||
|
subprocess.run(["nmcli", "connection", "delete", "LEDMatrix-Setup-AP"],
|
||||||
|
capture_output=True, timeout=10)
|
||||||
|
self._clear_led_message()
|
||||||
|
return False, "AP mode started but verification failed"
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error starting AP mode with nmcli hotspot: {e}")
|
logger.error(f"Error starting AP mode with nmcli: {e}")
|
||||||
|
self._remove_nm_dnsmasq_captive_conf()
|
||||||
self._show_led_message("Setup mode error", duration=5)
|
self._show_led_message("Setup mode error", duration=5)
|
||||||
return False, str(e)
|
return False, str(e)
|
||||||
|
|
||||||
@@ -1976,7 +2133,12 @@ class WiFiManager:
|
|||||||
|
|
||||||
for line in result.stdout.strip().split('\n'):
|
for line in result.stdout.strip().split('\n'):
|
||||||
parts = line.split(':')
|
parts = line.split(':')
|
||||||
if len(parts) >= 2 and 'hotspot' in parts[1].lower():
|
if len(parts) < 2:
|
||||||
|
continue
|
||||||
|
conn_name = parts[0].strip()
|
||||||
|
conn_type = parts[1].strip().lower()
|
||||||
|
# Match our known AP profile name OR the legacy nmcli hotspot type
|
||||||
|
if conn_name == "LEDMatrix-Setup-AP" or 'hotspot' in conn_type:
|
||||||
# Get actual IP address (may be 192.168.4.1 or 10.42.0.1 depending on config)
|
# Get actual IP address (may be 192.168.4.1 or 10.42.0.1 depending on config)
|
||||||
ip = '192.168.4.1'
|
ip = '192.168.4.1'
|
||||||
interface = parts[2] if len(parts) > 2 else self._wifi_interface
|
interface = parts[2] if len(parts) > 2 else self._wifi_interface
|
||||||
@@ -2072,45 +2234,9 @@ class WiFiManager:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"Could not remove dnsmasq drop-in config: {e}")
|
logger.warning(f"Could not remove dnsmasq drop-in config: {e}")
|
||||||
|
|
||||||
# Remove iptables port forwarding rules and disable IP forwarding (only for hostapd mode)
|
# Remove iptables redirect rules and restore ip_forward state (hostapd mode only)
|
||||||
if hostapd_active:
|
if hostapd_active:
|
||||||
try:
|
self._teardown_iptables_redirect()
|
||||||
# Check if iptables is available
|
|
||||||
iptables_check = subprocess.run(
|
|
||||||
["which", "iptables"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=2
|
|
||||||
)
|
|
||||||
|
|
||||||
if iptables_check.returncode == 0:
|
|
||||||
# Remove NAT redirect rule
|
|
||||||
subprocess.run(
|
|
||||||
["sudo", "iptables", "-t", "nat", "-D", "PREROUTING", "-i", self._wifi_interface, "-p", "tcp", "--dport", "80", "-j", "REDIRECT", "--to-port", "5000"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
|
|
||||||
# Remove INPUT rule
|
|
||||||
subprocess.run(
|
|
||||||
["sudo", "iptables", "-D", "INPUT", "-i", self._wifi_interface, "-p", "tcp", "--dport", "80", "-j", "ACCEPT"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
|
|
||||||
logger.info("Removed iptables port forwarding rules")
|
|
||||||
else:
|
|
||||||
logger.debug("iptables not available, skipping rule removal")
|
|
||||||
|
|
||||||
# Disable IP forwarding (restore to default client mode)
|
|
||||||
subprocess.run(
|
|
||||||
["sudo", "sysctl", "-w", "net.ipv4.ip_forward=0"],
|
|
||||||
capture_output=True,
|
|
||||||
timeout=5
|
|
||||||
)
|
|
||||||
logger.info("Disabled IP forwarding")
|
|
||||||
except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError) as e:
|
|
||||||
logger.warning(f"Could not remove iptables rules or disable forwarding: {e}")
|
|
||||||
# Continue anyway
|
|
||||||
|
|
||||||
# Clean up WiFi interface IP configuration
|
# Clean up WiFi interface IP configuration
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
@@ -2153,14 +2279,17 @@ class WiFiManager:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Final WiFi radio unblock attempt failed: {e}")
|
logger.error(f"Final WiFi radio unblock attempt failed: {e}")
|
||||||
else:
|
else:
|
||||||
# nmcli hotspot mode - restart not needed, just ensure WiFi radio is enabled
|
# nmcli AP mode — NM's ipv4.method=shared manages ip_forward automatically,
|
||||||
logger.info("Skipping NetworkManager restart (nmcli hotspot mode, restart not needed)")
|
# so we only need to remove the iptables redirect rules we added.
|
||||||
# Still ensure WiFi radio is enabled (may have been disabled by nmcli operations)
|
logger.info("Skipping NetworkManager restart (nmcli AP mode, restart not needed)")
|
||||||
# Use retries for safety
|
self._teardown_iptables_redirect()
|
||||||
|
self._remove_nm_dnsmasq_captive_conf()
|
||||||
|
# Ensure WiFi radio is enabled after nmcli operations
|
||||||
wifi_enabled = self._ensure_wifi_radio_enabled(max_retries=3)
|
wifi_enabled = self._ensure_wifi_radio_enabled(max_retries=3)
|
||||||
if not wifi_enabled:
|
if not wifi_enabled:
|
||||||
logger.warning("WiFi radio may be disabled after nmcli hotspot cleanup")
|
logger.warning("WiFi radio may be disabled after nmcli AP cleanup")
|
||||||
|
|
||||||
|
self._ap_enabled_at = None
|
||||||
logger.info("AP mode disabled successfully")
|
logger.info("AP mode disabled successfully")
|
||||||
return True, "AP mode disabled"
|
return True, "AP mode disabled"
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@@ -2176,8 +2305,10 @@ class WiFiManager:
|
|||||||
config_dir = HOSTAPD_CONFIG_PATH.parent
|
config_dir = HOSTAPD_CONFIG_PATH.parent
|
||||||
config_dir.mkdir(parents=True, exist_ok=True)
|
config_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
ap_ssid = self.config.get("ap_ssid", DEFAULT_AP_SSID)
|
# Use validated values — strips invalid chars and ensures channel is an int.
|
||||||
ap_channel = self.config.get("ap_channel", DEFAULT_AP_CHANNEL)
|
# Also strip newlines from SSID to prevent config-file injection.
|
||||||
|
ap_ssid, ap_channel = self._validate_ap_config()
|
||||||
|
ap_ssid = ap_ssid.replace('\n', '').replace('\r', '')
|
||||||
|
|
||||||
# Open network configuration (no password) for easy setup access
|
# Open network configuration (no password) for easy setup access
|
||||||
config_content = f"""interface={self._wifi_interface}
|
config_content = f"""interface={self._wifi_interface}
|
||||||
@@ -2305,12 +2436,11 @@ address=/detectportal.firefox.com/192.168.4.1
|
|||||||
f"Ethernet={ethernet_connected}, AP_active={ap_active}, "
|
f"Ethernet={ethernet_connected}, AP_active={ap_active}, "
|
||||||
f"auto_enable={auto_enable}, disconnected_checks={self._disconnected_checks}")
|
f"auto_enable={auto_enable}, disconnected_checks={self._disconnected_checks}")
|
||||||
|
|
||||||
# Determine if we should have AP mode active
|
# Determine if we should have AP mode active.
|
||||||
# AP mode should only be auto-enabled if:
|
# AP-enable uses only the nmcli association state (fast, no network calls).
|
||||||
# - auto_enable_ap_mode is True AND
|
# This keeps the same reliable behaviour as before: momentary packet loss
|
||||||
# - WiFi is NOT connected AND
|
# while on working WiFi does NOT trigger AP mode. The internet-reachability
|
||||||
# - Ethernet is NOT connected AND
|
# check is performed separately in the daemon watchdog for NM recovery.
|
||||||
# - We've had multiple consecutive disconnected checks (grace period)
|
|
||||||
is_disconnected = not status.connected and not ethernet_connected
|
is_disconnected = not status.connected and not ethernet_connected
|
||||||
|
|
||||||
if is_disconnected:
|
if is_disconnected:
|
||||||
@@ -2318,9 +2448,9 @@ address=/detectportal.firefox.com/192.168.4.1
|
|||||||
self._disconnected_checks += 1
|
self._disconnected_checks += 1
|
||||||
logger.debug(f"Network disconnected (check {self._disconnected_checks}/{self._disconnected_checks_required})")
|
logger.debug(f"Network disconnected (check {self._disconnected_checks}/{self._disconnected_checks_required})")
|
||||||
else:
|
else:
|
||||||
# Reset counter if we're connected
|
# Reset counter if we're associated
|
||||||
if self._disconnected_checks > 0:
|
if self._disconnected_checks > 0:
|
||||||
logger.debug(f"Network connected, resetting disconnected check counter")
|
logger.debug("Network connected, resetting disconnected check counter")
|
||||||
self._disconnected_checks = 0
|
self._disconnected_checks = 0
|
||||||
|
|
||||||
# Only enable AP if we've had enough consecutive disconnected checks
|
# Only enable AP if we've had enough consecutive disconnected checks
|
||||||
@@ -2366,6 +2496,24 @@ address=/detectportal.firefox.com/192.168.4.1
|
|||||||
# Don't disable it automatically, let it stay active
|
# Don't disable it automatically, let it stay active
|
||||||
logger.debug("AP mode is active (manually enabled), keeping active")
|
logger.debug("AP mode is active (manually enabled), keeping active")
|
||||||
|
|
||||||
|
# Idle-timeout check: disable AP if no client has connected within the window.
|
||||||
|
# Only applies when AP is active and we haven't just decided to enable/disable it.
|
||||||
|
if ap_active and self._ap_enabled_at is not None:
|
||||||
|
try:
|
||||||
|
idle_timeout_min = max(1, min(1440, int(self.config.get("ap_idle_timeout_minutes", 15))))
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
idle_timeout_min = 15
|
||||||
|
elapsed = time.time() - self._ap_enabled_at
|
||||||
|
if elapsed > idle_timeout_min * 60 and not self._has_ap_clients():
|
||||||
|
logger.info(
|
||||||
|
f"AP idle timeout ({idle_timeout_min} min, no clients) — disabling AP"
|
||||||
|
)
|
||||||
|
success, message = self.disable_ap_mode()
|
||||||
|
if success:
|
||||||
|
return True
|
||||||
|
else:
|
||||||
|
logger.warning(f"Failed to disable AP on idle timeout: {message}")
|
||||||
|
|
||||||
return False
|
return False
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error checking AP mode: {e}", exc_info=True)
|
logger.error(f"Error checking AP mode: {e}", exc_info=True)
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ Wants=network.target
|
|||||||
Type=simple
|
Type=simple
|
||||||
User=root
|
User=root
|
||||||
WorkingDirectory=__PROJECT_ROOT_DIR__
|
WorkingDirectory=__PROJECT_ROOT_DIR__
|
||||||
ExecStart=/usr/bin/python3 /home/ledpi/LEDMatrix/scripts/utils/wifi_monitor_daemon.py --interval 30
|
ExecStart=/usr/bin/python3 __PROJECT_ROOT_DIR__/scripts/utils/wifi_monitor_daemon.py --interval 30
|
||||||
Restart=on-failure
|
Restart=on-failure
|
||||||
RestartSec=10
|
RestartSec=10
|
||||||
StandardOutput=syslog
|
StandardOutput=syslog
|
||||||
|
|||||||
334
test/test_wifi_manager_ap.py
Normal file
334
test/test_wifi_manager_ap.py
Normal file
@@ -0,0 +1,334 @@
|
|||||||
|
"""
|
||||||
|
Unit tests for WiFi AP mode — src.wifi_manager.
|
||||||
|
|
||||||
|
Each test exercises logic that can be verified through the subprocess calls the
|
||||||
|
manager emits, without requiring root access, hardware, or a running Pi.
|
||||||
|
|
||||||
|
Scenarios covered:
|
||||||
|
1. nmcli AP profile is created with no security parameters (open/passwordless).
|
||||||
|
2. iptables PREROUTING and INPUT rules are added when the nmcli AP starts.
|
||||||
|
3. iptables rules and ip_forward are reverted when the AP is torn down.
|
||||||
|
4. LED matrix message includes the SSID, 'No password', and the setup URL.
|
||||||
|
5. Known AP profile names are deleted before the new profile is created.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from src.wifi_manager import WiFiManager
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Helpers
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _ok(stdout: str = "", stderr: str = "") -> MagicMock:
|
||||||
|
r = MagicMock()
|
||||||
|
r.returncode = 0
|
||||||
|
r.stdout = stdout
|
||||||
|
r.stderr = stderr
|
||||||
|
return r
|
||||||
|
|
||||||
|
|
||||||
|
def _fail(stdout: str = "", stderr: str = "error") -> MagicMock:
|
||||||
|
r = MagicMock()
|
||||||
|
r.returncode = 1
|
||||||
|
r.stdout = stdout
|
||||||
|
r.stderr = stderr
|
||||||
|
return r
|
||||||
|
|
||||||
|
|
||||||
|
def _find_path_side_effect(name: str) -> str:
|
||||||
|
"""Deterministic fake for _find_command_path."""
|
||||||
|
return {"iptables": "/usr/sbin/iptables", "sysctl": "/usr/sbin/sysctl"}.get(
|
||||||
|
name, f"/usr/bin/{name}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Fixtures
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def wifi_config(tmp_path: Path) -> Path:
|
||||||
|
"""Minimal wifi_config.json in a temporary directory."""
|
||||||
|
cfg_dir = tmp_path / "config"
|
||||||
|
cfg_dir.mkdir()
|
||||||
|
cfg = {
|
||||||
|
"ap_ssid": "LEDMatrix-Setup",
|
||||||
|
"ap_channel": 7,
|
||||||
|
"auto_enable_ap_mode": True,
|
||||||
|
"saved_networks": [],
|
||||||
|
}
|
||||||
|
p = cfg_dir / "wifi_config.json"
|
||||||
|
p.write_text(json.dumps(cfg))
|
||||||
|
return p
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def manager(wifi_config: Path, tmp_path: Path) -> WiFiManager:
|
||||||
|
"""
|
||||||
|
WiFiManager with all system calls stubbed out during construction and the
|
||||||
|
ip_forward save file redirected to a per-test temporary path.
|
||||||
|
"""
|
||||||
|
with patch("src.wifi_manager.subprocess.run", return_value=_ok(stdout="wlan0\n")), \
|
||||||
|
patch.object(WiFiManager, "_detect_trixie", return_value=False):
|
||||||
|
mgr = WiFiManager(config_path=wifi_config)
|
||||||
|
|
||||||
|
# Force clean, deterministic state regardless of what __init__ inferred
|
||||||
|
mgr._wifi_interface = "wlan0"
|
||||||
|
mgr.has_nmcli = True
|
||||||
|
mgr.has_hostapd = False
|
||||||
|
mgr.has_dnsmasq = False
|
||||||
|
mgr.has_iwlist = False
|
||||||
|
mgr._is_trixie = False
|
||||||
|
# Redirect the ip_forward save file to tmp so tests never share state
|
||||||
|
mgr._IP_FORWARD_SAVE_PATH = tmp_path / "ip_fwd_saved"
|
||||||
|
return mgr
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 1. AP profile is open (no password)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_nmcli_ap_profile_has_no_security_params(manager: WiFiManager) -> None:
|
||||||
|
"""
|
||||||
|
The 'nmcli connection add' command must not include key-mgmt, psk, or any
|
||||||
|
WPA-related parameter. On Bookworm/Trixie, NM creates a WPA2-protected
|
||||||
|
hotspot even when those values are set to 'none'/empty via a later
|
||||||
|
'connection modify', so the profile must be created without a security
|
||||||
|
section from the start.
|
||||||
|
"""
|
||||||
|
captured: list[list[str]] = []
|
||||||
|
|
||||||
|
def _run(cmd, **kw):
|
||||||
|
captured.append(list(cmd))
|
||||||
|
return _ok()
|
||||||
|
|
||||||
|
with patch("src.wifi_manager.subprocess.run", side_effect=_run), \
|
||||||
|
patch.object(manager, "disconnect_from_network", return_value=(True, "ok")), \
|
||||||
|
patch.object(manager, "_setup_iptables_redirect", return_value=True), \
|
||||||
|
patch.object(manager, "_get_ap_status_nmcli",
|
||||||
|
return_value={"active": True, "ip": "192.168.4.1"}), \
|
||||||
|
patch.object(manager, "_show_led_message"):
|
||||||
|
|
||||||
|
success, _ = manager._enable_ap_mode_nmcli_hotspot()
|
||||||
|
|
||||||
|
assert success, "AP enable should report success"
|
||||||
|
|
||||||
|
add_calls = [c for c in captured if "nmcli" in c and "connection" in c and "add" in c]
|
||||||
|
assert add_calls, "Expected at least one 'nmcli connection add' invocation"
|
||||||
|
|
||||||
|
add_str = " ".join(add_calls[0])
|
||||||
|
assert "key-mgmt" not in add_str, "AP profile must not set key-mgmt"
|
||||||
|
assert "psk" not in add_str, "AP profile must not include a PSK/password"
|
||||||
|
assert "wpa" not in add_str.lower(), "AP profile must not reference WPA"
|
||||||
|
assert "802-11-wireless.mode" in add_str, "AP profile must declare wireless mode"
|
||||||
|
# Verify the value for 802-11-wireless.mode is exactly "ap" — check the element
|
||||||
|
# that immediately follows the key in the command list, not a loose substring match.
|
||||||
|
cmd = add_calls[0]
|
||||||
|
try:
|
||||||
|
mode_idx = cmd.index("802-11-wireless.mode")
|
||||||
|
assert cmd[mode_idx + 1] == "ap", \
|
||||||
|
f"802-11-wireless.mode value must be exactly 'ap', got {cmd[mode_idx + 1]!r}"
|
||||||
|
except ValueError:
|
||||||
|
pytest.fail("802-11-wireless.mode not found as a list element in nmcli command")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 2. iptables NAT rules are added when the AP starts
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_iptables_nat_rules_added_on_ap_start(manager: WiFiManager) -> None:
|
||||||
|
"""
|
||||||
|
_setup_iptables_redirect must add:
|
||||||
|
- a PREROUTING REDIRECT rule that maps incoming TCP port 80 to port 5000, and
|
||||||
|
- an INPUT ACCEPT rule for port 5000 (the post-redirect destination port,
|
||||||
|
NOT port 80 which never hits the INPUT chain after PREROUTING rewrites it).
|
||||||
|
"""
|
||||||
|
captured: list[list[str]] = []
|
||||||
|
|
||||||
|
def _run(cmd, **kw):
|
||||||
|
captured.append(list(cmd))
|
||||||
|
# iptables -C (check) → rc=1 so the -A (add) branch executes
|
||||||
|
if "iptables" in " ".join(str(x) for x in cmd) and "-C" in cmd:
|
||||||
|
return _fail()
|
||||||
|
return _ok()
|
||||||
|
|
||||||
|
# Patch Path.read_text so /proc/sys/net/ipv4/ip_forward is readable on any OS
|
||||||
|
with patch("src.wifi_manager.subprocess.run", side_effect=_run), \
|
||||||
|
patch.object(manager, "_find_command_path", side_effect=_find_path_side_effect), \
|
||||||
|
patch("pathlib.Path.read_text", return_value="0\n"):
|
||||||
|
|
||||||
|
result = manager._setup_iptables_redirect()
|
||||||
|
|
||||||
|
assert result, "_setup_iptables_redirect must return True on success"
|
||||||
|
|
||||||
|
prerouting_adds = [c for c in captured if "iptables" in " ".join(c) and "-A" in c and "PREROUTING" in c]
|
||||||
|
assert prerouting_adds, "Expected 'iptables -A PREROUTING' invocation"
|
||||||
|
pr_str = " ".join(prerouting_adds[0])
|
||||||
|
assert "--dport" in pr_str and "80" in pr_str, "PREROUTING rule must match dport 80"
|
||||||
|
assert "5000" in pr_str, "PREROUTING rule must redirect to port 5000"
|
||||||
|
assert "REDIRECT" in pr_str, "PREROUTING rule must use REDIRECT target"
|
||||||
|
|
||||||
|
input_adds = [c for c in captured if "iptables" in " ".join(c) and "-A" in c and "INPUT" in c]
|
||||||
|
assert input_adds, "Expected 'iptables -A INPUT' invocation"
|
||||||
|
in_str = " ".join(input_adds[0])
|
||||||
|
assert "5000" in in_str, "INPUT rule must accept port 5000 (post-PREROUTING destination)"
|
||||||
|
assert "ACCEPT" in in_str, "INPUT rule must use ACCEPT target"
|
||||||
|
|
||||||
|
# Port 80 must NOT be used in the INPUT rule (it is already redirected by PREROUTING)
|
||||||
|
input_80 = [c for c in captured if "iptables" in " ".join(c) and "INPUT" in c and "--dport" in c and "80" in c]
|
||||||
|
assert not input_80, "INPUT rule must target port 5000, not port 80"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 3a. iptables rules and ip_forward reverted on teardown
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_iptables_rules_and_ip_forward_reverted_on_teardown(manager: WiFiManager) -> None:
|
||||||
|
"""
|
||||||
|
_teardown_iptables_redirect must:
|
||||||
|
- remove the PREROUTING and INPUT iptables rules, and
|
||||||
|
- restore ip_forward to the exact value recorded in the save file.
|
||||||
|
"""
|
||||||
|
original_fwd = "0"
|
||||||
|
manager._IP_FORWARD_SAVE_PATH.write_text(original_fwd)
|
||||||
|
# Teardown dispatches on the backend recorded during setup
|
||||||
|
manager._redirect_backend = "iptables"
|
||||||
|
|
||||||
|
captured: list[list[str]] = []
|
||||||
|
|
||||||
|
with patch("src.wifi_manager.subprocess.run",
|
||||||
|
side_effect=lambda cmd, **kw: (captured.append(list(cmd)) or _ok())), \
|
||||||
|
patch.object(manager, "_find_command_path", side_effect=_find_path_side_effect):
|
||||||
|
|
||||||
|
manager._teardown_iptables_redirect()
|
||||||
|
|
||||||
|
assert [c for c in captured if "iptables" in " ".join(c) and "-D" in c and "PREROUTING" in c], \
|
||||||
|
"Expected 'iptables -D PREROUTING' invocation"
|
||||||
|
|
||||||
|
assert [c for c in captured if "iptables" in " ".join(c) and "-D" in c and "INPUT" in c], \
|
||||||
|
"Expected 'iptables -D INPUT' invocation"
|
||||||
|
|
||||||
|
restore_calls = [
|
||||||
|
c for c in captured
|
||||||
|
if "sysctl" in " ".join(c) and f"ip_forward={original_fwd}" in " ".join(c)
|
||||||
|
]
|
||||||
|
assert restore_calls, f"Expected sysctl to restore ip_forward to {original_fwd!r}"
|
||||||
|
|
||||||
|
assert not manager._IP_FORWARD_SAVE_PATH.exists(), \
|
||||||
|
"Save file must be removed after successful teardown"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 3b. ip_forward untouched when no save file exists
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_ip_forward_not_restored_when_save_file_absent(manager: WiFiManager) -> None:
|
||||||
|
"""
|
||||||
|
When the save file is missing (setup never wrote it, e.g. because /proc was
|
||||||
|
unreadable or the write failed), teardown must NOT call sysctl so it does not
|
||||||
|
accidentally clobber ip_forward state owned by a VPN or NetworkManager.
|
||||||
|
"""
|
||||||
|
assert not manager._IP_FORWARD_SAVE_PATH.exists()
|
||||||
|
|
||||||
|
captured: list[list[str]] = []
|
||||||
|
|
||||||
|
with patch("src.wifi_manager.subprocess.run",
|
||||||
|
side_effect=lambda cmd, **kw: (captured.append(list(cmd)) or _ok())), \
|
||||||
|
patch.object(manager, "_find_command_path", side_effect=_find_path_side_effect):
|
||||||
|
|
||||||
|
manager._teardown_iptables_redirect()
|
||||||
|
|
||||||
|
sysctl_calls = [
|
||||||
|
c for c in captured
|
||||||
|
if "sysctl" in " ".join(c) and "ip_forward" in " ".join(c)
|
||||||
|
]
|
||||||
|
assert not sysctl_calls, \
|
||||||
|
"sysctl must not be called when no ip_forward save file exists"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 4. LED message content
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_led_message_shows_ssid_no_password_and_url(manager: WiFiManager) -> None:
|
||||||
|
"""
|
||||||
|
When the nmcli AP activates, the LED message must include:
|
||||||
|
- the AP SSID ('LEDMatrix-Setup')
|
||||||
|
- the string 'No password'
|
||||||
|
- the AP IP address (192.168.4.1) and Flask port (5000)
|
||||||
|
"""
|
||||||
|
led_messages: list[str] = []
|
||||||
|
|
||||||
|
with patch("src.wifi_manager.subprocess.run", return_value=_ok()), \
|
||||||
|
patch.object(manager, "disconnect_from_network", return_value=(True, "ok")), \
|
||||||
|
patch.object(manager, "_setup_iptables_redirect", return_value=True), \
|
||||||
|
patch.object(manager, "_get_ap_status_nmcli",
|
||||||
|
return_value={"active": True, "ip": "192.168.4.1"}), \
|
||||||
|
patch.object(manager, "_show_led_message",
|
||||||
|
side_effect=lambda msg, **kw: led_messages.append(msg)):
|
||||||
|
|
||||||
|
success, _ = manager._enable_ap_mode_nmcli_hotspot()
|
||||||
|
|
||||||
|
assert success, "AP enable should report success"
|
||||||
|
assert led_messages, "Expected at least one _show_led_message call"
|
||||||
|
|
||||||
|
combined = "\n".join(led_messages)
|
||||||
|
assert "No password" in combined, "LED message must say 'No password'"
|
||||||
|
assert "LEDMatrix-Setup" in combined, "LED message must include the AP SSID"
|
||||||
|
assert "192.168.4.1" in combined, "LED message must include the AP IP address"
|
||||||
|
assert "5000" in combined, "LED message must include the Flask port"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 5. Stale AP profiles deleted before the new one is created
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_existing_ap_profiles_deleted_before_new_profile_created(manager: WiFiManager) -> None:
|
||||||
|
"""
|
||||||
|
Before 'nmcli connection add', the manager must issue
|
||||||
|
'nmcli connection down/delete' for every known AP profile name so stale
|
||||||
|
profiles (from a previous crash or partial setup) cannot block the new one.
|
||||||
|
"""
|
||||||
|
captured: list[list[str]] = []
|
||||||
|
|
||||||
|
def _run(cmd, **kw):
|
||||||
|
captured.append(list(cmd))
|
||||||
|
return _ok()
|
||||||
|
|
||||||
|
with patch("src.wifi_manager.subprocess.run", side_effect=_run), \
|
||||||
|
patch.object(manager, "disconnect_from_network", return_value=(True, "ok")), \
|
||||||
|
patch.object(manager, "_setup_iptables_redirect", return_value=True), \
|
||||||
|
patch.object(manager, "_get_ap_status_nmcli",
|
||||||
|
return_value={"active": True, "ip": "192.168.4.1"}), \
|
||||||
|
patch.object(manager, "_show_led_message"):
|
||||||
|
|
||||||
|
success, _ = manager._enable_ap_mode_nmcli_hotspot()
|
||||||
|
|
||||||
|
assert success
|
||||||
|
|
||||||
|
cmd_strs = [" ".join(c) for c in captured]
|
||||||
|
|
||||||
|
for profile in ("LEDMatrix-Setup-AP", "Hotspot", "TickerSetup-AP"):
|
||||||
|
assert any("connection delete" in s and profile in s for s in cmd_strs), \
|
||||||
|
f"Expected 'nmcli connection delete {profile}' before creating the new profile"
|
||||||
|
|
||||||
|
add_indices = [i for i, s in enumerate(cmd_strs) if "connection add" in s]
|
||||||
|
del_indices = [i for i, s in enumerate(cmd_strs) if "connection delete" in s]
|
||||||
|
|
||||||
|
assert add_indices, "Expected 'nmcli connection add' call"
|
||||||
|
assert del_indices, "Expected 'nmcli connection delete' calls"
|
||||||
|
assert max(del_indices) < min(add_indices), \
|
||||||
|
"All connection deletions must complete before the new profile is created"
|
||||||
@@ -218,7 +218,7 @@ def _ensure_display_service_running():
|
|||||||
if status.get('active'):
|
if status.get('active'):
|
||||||
status['started'] = False
|
status['started'] = False
|
||||||
return status
|
return status
|
||||||
result = _run_systemctl_command(['sudo', 'systemctl', 'start', 'ledmatrix'])
|
result = _run_systemctl_command(['sudo', 'systemctl', 'start', 'ledmatrix.service'])
|
||||||
service_status = _get_display_service_status()
|
service_status = _get_display_service_status()
|
||||||
result['started'] = result.get('returncode') == 0
|
result['started'] = result.get('returncode') == 0
|
||||||
result['active'] = service_status.get('active')
|
result['active'] = service_status.get('active')
|
||||||
@@ -227,7 +227,7 @@ def _ensure_display_service_running():
|
|||||||
|
|
||||||
def _stop_display_service():
|
def _stop_display_service():
|
||||||
"""Stop the ledmatrix display service."""
|
"""Stop the ledmatrix display service."""
|
||||||
result = _run_systemctl_command(['sudo', 'systemctl', 'stop', 'ledmatrix'])
|
result = _run_systemctl_command(['sudo', 'systemctl', 'stop', 'ledmatrix.service'])
|
||||||
status = _get_display_service_status()
|
status = _get_display_service_status()
|
||||||
result['active'] = status.get('active')
|
result['active'] = status.get('active')
|
||||||
result['status'] = status
|
result['status'] = status
|
||||||
@@ -1716,33 +1716,34 @@ def execute_system_action():
|
|||||||
if mode:
|
if mode:
|
||||||
# For on-demand modes, we would need to integrate with the display controller
|
# For on-demand modes, we would need to integrate with the display controller
|
||||||
# For now, just start the display service
|
# For now, just start the display service
|
||||||
result = subprocess.run(['sudo', 'systemctl', 'start', 'ledmatrix'],
|
result = subprocess.run(['sudo', 'systemctl', 'start', 'ledmatrix.service'],
|
||||||
capture_output=True, text=True)
|
capture_output=True, text=True, timeout=15)
|
||||||
return jsonify({
|
return jsonify({
|
||||||
'status': 'success' if result.returncode == 0 else 'error',
|
'status': 'success' if result.returncode == 0 else 'error',
|
||||||
'message': f'Started display in {mode} mode',
|
'message': f'Started display in {mode} mode' if result.returncode == 0
|
||||||
|
else f'Failed to start display in {mode} mode: {result.stderr.strip() or "check sudo systemctl status ledmatrix.service"}',
|
||||||
'returncode': result.returncode,
|
'returncode': result.returncode,
|
||||||
'stdout': result.stdout,
|
'stdout': result.stdout,
|
||||||
'stderr': result.stderr
|
'stderr': result.stderr
|
||||||
})
|
})
|
||||||
else:
|
else:
|
||||||
result = subprocess.run(['sudo', 'systemctl', 'start', 'ledmatrix'],
|
result = subprocess.run(['sudo', 'systemctl', 'start', 'ledmatrix.service'],
|
||||||
capture_output=True, text=True)
|
capture_output=True, text=True, timeout=15)
|
||||||
elif action == 'stop_display':
|
elif action == 'stop_display':
|
||||||
result = subprocess.run(['sudo', 'systemctl', 'stop', 'ledmatrix'],
|
result = subprocess.run(['sudo', 'systemctl', 'stop', 'ledmatrix.service'],
|
||||||
capture_output=True, text=True)
|
capture_output=True, text=True, timeout=15)
|
||||||
elif action == 'enable_autostart':
|
elif action == 'enable_autostart':
|
||||||
result = subprocess.run(['sudo', 'systemctl', 'enable', 'ledmatrix'],
|
result = subprocess.run(['sudo', 'systemctl', 'enable', 'ledmatrix.service'],
|
||||||
capture_output=True, text=True)
|
capture_output=True, text=True, timeout=15)
|
||||||
elif action == 'disable_autostart':
|
elif action == 'disable_autostart':
|
||||||
result = subprocess.run(['sudo', 'systemctl', 'disable', 'ledmatrix'],
|
result = subprocess.run(['sudo', 'systemctl', 'disable', 'ledmatrix.service'],
|
||||||
capture_output=True, text=True)
|
capture_output=True, text=True, timeout=15)
|
||||||
elif action == 'reboot_system':
|
elif action == 'reboot_system':
|
||||||
result = subprocess.run(['sudo', 'reboot'],
|
result = subprocess.run(['sudo', 'reboot'],
|
||||||
capture_output=True, text=True)
|
capture_output=True, text=True, timeout=10)
|
||||||
elif action == 'shutdown_system':
|
elif action == 'shutdown_system':
|
||||||
result = subprocess.run(['sudo', 'poweroff'],
|
result = subprocess.run(['sudo', 'poweroff'],
|
||||||
capture_output=True, text=True)
|
capture_output=True, text=True, timeout=10)
|
||||||
elif action == 'git_pull':
|
elif action == 'git_pull':
|
||||||
# Use PROJECT_ROOT instead of hardcoded path
|
# Use PROJECT_ROOT instead of hardcoded path
|
||||||
project_dir = str(PROJECT_ROOT)
|
project_dir = str(PROJECT_ROOT)
|
||||||
@@ -1823,12 +1824,11 @@ def execute_system_action():
|
|||||||
'stderr': result.stderr
|
'stderr': result.stderr
|
||||||
})
|
})
|
||||||
elif action == 'restart_display_service':
|
elif action == 'restart_display_service':
|
||||||
result = subprocess.run(['sudo', 'systemctl', 'restart', 'ledmatrix'],
|
result = subprocess.run(['sudo', 'systemctl', 'restart', 'ledmatrix.service'],
|
||||||
capture_output=True, text=True)
|
capture_output=True, text=True, timeout=15)
|
||||||
elif action == 'restart_web_service':
|
elif action == 'restart_web_service':
|
||||||
# Try to restart the web service (assuming it's ledmatrix-web.service)
|
result = subprocess.run(['sudo', 'systemctl', 'restart', 'ledmatrix-web.service'],
|
||||||
result = subprocess.run(['sudo', 'systemctl', 'restart', 'ledmatrix-web'],
|
capture_output=True, text=True, timeout=15)
|
||||||
capture_output=True, text=True)
|
|
||||||
else:
|
else:
|
||||||
return jsonify({'status': 'error', 'message': f'Unknown action: {action}'}), 400
|
return jsonify({'status': 'error', 'message': f'Unknown action: {action}'}), 400
|
||||||
|
|
||||||
@@ -7133,9 +7133,14 @@ def connect_wifi():
|
|||||||
'message': message
|
'message': message
|
||||||
})
|
})
|
||||||
else:
|
else:
|
||||||
|
# Propagate structured error type so the captive portal UI can show
|
||||||
|
# "Wrong password — try again" instead of a generic failure message.
|
||||||
|
error_type = "wrong_password" if (message or "").startswith("wrong_password:") else "connection_failed"
|
||||||
|
clean_message = (message or "").removeprefix("wrong_password:").lstrip() or "Failed to connect to network"
|
||||||
return jsonify({
|
return jsonify({
|
||||||
'status': 'error',
|
'status': 'error',
|
||||||
'message': message or 'Failed to connect to network'
|
'message': clean_message,
|
||||||
|
'error_type': error_type
|
||||||
}), 400
|
}), 400
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.exception("[WiFi] Failed connecting to WiFi network")
|
logger.exception("[WiFi] Failed connecting to WiFi network")
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify
|
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
from html import escape as html_escape
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from src.web_interface.secret_helpers import mask_secret_fields
|
from src.web_interface.secret_helpers import mask_secret_fields
|
||||||
|
|
||||||
@@ -354,7 +355,7 @@ def _load_plugin_config_partial(plugin_id):
|
|||||||
plugin_info = pages_v3.plugin_manager.get_plugin_info(plugin_id)
|
plugin_info = pages_v3.plugin_manager.get_plugin_info(plugin_id)
|
||||||
|
|
||||||
if not plugin_info:
|
if not plugin_info:
|
||||||
return f'<div class="text-red-500 p-4">Plugin "{plugin_id}" not found</div>', 404
|
return f'<div class="text-red-500 p-4">Plugin "{html_escape(plugin_id)}" not found</div>', 404
|
||||||
|
|
||||||
# Get plugin instance (may be None if not loaded)
|
# Get plugin instance (may be None if not loaded)
|
||||||
plugin_instance = pages_v3.plugin_manager.get_plugin(plugin_id)
|
plugin_instance = pages_v3.plugin_manager.get_plugin(plugin_id)
|
||||||
|
|||||||
@@ -91,7 +91,7 @@
|
|||||||
const xOptions = config['x-options'] || config['x_options'] || {};
|
const xOptions = config['x-options'] || config['x_options'] || {};
|
||||||
const requestedFormat = xOptions.format || 'long';
|
const requestedFormat = xOptions.format || 'long';
|
||||||
// Validate format exists in DAY_LABELS, default to 'long' if not
|
// Validate format exists in DAY_LABELS, default to 'long' if not
|
||||||
const format = DAY_LABELS.hasOwnProperty(requestedFormat) ? requestedFormat : 'long';
|
const format = Object.prototype.hasOwnProperty.call(DAY_LABELS, requestedFormat) ? requestedFormat : 'long';
|
||||||
const layout = xOptions.layout || 'horizontal';
|
const layout = xOptions.layout || 'horizontal';
|
||||||
const showSelectAll = xOptions.selectAll !== false;
|
const showSelectAll = xOptions.selectAll !== false;
|
||||||
|
|
||||||
|
|||||||
@@ -898,6 +898,10 @@ window.currentPluginConfig = null;
|
|||||||
const CACHE_DURATION = 5 * 60 * 1000; // 5 minutes in milliseconds
|
const CACHE_DURATION = 5 * 60 * 1000; // 5 minutes in milliseconds
|
||||||
let storeFilteredList = [];
|
let storeFilteredList = [];
|
||||||
|
|
||||||
|
function storeCacheExpired() {
|
||||||
|
return !cacheTimestamp || (Date.now() - cacheTimestamp >= CACHE_DURATION);
|
||||||
|
}
|
||||||
|
|
||||||
// ── Plugin Store Filter State ───────────────────────────────────────────
|
// ── Plugin Store Filter State ───────────────────────────────────────────
|
||||||
const storeFilterState = {
|
const storeFilterState = {
|
||||||
sort: safeLocalStorage.getItem('storeSort') || 'a-z',
|
sort: safeLocalStorage.getItem('storeSort') || 'a-z',
|
||||||
@@ -1214,12 +1218,18 @@ function initializePlugins() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Load both installed plugins and plugin store.
|
// Load both installed plugins and plugin store.
|
||||||
// On HTMX re-swaps use cached store data (fetchCommitInfo=false) to avoid
|
// On HTMX re-swaps with a still-warm cache, skip GitHub metadata to avoid
|
||||||
// re-hitting GitHub on every tab switch; only fetch fresh on first load.
|
// re-hitting the API on every tab switch. If the cache TTL has expired even
|
||||||
const isReswap = !!window.pluginManager._reswap;
|
// during a re-swap, fetch fresh data including GitHub commit/version info.
|
||||||
|
const isReswapWarm = !!window.pluginManager._reswap && !storeCacheExpired();
|
||||||
window.pluginManager._reswap = false;
|
window.pluginManager._reswap = false;
|
||||||
loadInstalledPlugins();
|
// Await the installed-plugins fetch so window.installedPlugins is populated before
|
||||||
searchPluginStore(!isReswap);
|
// searchPluginStore renders Installed/Reinstall badges against it.
|
||||||
|
loadInstalledPlugins().then(() => {
|
||||||
|
searchPluginStore(!isReswapWarm);
|
||||||
|
}).catch(err => {
|
||||||
|
console.error('[PluginStore] loadInstalledPlugins failed:', err);
|
||||||
|
});
|
||||||
|
|
||||||
// Setup search functionality (with guard against duplicate listeners)
|
// Setup search functionality (with guard against duplicate listeners)
|
||||||
const searchInput = document.getElementById('plugin-search');
|
const searchInput = document.getElementById('plugin-search');
|
||||||
@@ -5133,10 +5143,13 @@ function refreshPlugins() {
|
|||||||
pluginStoreCache = null;
|
pluginStoreCache = null;
|
||||||
cacheTimestamp = null;
|
cacheTimestamp = null;
|
||||||
|
|
||||||
refreshInstalledPlugins(); // invalidates cache before fetching
|
// refreshInstalledPlugins() is async (returns a Promise via loadInstalledPlugins).
|
||||||
// Fetch latest metadata from GitHub when refreshing
|
// Only search the store and notify after window.installedPlugins is updated so
|
||||||
searchPluginStore(true);
|
// that Installed/Reinstall badges reflect the freshly fetched state.
|
||||||
showNotification('Plugins refreshed with latest metadata from GitHub', 'success');
|
refreshInstalledPlugins().then(() => {
|
||||||
|
searchPluginStore(true);
|
||||||
|
showNotification('Plugins refreshed with latest metadata from GitHub', 'success');
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
function restartDisplay() {
|
function restartDisplay() {
|
||||||
|
|||||||
@@ -191,7 +191,10 @@ function doConnect() {
|
|||||||
// Poll for the new IP
|
// Poll for the new IP
|
||||||
setTimeout(function() { checkNewIP(ssid); }, 3000);
|
setTimeout(function() { checkNewIP(ssid); }, 3000);
|
||||||
} else {
|
} else {
|
||||||
showMsg(data.message || 'Connection failed', 'err');
|
var msg = data.error_type === 'wrong_password'
|
||||||
|
? 'Incorrect password — please try again'
|
||||||
|
: (data.message || 'Connection failed');
|
||||||
|
showMsg(msg, 'err');
|
||||||
connecting = false;
|
connecting = false;
|
||||||
btn.disabled = false;
|
btn.disabled = false;
|
||||||
btn.innerHTML = 'Connect';
|
btn.innerHTML = 'Connect';
|
||||||
|
|||||||
Reference in New Issue
Block a user