fix: gate channel_band 6GHz to 181-233, reject open-6GHz, self-heal load-time 6GHz hints, harden hop/BSSID handling
This commit is contained in:
@@ -1607,6 +1607,9 @@ DFS_CHANNELS = frozenset([52, 56, 60, 64, 100, 104, 108, 112, 116, 120,
|
|||||||
124, 128, 132, 136, 140, 144])
|
124, 128, 132, 136, 140, 144])
|
||||||
|
|
||||||
|
|
||||||
|
_last_reconcile = 0.0
|
||||||
|
|
||||||
|
|
||||||
def channel_band(ch):
|
def channel_band(ch):
|
||||||
if ch is None:
|
if ch is None:
|
||||||
return None
|
return None
|
||||||
@@ -1618,7 +1621,7 @@ def channel_band(ch):
|
|||||||
return BAND_2G
|
return BAND_2G
|
||||||
if 36 <= ch <= 177:
|
if 36 <= ch <= 177:
|
||||||
return BAND_5G
|
return BAND_5G
|
||||||
if 1 <= ch <= 233 and (ch - 1) % 4 == 0:
|
if 177 < ch <= 233 and (ch - 1) % 4 == 0:
|
||||||
return BAND_6G
|
return BAND_6G
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@@ -1646,6 +1649,8 @@ def _uci_section(section):
|
|||||||
|
|
||||||
|
|
||||||
def h_pineap_wifi_get_ap(ctx):
|
def h_pineap_wifi_get_ap(ctx):
|
||||||
|
global _last_reconcile
|
||||||
|
|
||||||
def _iface_state(open_name, wpa_name, radio_name):
|
def _iface_state(open_name, wpa_name, radio_name):
|
||||||
open_cfg = _uci_wifi_iface(open_name)
|
open_cfg = _uci_wifi_iface(open_name)
|
||||||
wpa_cfg = _uci_wifi_iface(wpa_name)
|
wpa_cfg = _uci_wifi_iface(wpa_name)
|
||||||
@@ -1685,7 +1690,9 @@ def h_pineap_wifi_get_ap(ctx):
|
|||||||
for name in ('wlan1open', 'wlan1wpa'):
|
for name in ('wlan1open', 'wlan1wpa'):
|
||||||
cfg = _uci_wifi_iface(name)
|
cfg = _uci_wifi_iface(name)
|
||||||
if cfg and cfg.get('disabled') != '1' and not os.path.exists('/sys/class/net/%s' % name):
|
if cfg and cfg.get('disabled') != '1' and not os.path.exists('/sys/class/net/%s' % name):
|
||||||
device_run(['wifi', 'reload'])
|
if time.time() - _last_reconcile > 30:
|
||||||
|
_last_reconcile = time.time()
|
||||||
|
device_run(['wifi', 'reload'])
|
||||||
break
|
break
|
||||||
return 200, {
|
return 200, {
|
||||||
'open': {
|
'open': {
|
||||||
@@ -1746,6 +1753,8 @@ def _open_channel(value):
|
|||||||
|
|
||||||
def _read_hop():
|
def _read_hop():
|
||||||
rc, out, err = device_run(['uci', 'get', 'pineapd.wlan1mon.hop'])
|
rc, out, err = device_run(['uci', 'get', 'pineapd.wlan1mon.hop'])
|
||||||
|
if rc != 0:
|
||||||
|
return None
|
||||||
return out.strip()
|
return out.strip()
|
||||||
|
|
||||||
|
|
||||||
@@ -1782,6 +1791,8 @@ def _apply_radio1_ap(openap, wpa):
|
|||||||
iface = 'wlan1wpa'
|
iface = 'wlan1wpa'
|
||||||
if band not in (BAND_5G, BAND_6G):
|
if band not in (BAND_5G, BAND_6G):
|
||||||
raise ValueError('radio1 AP requires a 5GHz or 6GHz channel')
|
raise ValueError('radio1 AP requires a 5GHz or 6GHz channel')
|
||||||
|
if band == BAND_6G and wpa is None:
|
||||||
|
raise ValueError('6GHz open APs are not supported (6GHz requires WPA3/OWE)')
|
||||||
if band == BAND_6G and wpa is not None:
|
if band == BAND_6G and wpa is not None:
|
||||||
if (wpa.get('enctype') or 'psk2') not in ('sae', 'owe'):
|
if (wpa.get('enctype') or 'psk2') not in ('sae', 'owe'):
|
||||||
raise ValueError('6GHz requires WPA3 (sae or owe)')
|
raise ValueError('6GHz requires WPA3 (sae or owe)')
|
||||||
@@ -1810,7 +1821,9 @@ def _apply_radio1_ap(openap, wpa):
|
|||||||
device_run(['uci', 'set', 'wireless.%s.encryption=none' % iface])
|
device_run(['uci', 'set', 'wireless.%s.encryption=none' % iface])
|
||||||
bssid = (openap or {}).get('bssid') or ''
|
bssid = (openap or {}).get('bssid') or ''
|
||||||
if bssid:
|
if bssid:
|
||||||
device_run(['uci', 'set', 'wireless.%s.macaddr=%s' % (iface, bssid)])
|
if not re.match(r'^[0-9A-F]{2}(?::[0-9A-F]{2}){5}$', bssid.upper()):
|
||||||
|
raise ValueError('invalid BSSID format')
|
||||||
|
device_run(['uci', 'set', 'wireless.%s.macaddr=%s' % (iface, bssid.upper())])
|
||||||
device_run(['uci', 'commit', 'wireless'])
|
device_run(['uci', 'commit', 'wireless'])
|
||||||
_pause_hop()
|
_pause_hop()
|
||||||
device_run(['wifi', 'reload'])
|
device_run(['wifi', 'reload'])
|
||||||
|
|||||||
@@ -493,10 +493,11 @@ views.pineap_open = (root) => {
|
|||||||
const chSel = h('select', { id: 'oa-channel' });
|
const chSel = h('select', { id: 'oa-channel' });
|
||||||
chanSelect(chSel, null);
|
chanSelect(chSel, null);
|
||||||
const bandHint = h('div', { class: 'muted', style: 'font-size:12px;margin-top:4px' });
|
const bandHint = h('div', { class: 'muted', style: 'font-size:12px;margin-top:4px' });
|
||||||
chSel.addEventListener('change', () => {
|
function applyOaHint() {
|
||||||
const b = bandOfChannel(chSel.value);
|
const b = bandOfChannel(chSel.value);
|
||||||
bandHint.textContent = b === '6' ? '6 GHz open APs require WPA3/OWE on real clients — most devices will not associate to an open 6 GHz network.' : '';
|
bandHint.textContent = b === '6' ? '6 GHz open APs require WPA3/OWE on real clients — most devices will not associate to an open 6 GHz network.' : '';
|
||||||
});
|
}
|
||||||
|
chSel.addEventListener('change', applyOaHint);
|
||||||
const coSel = h('select', { id: 'oa-country' });
|
const coSel = h('select', { id: 'oa-country' });
|
||||||
OPEN_COUNTRIES.forEach(([v, l]) => coSel.appendChild(h('option', { value: v, text: l })));
|
OPEN_COUNTRIES.forEach(([v, l]) => coSel.appendChild(h('option', { value: v, text: l })));
|
||||||
const hiddenCb = h('input', { type: 'checkbox', id: 'oa-hidden' });
|
const hiddenCb = h('input', { type: 'checkbox', id: 'oa-hidden' });
|
||||||
@@ -633,6 +634,7 @@ views.pineap_open = (root) => {
|
|||||||
chSel.value = String(open.channel);
|
chSel.value = String(open.channel);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
applyOaHint();
|
||||||
if (open.country) coSel.value = open.country;
|
if (open.country) coSel.value = open.country;
|
||||||
hiddenCb.checked = !!open.hidden;
|
hiddenCb.checked = !!open.hidden;
|
||||||
state.enabledLoaded = !!(a.open);
|
state.enabledLoaded = !!(a.open);
|
||||||
@@ -670,13 +672,13 @@ views.pineap_evilwpa = (root) => {
|
|||||||
const wpaChan = h('select', { id: 'ew-channel' });
|
const wpaChan = h('select', { id: 'ew-channel' });
|
||||||
chanSelect(wpaChan, null);
|
chanSelect(wpaChan, null);
|
||||||
const wpaHint = h('div', { class: 'muted', style: 'font-size:12px;margin-top:4px' });
|
const wpaHint = h('div', { class: 'muted', style: 'font-size:12px;margin-top:4px' });
|
||||||
wpaChan.addEventListener('change', () => {
|
function applyWpaHint() {
|
||||||
const b = bandOfChannel(wpaChan.value);
|
const six = bandOfChannel(wpaChan.value) === '6';
|
||||||
const six = b === '6';
|
|
||||||
Array.prototype.forEach.call(encSel.options, (o) => { o.disabled = six && o.value === 'psk2'; });
|
Array.prototype.forEach.call(encSel.options, (o) => { o.disabled = six && o.value === 'psk2'; });
|
||||||
if (six && encSel.value === 'psk2') encSel.value = 'sae';
|
if (six && encSel.value === 'psk2') encSel.value = 'sae';
|
||||||
wpaHint.textContent = six ? '6 GHz requires WPA3 (SAE or OWE).' : '';
|
wpaHint.textContent = six ? '6 GHz requires WPA3 (SAE or OWE).' : '';
|
||||||
});
|
}
|
||||||
|
wpaChan.addEventListener('change', applyWpaHint);
|
||||||
cfg.appendChild(h('label', {}, 'SSID', ssidIn));
|
cfg.appendChild(h('label', {}, 'SSID', ssidIn));
|
||||||
cfg.appendChild(h('label', {}, 'Passphrase', pskIn));
|
cfg.appendChild(h('label', {}, 'Passphrase', pskIn));
|
||||||
cfg.appendChild(h('label', {}, 'Encryption', encSel));
|
cfg.appendChild(h('label', {}, 'Encryption', encSel));
|
||||||
@@ -747,6 +749,7 @@ views.pineap_evilwpa = (root) => {
|
|||||||
wpaChan.value = String(w.channel);
|
wpaChan.value = String(w.channel);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
applyWpaHint();
|
||||||
}).catch(() => {});
|
}).catch(() => {});
|
||||||
PagerAPI.get('/api/pineap/get_config').then((r) => {
|
PagerAPI.get('/api/pineap/get_config').then((r) => {
|
||||||
const p = r.data || {};
|
const p = r.data || {};
|
||||||
|
|||||||
@@ -24,7 +24,7 @@ class ChannelBandTest(unittest.TestCase):
|
|||||||
self.assertEqual(server.channel_band(ch), server.BAND_6G)
|
self.assertEqual(server.channel_band(ch), server.BAND_6G)
|
||||||
|
|
||||||
def test_invalid(self):
|
def test_invalid(self):
|
||||||
for ch in (0, 15, 35, 178, 234, None, 'x'):
|
for ch in (0, 15, 17, 21, 33, 35, 178, 234, None, 'x'):
|
||||||
self.assertIsNone(server.channel_band(ch))
|
self.assertIsNone(server.channel_band(ch))
|
||||||
|
|
||||||
|
|
||||||
@@ -34,6 +34,12 @@ class ChannelBandsConsistencyTest(unittest.TestCase):
|
|||||||
for ch in channels:
|
for ch in channels:
|
||||||
self.assertEqual(server.channel_band(ch), band, '%s should be %s' % (ch, band))
|
self.assertEqual(server.channel_band(ch), band, '%s should be %s' % (ch, band))
|
||||||
|
|
||||||
|
def test_no_out_of_list_channels(self):
|
||||||
|
for ch in list(range(0, 235)):
|
||||||
|
band = server.channel_band(ch)
|
||||||
|
if band is not None:
|
||||||
|
self.assertIn(ch, server.CHANNEL_BANDS[band], '%s should be listed for %s' % (ch, band))
|
||||||
|
|
||||||
def test_boundaries(self):
|
def test_boundaries(self):
|
||||||
self.assertEqual(server.CHANNEL_BANDS[server.BAND_2G][-1], 14)
|
self.assertEqual(server.CHANNEL_BANDS[server.BAND_2G][-1], 14)
|
||||||
self.assertEqual(server.CHANNEL_BANDS[server.BAND_5G][-1], 177)
|
self.assertEqual(server.CHANNEL_BANDS[server.BAND_5G][-1], 177)
|
||||||
@@ -213,6 +219,13 @@ class SetApRadio1Test(unittest.TestCase):
|
|||||||
'hidden': False, 'enabled': True, 'channel': 181}}))
|
'hidden': False, 'enabled': True, 'channel': 181}}))
|
||||||
self.assertEqual(status, 400)
|
self.assertEqual(status, 400)
|
||||||
|
|
||||||
|
def test_6g_open_rejected(self):
|
||||||
|
status, payload = server.h_pineap_wifi_set_ap(ctx({'open': {
|
||||||
|
'ssid': 'CorpGuest', 'hidden': False, 'enabled': True,
|
||||||
|
'channel': 181, 'country': 'US'}}))
|
||||||
|
self.assertEqual(status, 400)
|
||||||
|
self.assertIn('6GHz', payload['error'])
|
||||||
|
|
||||||
def test_6g_disable_removes_radio1(self):
|
def test_6g_disable_removes_radio1(self):
|
||||||
self.uci['pineapd.wlan1mon.hop'] = '0'
|
self.uci['pineapd.wlan1mon.hop'] = '0'
|
||||||
self.uci['wlan1wpa'] = {'device': 'radio1'}
|
self.uci['wlan1wpa'] = {'device': 'radio1'}
|
||||||
@@ -223,6 +236,29 @@ class SetApRadio1Test(unittest.TestCase):
|
|||||||
self.assertEqual(self.uci['wireless.radio1.channel'], 'auto')
|
self.assertEqual(self.uci['wireless.radio1.channel'], 'auto')
|
||||||
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')
|
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')
|
||||||
|
|
||||||
|
def test_5g_open_rejects_bad_bssid(self):
|
||||||
|
status, payload = server.h_pineap_wifi_set_ap(ctx({'open': {
|
||||||
|
'ssid': 'CorpGuest', 'bssid': 'not-a-mac', 'hidden': False,
|
||||||
|
'enabled': True, 'channel': 36, 'country': 'US'}}))
|
||||||
|
self.assertEqual(status, 400)
|
||||||
|
|
||||||
|
def test_hop_read_failure_still_pauses(self):
|
||||||
|
def fake_run(args, timeout=20, input_data=None):
|
||||||
|
self.runs.append((list(args), input_data))
|
||||||
|
a = list(args)
|
||||||
|
if a[:2] == ['uci', 'get']:
|
||||||
|
return (1, '', '')
|
||||||
|
if a[:2] == ['uci', 'set']:
|
||||||
|
k, _, v = a[2].partition('=')
|
||||||
|
self.uci[k] = v
|
||||||
|
return (0, '', '')
|
||||||
|
|
||||||
|
server.device_run = fake_run
|
||||||
|
server.h_pineap_wifi_set_ap(ctx({'open': {
|
||||||
|
'ssid': 'CorpGuest', 'hidden': False, 'enabled': True,
|
||||||
|
'channel': 36, 'country': 'US'}}))
|
||||||
|
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '0')
|
||||||
|
|
||||||
def test_2g_still_uses_daemon_path(self):
|
def test_2g_still_uses_daemon_path(self):
|
||||||
calls = []
|
calls = []
|
||||||
|
|
||||||
@@ -295,6 +331,7 @@ class GetApRadioChannelFallbackTest(unittest.TestCase):
|
|||||||
|
|
||||||
class GetApReconcileTest(unittest.TestCase):
|
class GetApReconcileTest(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
|
server._last_reconcile = 0.0
|
||||||
self.uci = {'wlan1wpa': {'device': 'radio1', 'mode': 'ap', 'ssid': 'Corp',
|
self.uci = {'wlan1wpa': {'device': 'radio1', 'mode': 'ap', 'ssid': 'Corp',
|
||||||
'disabled': '0', 'encryption': 'sae', 'channel': '36'}}
|
'disabled': '0', 'encryption': 'sae', 'channel': '36'}}
|
||||||
self.runs = []
|
self.runs = []
|
||||||
|
|||||||
Reference in New Issue
Block a user