"""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