Support wpa_supplicant config format on OPi/Armbian targets

- Add detect_wifi_config_type() to identify config format (netplan vs wpa_supplicant)
- Add read_wpa_supplicant() to parse wpa_supplicant-wlan0.conf
- Add write_wpa_supplicant() to write updated config with proper formatting
- Update run_interactive(), run_list(), run_add() to detect and dispatch to correct handler
- Both formats now auto-detected; existing netplan paths continue to work
- wpa_supplicant format used by OPi Zero 3 / Armbian defaults
This commit is contained in:
Cobra
2026-04-14 13:14:35 -04:00
parent b67127677f
commit cfa2d9e0e7
+368 -56
View File
@@ -30,6 +30,7 @@ from rich.text import Text
console = Console() console = Console()
NETPLAN_PATH = "etc/netplan/20-wifi.yaml" NETPLAN_PATH = "etc/netplan/20-wifi.yaml"
WPA_SUPPLICANT_PATH = "etc/wpa_supplicant/wpa_supplicant-wlan0.conf"
SD_LABEL = "armbi_root" SD_LABEL = "armbi_root"
@@ -83,6 +84,23 @@ def unmount_sd(mountpoint: str) -> None:
) )
def detect_wifi_config_type(mountpoint: str) -> str:
"""Detect the WiFi config format used on the mounted SD card.
Returns 'wpa_supplicant' if wpa_supplicant config is found,
'netplan' if netplan dir exists, otherwise 'netplan' (default).
"""
wpa_path = Path(mountpoint) / WPA_SUPPLICANT_PATH
if wpa_path.exists():
return "wpa_supplicant"
netplan_dir = Path(mountpoint) / "etc/netplan"
if netplan_dir.exists():
return "netplan"
return "netplan"
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Netplan read / write # Netplan read / write
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -158,6 +176,133 @@ def _deep_copy(obj):
return json.loads(json.dumps(obj)) return json.loads(json.dumps(obj))
# ---------------------------------------------------------------------------
# WPA Supplicant read / write
# ---------------------------------------------------------------------------
def read_wpa_supplicant(mountpoint: str) -> Tuple[Dict[str, Dict], str]:
"""Read and parse wpa_supplicant config.
Returns (aps_dict, country_code) where aps_dict is {ssid: {"password": "..."}}
or {ssid: {}} for open networks. Country code extracted from 'country=XX' line.
"""
wpa_path = Path(mountpoint) / WPA_SUPPLICANT_PATH
country = "US"
aps = {}
if not wpa_path.exists():
return aps, country
try:
with open(wpa_path, "r") as fh:
content = fh.read()
except Exception:
return aps, country
# Parse country line
for line in content.split("\n"):
line = line.strip()
if line.startswith("country="):
country = line.split("=", 1)[1].strip()
break
# Parse network blocks
in_network = False
current_ssid = None
current_psk = None
for line in content.split("\n"):
line = line.strip()
if line == "network={":
in_network = True
current_ssid = None
current_psk = None
continue
if in_network:
if line == "}":
if current_ssid:
aps[current_ssid] = {"password": current_psk} if current_psk else {}
in_network = False
current_ssid = None
current_psk = None
continue
if line.startswith("ssid="):
# Strip quotes
ssid_val = line.split("=", 1)[1].strip()
if ssid_val.startswith('"') and ssid_val.endswith('"'):
ssid_val = ssid_val[1:-1]
current_ssid = ssid_val
elif line.startswith("psk="):
# Strip quotes
psk_val = line.split("=", 1)[1].strip()
if psk_val.startswith('"') and psk_val.endswith('"'):
psk_val = psk_val[1:-1]
current_psk = psk_val
return aps, country
def write_wpa_supplicant(mountpoint: str, aps: Dict[str, Dict], country: str) -> None:
"""Write wpa_supplicant config file.
Args:
mountpoint: Root mountpoint of the SD card
aps: {ssid: {"password": "..."}} or {ssid: {}} for open networks
country: 2-letter country code
"""
wpa_path = Path(mountpoint) / WPA_SUPPLICANT_PATH
# Ensure the directory exists
wpa_path.parent.mkdir(parents=True, exist_ok=True)
# Build the config content
lines = [
"ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev",
"update_config=1",
f"country={country.upper()}",
"",
]
for ssid, details in aps.items():
password = None
if isinstance(details, dict):
password = details.get("password")
lines.append("network={")
lines.append(f' ssid="{ssid}"')
if password:
lines.append(f' psk="{password}"')
lines.append(" key_mgmt=WPA-PSK")
else:
lines.append(" key_mgmt=NONE")
lines.append(" scan_ssid=1")
lines.append("}")
lines.append("")
config_text = "\n".join(lines)
# Write via sudo tee
result = subprocess.run(
["sudo", "tee", str(wpa_path)],
input=config_text, capture_output=True, text=True,
)
if result.returncode != 0:
raise RuntimeError(f"Failed to write wpa_supplicant file: {result.stderr.strip()}")
# Set permissions 600
chmod_result = subprocess.run(
["sudo", "chmod", "600", str(wpa_path)],
capture_output=True, text=True,
)
if chmod_result.returncode != 0:
console.print(f"[yellow]Warning: chmod 600 failed: {chmod_result.stderr.strip()}[/yellow]")
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Config accessors # Config accessors
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -369,56 +514,191 @@ def run_interactive(device: str, mountpoint: str) -> None:
console.print("[green]OK[/green]") console.print("[green]OK[/green]")
try: try:
config = read_netplan(mountpoint) config_type = detect_wifi_config_type(mountpoint)
dirty = False
while True: if config_type == "wpa_supplicant":
print_networks(config) # WPA Supplicant mode: work with aps dict and country string directly
country = get_country(config) aps, country = read_wpa_supplicant(mountpoint)
console.print(f" Country: [yellow]{country}[/yellow]") dirty = False
console.print()
console.print(" [bold]1.[/bold] Add / update network")
console.print(" [bold]2.[/bold] Remove network")
console.print(" [bold]3.[/bold] Update password")
console.print(" [bold]4.[/bold] Change country code")
console.print(" [bold]5.[/bold] Done (write & exit)")
console.print(" [bold]0.[/bold] Quit without saving")
console.print()
choice = Prompt.ask(" [cyan]Select[/cyan]").strip() while True:
# Print networks table for wpa_supplicant
console.print()
if not aps:
console.print(" [dim]No WiFi networks configured.[/dim]")
else:
table = Table(title=f"Configured Networks (country: {country})",
show_lines=True, border_style="cyan")
table.add_column("#", justify="right", min_width=3, style="dim")
table.add_column("SSID", min_width=24, style="bold")
table.add_column("Password", min_width=20)
if choice == "1": for idx, (ssid, details) in enumerate(aps.items(), start=1):
if action_add_network(config): pwd = details.get("password", "") if isinstance(details, dict) else ""
dirty = True table.add_row(str(idx), ssid, pwd or "[dim](none)[/dim]")
elif choice == "2":
if action_remove_network(config): console.print(table)
dirty = True console.print()
elif choice == "3": console.print(f" Country: [yellow]{country}[/yellow]")
if action_update_password(config): console.print()
dirty = True console.print(" [bold]1.[/bold] Add / update network")
elif choice == "4": console.print(" [bold]2.[/bold] Remove network")
if action_change_country(config): console.print(" [bold]3.[/bold] Update password")
dirty = True console.print(" [bold]4.[/bold] Change country code")
elif choice == "5": console.print(" [bold]5.[/bold] Done (write & exit)")
break console.print(" [bold]0.[/bold] Quit without saving")
elif choice == "0": console.print()
console.print("\n [yellow]Discarding changes.[/yellow]")
dirty = False choice = Prompt.ask(" [cyan]Select[/cyan]").strip()
return
if choice == "1":
# Add network
console.print()
ssid = Prompt.ask(" [cyan]SSID[/cyan]").strip()
if ssid:
password = Prompt.ask(" [cyan]Password[/cyan] (leave blank for open network)").strip()
aps[ssid] = {"password": password} if password else {}
verb = "Updated" if ssid in aps else "Added"
console.print(f" [green]Added[/green] [bold]{ssid}[/bold]")
dirty = True
else:
console.print(" [yellow]Cancelled — empty SSID.[/yellow]")
elif choice == "2":
# Remove network
if not aps:
console.print(" [yellow]No networks configured.[/yellow]")
else:
ssid_list = list(aps.keys())
console.print()
for idx, ssid in enumerate(ssid_list, start=1):
console.print(f" [bold]{idx}.[/bold] {ssid}")
console.print()
choice_rem = Prompt.ask(" Remove # (or [dim]Enter[/dim] to cancel)").strip()
if choice_rem:
try:
n = int(choice_rem)
if 1 <= n <= len(ssid_list):
target = ssid_list[n - 1]
del aps[target]
console.print(f" [green]Removed[/green] [bold]{target}[/bold]")
dirty = True
else:
console.print(" [yellow]Invalid selection.[/yellow]")
except ValueError:
console.print(" [yellow]Invalid selection.[/yellow]")
elif choice == "3":
# Update password
if not aps:
console.print(" [yellow]No networks configured.[/yellow]")
else:
ssid_list = list(aps.keys())
console.print()
for idx, ssid in enumerate(ssid_list, start=1):
console.print(f" [bold]{idx}.[/bold] {ssid}")
console.print()
choice_upd = Prompt.ask(" Update password for # (or [dim]Enter[/dim] to cancel)").strip()
if choice_upd:
try:
n = int(choice_upd)
if 1 <= n <= len(ssid_list):
target = ssid_list[n - 1]
password = Prompt.ask(f" New password for [bold]{target}[/bold] "
"(leave blank for open network)").strip()
aps[target] = {"password": password} if password else {}
console.print(f" [green]Password updated[/green] for [bold]{target}[/bold]")
dirty = True
else:
console.print(" [yellow]Invalid selection.[/yellow]")
except ValueError:
console.print(" [yellow]Invalid selection.[/yellow]")
elif choice == "4":
# Change country
console.print()
new_country = Prompt.ask(
f" Country code (current: [yellow]{country}[/yellow])"
).strip().upper()
if new_country:
if len(new_country) != 2 or not new_country.isalpha():
console.print(" [yellow]Invalid country code — must be 2 letters (e.g. US, GB, DE).[/yellow]")
else:
country = new_country
console.print(f" [green]Country set to[/green] [bold]{new_country}[/bold]")
dirty = True
else:
console.print(" [yellow]Cancelled.[/yellow]")
elif choice == "5":
break
elif choice == "0":
console.print("\n [yellow]Discarding changes.[/yellow]")
dirty = False
return
else:
console.print(" [yellow]Invalid option.[/yellow]")
if dirty:
console.print()
console.print(" Writing config ... ", end="")
try:
write_wpa_supplicant(mountpoint, aps, country)
console.print("[green]OK[/green]")
console.print(f" [green]Saved[/green] → [dim]{WPA_SUPPLICANT_PATH}[/dim] (chmod 600)")
except RuntimeError as exc:
console.print(f"[red]FAILED[/red]\n {exc}")
else: else:
console.print(" [yellow]Invalid option.[/yellow]") console.print(" [dim]No changes to write.[/dim]")
if dirty:
console.print()
console.print(" Writing config ... ", end="")
try:
write_netplan(mountpoint, config)
console.print("[green]OK[/green]")
console.print(f" [green]Saved[/green] → [dim]{NETPLAN_PATH}[/dim] (chmod 600)")
except RuntimeError as exc:
console.print(f"[red]FAILED[/red]\n {exc}")
else: else:
console.print(" [dim]No changes to write.[/dim]") # Netplan mode: use existing config dict-based approach
config = read_netplan(mountpoint)
dirty = False
while True:
print_networks(config)
country = get_country(config)
console.print(f" Country: [yellow]{country}[/yellow]")
console.print()
console.print(" [bold]1.[/bold] Add / update network")
console.print(" [bold]2.[/bold] Remove network")
console.print(" [bold]3.[/bold] Update password")
console.print(" [bold]4.[/bold] Change country code")
console.print(" [bold]5.[/bold] Done (write & exit)")
console.print(" [bold]0.[/bold] Quit without saving")
console.print()
choice = Prompt.ask(" [cyan]Select[/cyan]").strip()
if choice == "1":
if action_add_network(config):
dirty = True
elif choice == "2":
if action_remove_network(config):
dirty = True
elif choice == "3":
if action_update_password(config):
dirty = True
elif choice == "4":
if action_change_country(config):
dirty = True
elif choice == "5":
break
elif choice == "0":
console.print("\n [yellow]Discarding changes.[/yellow]")
dirty = False
return
else:
console.print(" [yellow]Invalid option.[/yellow]")
if dirty:
console.print()
console.print(" Writing config ... ", end="")
try:
write_netplan(mountpoint, config)
console.print("[green]OK[/green]")
console.print(f" [green]Saved[/green] → [dim]{NETPLAN_PATH}[/dim] (chmod 600)")
except RuntimeError as exc:
console.print(f"[red]FAILED[/red]\n {exc}")
else:
console.print(" [dim]No changes to write.[/dim]")
finally: finally:
console.print() console.print()
@@ -433,10 +713,32 @@ def run_list(device: str, mountpoint: str) -> None:
return return
try: try:
config = read_netplan(mountpoint) config_type = detect_wifi_config_type(mountpoint)
print_networks(config)
country = get_country(config) if config_type == "wpa_supplicant":
console.print(f" Regulatory domain: [yellow]{country}[/yellow]") aps, country = read_wpa_supplicant(mountpoint)
console.print()
if not aps:
console.print(" [dim]No WiFi networks configured.[/dim]")
else:
table = Table(title=f"Configured Networks (country: {country})",
show_lines=True, border_style="cyan")
table.add_column("#", justify="right", min_width=3, style="dim")
table.add_column("SSID", min_width=24, style="bold")
table.add_column("Password", min_width=20)
for idx, (ssid, details) in enumerate(aps.items(), start=1):
pwd = details.get("password", "") if isinstance(details, dict) else ""
table.add_row(str(idx), ssid, pwd or "[dim](none)[/dim]")
console.print(table)
console.print()
console.print(f" Regulatory domain: [yellow]{country}[/yellow]")
else:
config = read_netplan(mountpoint)
print_networks(config)
country = get_country(config)
console.print(f" Regulatory domain: [yellow]{country}[/yellow]")
finally: finally:
unmount_sd(mountpoint) unmount_sd(mountpoint)
@@ -447,16 +749,26 @@ def run_add(device: str, mountpoint: str, ssid: str, password: str) -> None:
return return
try: try:
config = read_netplan(mountpoint) config_type = detect_wifi_config_type(mountpoint)
aps = get_access_points(config)
existed = ssid in aps
aps[ssid] = {"password": password} if password else {}
set_access_points(config, aps)
write_netplan(mountpoint, config) if config_type == "wpa_supplicant":
aps, country = read_wpa_supplicant(mountpoint)
existed = ssid in aps
aps[ssid] = {"password": password} if password else {}
write_wpa_supplicant(mountpoint, aps, country)
verb = "Updated" if existed else "Added"
console.print(f" [green]{verb}[/green] [bold]{ssid}[/bold] (chmod 600 applied)")
else:
config = read_netplan(mountpoint)
aps = get_access_points(config)
existed = ssid in aps
aps[ssid] = {"password": password} if password else {}
set_access_points(config, aps)
verb = "Updated" if existed else "Added" write_netplan(mountpoint, config)
console.print(f" [green]{verb}[/green] [bold]{ssid}[/bold] (chmod 600 applied)")
verb = "Updated" if existed else "Added"
console.print(f" [green]{verb}[/green] [bold]{ssid}[/bold] (chmod 600 applied)")
except RuntimeError as exc: except RuntimeError as exc:
console.print(f" [red]Error:[/red] {exc}") console.print(f" [red]Error:[/red] {exc}")
finally: finally: