feat: radio1 5GHz/6GHz rogue AP via UCI with hop pause
This commit is contained in:
@@ -1717,7 +1717,7 @@ def _apply_open_radio(openap):
|
||||
daemon's iface-level channel write does not affect the actual radio, so
|
||||
apply channel/country here and reload wifi when they change."""
|
||||
changed = False
|
||||
radio = _uci_wifi_iface('radio0')
|
||||
radio = _uci_wifi_iface('radio0') or {}
|
||||
for key in ('channel', 'country'):
|
||||
value = openap.get(key)
|
||||
if value is None:
|
||||
@@ -1739,11 +1739,100 @@ def _open_channel(value):
|
||||
return 1
|
||||
|
||||
|
||||
def _read_hop():
|
||||
rc, out, err = device_run(['uci', 'get', 'pineapd.wlan1mon.hop'])
|
||||
return out.strip()
|
||||
|
||||
|
||||
def _pause_hop():
|
||||
if _read_hop() != '0':
|
||||
device_run(['uci', 'set', 'pineapd.wlan1mon.hop=0'])
|
||||
device_run(['uci', 'commit', 'pineapd'])
|
||||
device_run(['/etc/init.d/pineapd', 'reload'])
|
||||
|
||||
|
||||
def _resume_hop():
|
||||
if _read_hop() == '0':
|
||||
device_run(['uci', 'set', 'pineapd.wlan1mon.hop=1'])
|
||||
device_run(['uci', 'commit', 'pineapd'])
|
||||
device_run(['/etc/init.d/pineapd', 'reload'])
|
||||
|
||||
|
||||
def _remove_radio1_ap():
|
||||
device_run(['uci', 'delete', 'wireless.wlan1open'])
|
||||
device_run(['uci', 'delete', 'wireless.wlan1wpa'])
|
||||
device_run(['uci', 'set', 'wireless.radio1.channel=auto'])
|
||||
device_run(['uci', 'set', 'wireless.radio1.band=5g'])
|
||||
device_run(['uci', 'commit', 'wireless'])
|
||||
_resume_hop()
|
||||
|
||||
|
||||
def _apply_radio1_ap(openap, wpa):
|
||||
band = None
|
||||
if openap is not None:
|
||||
band = channel_band(openap.get('channel'))
|
||||
iface = 'wlan1open'
|
||||
if wpa is not None:
|
||||
band = channel_band(wpa.get('channel'))
|
||||
iface = 'wlan1wpa'
|
||||
if band not in (BAND_5G, BAND_6G):
|
||||
raise ValueError('radio1 AP requires a 5GHz or 6GHz channel')
|
||||
if band == BAND_6G and wpa is not None:
|
||||
if (wpa.get('enctype') or 'psk2') not in ('sae', 'owe'):
|
||||
raise ValueError('6GHz requires WPA3 (sae or owe)')
|
||||
device_run(['uci', 'delete', 'wireless.wlan1open'])
|
||||
device_run(['uci', 'delete', 'wireless.wlan1wpa'])
|
||||
device_run(['uci', 'set', 'wireless.radio1.band=%s' % ('6g' if band == BAND_6G else '5g')])
|
||||
ch = int(wpa.get('channel') if wpa is not None else openap.get('channel'))
|
||||
device_run(['uci', 'set', 'wireless.radio1.channel=%d' % ch])
|
||||
device_run(['uci', 'set', 'wireless.radio1.htmode=%s' % band_htmode(band)])
|
||||
country = (wpa or openap or {}).get('country')
|
||||
if country:
|
||||
device_run(['uci', 'set', 'wireless.radio1.country=%s' % country])
|
||||
cfg = wpa if wpa is not None else openap
|
||||
device_run(['uci', 'set', 'wireless.%s=wifi-iface' % iface])
|
||||
device_run(['uci', 'set', 'wireless.%s.device=radio1' % iface])
|
||||
device_run(['uci', 'set', 'wireless.%s.mode=ap' % iface])
|
||||
device_run(['uci', 'set', 'wireless.%s.ssid=%s' % (iface, cfg.get('ssid') or '')])
|
||||
device_run(['uci', 'set', 'wireless.%s.hidden=%d' % (iface, 1 if cfg.get('hidden') else 0)])
|
||||
device_run(['uci', 'set', 'wireless.%s.channel=%d' % (iface, int(cfg.get('channel')))])
|
||||
if wpa is not None:
|
||||
device_run(['uci', 'set', 'wireless.%s.encryption=%s' % (iface, wpa.get('enctype') or 'psk2')])
|
||||
device_run(['uci', 'set', 'wireless.%s.key=%s' % (iface, wpa.get('passphrase') or '')])
|
||||
else:
|
||||
device_run(['uci', 'set', 'wireless.%s.encryption=none' % iface])
|
||||
bssid = (openap or {}).get('bssid') or ''
|
||||
if bssid:
|
||||
device_run(['uci', 'set', 'wireless.%s.macaddr=%s' % (iface, bssid)])
|
||||
device_run(['uci', 'commit', 'wireless'])
|
||||
_pause_hop()
|
||||
device_run(['wifi', 'reload'])
|
||||
|
||||
|
||||
def h_pineap_wifi_set_ap(ctx):
|
||||
body = ctx.body or {}
|
||||
configs = []
|
||||
wpa = body.get('wpa') or {}
|
||||
openap = body.get('open') or {}
|
||||
wpa_band = channel_band(wpa.get('channel')) if wpa.get('channel') is not None else None
|
||||
open_band = channel_band(openap.get('channel')) if openap.get('channel') is not None else None
|
||||
use_radio1 = wpa_band in (BAND_5G, BAND_6G) or open_band in (BAND_5G, BAND_6G)
|
||||
if use_radio1:
|
||||
wpa_active = bool(wpa.get('enabled', True)) and wpa.get('channel') is not None
|
||||
open_active = bool(openap.get('enabled', True)) and openap.get('channel') is not None
|
||||
if not (wpa_active or open_active):
|
||||
_remove_radio1_ap()
|
||||
device_run(['wifi', 'reload'])
|
||||
return 200, {'ok': True}
|
||||
try:
|
||||
_apply_radio1_ap(openap if open_active else None,
|
||||
wpa if wpa_active else None)
|
||||
except ValueError as exc:
|
||||
return 400, {'error': str(exc)}
|
||||
return 200, {'ok': True}
|
||||
if _uci_wifi_iface('wlan1open') or _uci_wifi_iface('wlan1wpa'):
|
||||
_remove_radio1_ap()
|
||||
device_run(['wifi', 'reload'])
|
||||
configs = []
|
||||
if wpa.get('ssid') or wpa.get('enabled') is not None:
|
||||
configs.append({
|
||||
'interface': 'wlan0wpa',
|
||||
|
||||
@@ -159,6 +159,97 @@ class GetApRadio1AbsentTest(unittest.TestCase):
|
||||
self.assertEqual(payload['wpa']['channel'], 1)
|
||||
|
||||
|
||||
class SetApRadio1Test(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.uci = {}
|
||||
self.runs = []
|
||||
|
||||
def fake_uci_get(section):
|
||||
return self.uci.get(section)
|
||||
|
||||
def fake_run(args, timeout=20, input_data=None):
|
||||
self.runs.append((list(args), input_data))
|
||||
a = list(args)
|
||||
if a[:2] == ['uci', 'set']:
|
||||
k, _, v = a[2].partition('=')
|
||||
self.uci[k] = v
|
||||
if a[:2] == ['uci', 'get']:
|
||||
return (0, self.uci.get(a[2], '') + '\n', '')
|
||||
return (0, '', '')
|
||||
|
||||
server._uci_wifi_iface = fake_uci_get
|
||||
server._uci_section = fake_uci_get
|
||||
server.device_run = fake_run
|
||||
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
|
||||
200, {'success': True})
|
||||
|
||||
def test_5g_open_writes_radio1_sections(self):
|
||||
server.h_pineap_wifi_set_ap(ctx({'open': {
|
||||
'ssid': 'CorpGuest', 'hidden': False, 'enabled': True,
|
||||
'channel': 36, 'country': 'US'}}))
|
||||
self.assertEqual(self.uci['wireless.radio1.band'], '5g')
|
||||
self.assertEqual(self.uci['wireless.radio1.channel'], '36')
|
||||
self.assertEqual(self.uci['wireless.radio1.htmode'], 'VHT80')
|
||||
self.assertEqual(self.uci['wireless.wlan1open'], 'wifi-iface')
|
||||
self.assertEqual(self.uci['wireless.wlan1open.device'], 'radio1')
|
||||
self.assertEqual(self.uci['wireless.wlan1open.ssid'], 'CorpGuest')
|
||||
self.assertEqual(self.uci['wireless.wlan1open.encryption'], 'none')
|
||||
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '0')
|
||||
self.assertIn(['wifi', 'reload'], [r[0] for r in self.runs])
|
||||
self.assertIn(['/etc/init.d/pineapd', 'reload'], [r[0] for r in self.runs])
|
||||
|
||||
def test_6g_wpa_sae(self):
|
||||
server.h_pineap_wifi_set_ap(ctx({'wpa': {
|
||||
'ssid': 'Corp', 'passphrase': 'secret123', 'enctype': 'sae',
|
||||
'hidden': False, 'enabled': True, 'channel': 181}}))
|
||||
self.assertEqual(self.uci['wireless.radio1.band'], '6g')
|
||||
self.assertEqual(self.uci['wireless.radio1.htmode'], 'HE80')
|
||||
self.assertEqual(self.uci['wireless.wlan1wpa.encryption'], 'sae')
|
||||
|
||||
def test_6g_rejects_psk2(self):
|
||||
status, payload = server.h_pineap_wifi_set_ap(ctx({'wpa': {
|
||||
'ssid': 'Corp', 'passphrase': 'secret123', 'enctype': 'psk2',
|
||||
'hidden': False, 'enabled': True, 'channel': 181}}))
|
||||
self.assertEqual(status, 400)
|
||||
|
||||
def test_6g_disable_removes_radio1(self):
|
||||
self.uci['pineapd.wlan1mon.hop'] = '0'
|
||||
self.uci['wlan1wpa'] = {'device': 'radio1'}
|
||||
status, payload = server.h_pineap_wifi_set_ap(ctx({'wpa': {
|
||||
'ssid': 'Corp', 'passphrase': 'secret123', 'enctype': 'sae',
|
||||
'hidden': False, 'enabled': False, 'channel': 181}}))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(self.uci['wireless.radio1.channel'], 'auto')
|
||||
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')
|
||||
|
||||
def test_2g_still_uses_daemon_path(self):
|
||||
calls = []
|
||||
|
||||
def fake_sock(method, path, body=None, timeout=10):
|
||||
if method == 'GET':
|
||||
return 200, {'loghandshake': True}
|
||||
calls.append((method, path, body))
|
||||
return 200, {'success': True}
|
||||
|
||||
server.daemon_sock_call = fake_sock
|
||||
server.h_pineap_wifi_set_ap(ctx({'open': {
|
||||
'ssid': 'pager-open', 'hidden': False, 'enabled': True,
|
||||
'channel': 11, 'country': 'US'}}))
|
||||
put = [c for c in calls if c[0] == 'PUT']
|
||||
self.assertEqual(len(put), 1)
|
||||
self.assertEqual(put[0][1], '/api/settings/wifi/set_ap')
|
||||
self.assertEqual(put[0][2]['configs'][0]['interface'], 'wlan0open')
|
||||
|
||||
def test_2g_removes_existing_radio1(self):
|
||||
self.uci['pineapd.wlan1mon.hop'] = '0'
|
||||
self.uci['wlan1open'] = {'device': 'radio1'}
|
||||
server.h_pineap_wifi_set_ap(ctx({'open': {
|
||||
'ssid': 'pager-open', 'hidden': False, 'enabled': True,
|
||||
'channel': 11, 'country': 'US'}}))
|
||||
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')
|
||||
self.assertEqual(self.uci.get('wireless.radio1.channel'), 'auto')
|
||||
|
||||
|
||||
class GetApRadioChannelFallbackTest(unittest.TestCase):
|
||||
"""Regression: iface without a channel option inherits the radio channel."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user