182 lines
6.9 KiB
Python
182 lines
6.9 KiB
Python
"""Mark VIII RF role manager: radio1/phy1 is shared between an uplink STA
|
|
(``wlan1up``) and attack work, so the roles are made mutually exclusive.
|
|
Uplink pauses channel hopping; attack/idle resumes it."""
|
|
import time
|
|
|
|
ROLE_KEY = 'mk8.rfplan.role'
|
|
IFACE = 'wlan1up'
|
|
|
|
# wifi reload returns while wpa_supplicant is still scanning/authenticating;
|
|
# poll instead of checking once or every real uplink would false-fail.
|
|
ASSOC_ATTEMPTS = 5
|
|
# Security modes tried in order for PSK uplinks. sae-mixed covers
|
|
# WPA2/WPA3 transition APs; plain SAE covers WPA3-only (PMF required);
|
|
# psk2 covers legacy WPA2-PSK. ieee80211w matches each mode's PMF need.
|
|
PSK_MODE_CHAIN = (('sae-mixed', '1'), ('sae', '2'), ('psk2', '0'))
|
|
ASSOC_WAIT_SECONDS = 2
|
|
|
|
|
|
def current_role():
|
|
from server import _uci_values
|
|
cfg = _uci_values('wireless.%s' % IFACE) or {}
|
|
if cfg.get('disabled') != '1' and cfg.get('mode') == 'sta':
|
|
return 'uplink'
|
|
return 'idle'
|
|
|
|
|
|
def _sta_netdev():
|
|
"""Actual netdev carrying the radio1 STA. OpenWrt ignores a requested
|
|
ifname for mac80211 STA ifaces (comes up as phy1-sta0), so resolve by
|
|
phy membership + managed type instead of by name."""
|
|
from server import device_run
|
|
rc, out, err = device_run(['iw', 'dev'], timeout=10)
|
|
if rc != 0:
|
|
return None
|
|
current = None
|
|
managed = []
|
|
for line in (out or '').splitlines():
|
|
line = line.strip()
|
|
if line.startswith('Interface '):
|
|
current = line.split()[1]
|
|
elif line.startswith('type managed') and current:
|
|
if not current.startswith('wlan0'):
|
|
managed.append(current)
|
|
current = None
|
|
for name in managed:
|
|
rc2, o2, _ = device_run(
|
|
['readlink', '/sys/class/net/%s/phy80211' % name], timeout=10)
|
|
if rc2 == 0 and 'phy1' in (o2 or ''):
|
|
return name
|
|
return None
|
|
|
|
|
|
def associated():
|
|
"""BSSID of the uplink AP when the radio1 STA is associated, else None."""
|
|
from server import device_run
|
|
dev = _sta_netdev()
|
|
if not dev:
|
|
return None
|
|
rc, out, err = device_run(['iw', 'dev', dev, 'link'], timeout=10)
|
|
if rc != 0 or 'Connected' not in (out or ''):
|
|
return None
|
|
for line in (out or '').splitlines():
|
|
line = line.strip()
|
|
if line.startswith('Connected to '):
|
|
parts = line.split()
|
|
if len(parts) >= 3:
|
|
return parts[2]
|
|
return None
|
|
|
|
|
|
def hop_paused():
|
|
from server import _read_hop
|
|
return _read_hop() == '0'
|
|
|
|
|
|
def _ensure_cli_network():
|
|
"""Make network 'cli' usable for the STA. Returns True when a network
|
|
change was staged and still needs ``uci commit network``. Stock firmware
|
|
ships 'cli' present but disabled; create a minimal DHCP interface when it
|
|
is missing entirely so netifd can bring wlan1up up either way."""
|
|
from server import device_run
|
|
rc, _, _ = device_run(['uci', '-q', 'get', 'network.cli'])
|
|
if rc != 0:
|
|
device_run(['uci', 'set', 'network.cli=interface'])
|
|
device_run(['uci', 'set', 'network.cli.proto=dhcp'])
|
|
return True
|
|
rc, out, _ = device_run(['uci', '-q', 'get', 'network.cli.disabled'])
|
|
if rc == 0 and out.strip() == '1':
|
|
device_run(['uci', 'set', 'network.cli.disabled=0'])
|
|
return True
|
|
return False
|
|
|
|
|
|
def set_role(role, ssid=None, psk=None):
|
|
from server import device_run, _pause_hop, _resume_hop
|
|
if role not in ('uplink', 'attack', 'idle'):
|
|
return {'ok': False, 'error': 'role must be uplink, attack or idle'}
|
|
if role == 'uplink':
|
|
if not ssid:
|
|
return {'ok': False, 'error': 'ssid required'}
|
|
base_cmds = [
|
|
['uci', 'set', 'wireless.wlan1up=wifi-iface'],
|
|
['uci', 'set', 'wireless.wlan1up.device=radio1'],
|
|
['uci', 'set', 'wireless.wlan1up.mode=sta'],
|
|
['uci', 'set', 'wireless.wlan1up.network=cli'],
|
|
['uci', 'set', 'wireless.wlan1up.ssid=%s' % ssid],
|
|
['uci', 'set', 'wireless.wlan1up.disabled=0'],
|
|
]
|
|
if psk:
|
|
base_cmds.append(['uci', 'set',
|
|
'wireless.wlan1up.key=%s' % psk])
|
|
for c in base_cmds:
|
|
device_run(c)
|
|
if _ensure_cli_network():
|
|
# netifd consumes committed config only; staging without commit
|
|
# would leave the STA with no L3 attachment.
|
|
device_run(['uci', 'commit', 'network'])
|
|
_pause_hop()
|
|
assoc = None
|
|
used_mode = None
|
|
modes = PSK_MODE_CHAIN if psk else [('none', None)]
|
|
for enc, pmf in modes:
|
|
device_run(['uci', 'set', 'wireless.wlan1up.encryption=%s' % enc])
|
|
if pmf is not None:
|
|
device_run(['uci', 'set',
|
|
'wireless.wlan1up.ieee80211w=%s' % pmf])
|
|
device_run(['uci', 'commit', 'wireless'])
|
|
device_run(['wifi', 'reload'], timeout=60)
|
|
for _ in range(ASSOC_ATTEMPTS):
|
|
time.sleep(ASSOC_WAIT_SECONDS)
|
|
assoc = associated()
|
|
if assoc:
|
|
break
|
|
if assoc:
|
|
used_mode = enc
|
|
break
|
|
if not assoc or not used_mode:
|
|
disable_uplink()
|
|
# UCI alone does not converge runtime: without a reload wlan1up
|
|
# keeps scanning/authenticating and pins phy1 until some unrelated
|
|
# future reload, while current_role() already reports idle.
|
|
# Converge now like the idle branch, then reapply the hop policy.
|
|
device_run(['wifi', 'reload'], timeout=60)
|
|
_resume_hop()
|
|
return {'ok': False,
|
|
'error': 'association failed; reverted',
|
|
'tried_modes': [m for m, _ in modes]}
|
|
return {'ok': True, 'role': 'uplink', 'assoc': assoc,
|
|
'mode': used_mode}
|
|
# attack/idle: tear down the STA so radio1 is free again.
|
|
disable_uplink()
|
|
_resume_hop()
|
|
if role == 'attack':
|
|
# The deploy path performs its own wifi reload right after; teardown
|
|
# converges there without a second reload churn on this phy.
|
|
try:
|
|
import mk8_events
|
|
mk8_events.log_event(
|
|
'rfplan', msg='rfplan role attack applied; STA teardown '
|
|
'applies at next wifi reload')
|
|
except Exception:
|
|
pass
|
|
else:
|
|
# idle has no guaranteed follow-up reload anywhere else, so converge
|
|
# now while the gated watchdog is still armed.
|
|
device_run(['wifi', 'reload'], timeout=60)
|
|
return {'ok': True, 'role': role}
|
|
|
|
|
|
def disable_uplink():
|
|
from server import device_run
|
|
device_run(['uci', 'set', 'wireless.wlan1up.disabled=1'])
|
|
device_run(['uci', 'commit', 'wireless'])
|
|
|
|
|
|
def ensure_attack():
|
|
"""Exclusivity hook for radio1 attack-AP enable paths: switch the
|
|
uplink off first so one phy never carries STA + AP at once."""
|
|
if current_role() == 'uplink':
|
|
return set_role('attack')
|
|
return None
|