feat: truth-first get_ap (dual radio + enterprise) and derived pineap mode
get_ap now reports radio0, radio1 and enterprise APs separately from UCI (never a stored/cached branch), with radio channel fallback. Mode is derived from live enabled/collect state instead of showing 'unknown'; device truth wins over stale stored presets.
This commit is contained in:
@@ -2493,78 +2493,104 @@ def _uci_section(section):
|
||||
return _uci_values(section)
|
||||
|
||||
|
||||
def _ap_iface_dict(cfg, radio_cfg, pool=None):
|
||||
"""Normalize one wifi-iface UCI dict into the API shape used by the UI."""
|
||||
channel = cfg.get('channel') or ''
|
||||
try:
|
||||
channel = int(channel)
|
||||
except (TypeError, ValueError):
|
||||
channel = None
|
||||
if channel is None:
|
||||
# An AP without its own channel broadcasts on the radio's channel.
|
||||
try:
|
||||
channel = int((radio_cfg or {}).get('channel') or '')
|
||||
except (TypeError, ValueError):
|
||||
channel = None
|
||||
encryption = cfg.get('encryption') or ''
|
||||
if encryption.startswith('psk2'):
|
||||
encryption = 'psk2'
|
||||
elif encryption.startswith('sae'):
|
||||
encryption = 'sae'
|
||||
elif encryption.startswith('owe'):
|
||||
encryption = 'owe'
|
||||
elif encryption.startswith('wpa2'):
|
||||
encryption = 'wpa2'
|
||||
elif encryption.startswith('wpa3'):
|
||||
encryption = 'wpa3'
|
||||
return {
|
||||
'enabled': cfg.get('disabled') == '0',
|
||||
'ssid': cfg.get('ssid') or '',
|
||||
'hidden': cfg.get('hidden') == '1',
|
||||
'channel': channel,
|
||||
'country': (radio_cfg or {}).get('country') or '',
|
||||
'enctype': encryption,
|
||||
'passphrase': cfg.get('key') or '',
|
||||
'bssid': cfg.get('macaddr') or '',
|
||||
'target': (pool or {}).get('target') or None,
|
||||
}
|
||||
|
||||
|
||||
def _radio_dict(name):
|
||||
cfg = _uci_wifi_iface(name) or {}
|
||||
return {
|
||||
'band': {'2g': BAND_2G, '5g': BAND_5G, '6g': BAND_6G}.get(
|
||||
cfg.get('band'), None),
|
||||
'channel': cfg.get('channel') or 'auto',
|
||||
'htmode': cfg.get('htmode') or '',
|
||||
'country': cfg.get('country') or '',
|
||||
'disabled': cfg.get('disabled') == '1',
|
||||
}
|
||||
|
||||
|
||||
def _iface_live(name):
|
||||
return os.path.exists('/sys/class/net/%s' % name)
|
||||
|
||||
|
||||
def h_pineap_wifi_get_ap(ctx):
|
||||
global _last_reconcile
|
||||
|
||||
def _iface_state(open_name, wpa_name, radio_name):
|
||||
open_cfg = _uci_wifi_iface(open_name)
|
||||
wpa_cfg = _uci_wifi_iface(wpa_name)
|
||||
radio_cfg = _uci_wifi_iface(radio_name)
|
||||
|
||||
def _chan(cfg):
|
||||
channel = cfg.get('channel') or ''
|
||||
try:
|
||||
return int(channel)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
radio_channel = _chan(radio_cfg)
|
||||
open_channel = _chan(open_cfg) or radio_channel
|
||||
wpa_channel = _chan(wpa_cfg) or radio_channel
|
||||
encryption = wpa_cfg.get('encryption') or ''
|
||||
if encryption.startswith('psk2'):
|
||||
encryption = 'psk2'
|
||||
elif encryption.startswith('sae'):
|
||||
encryption = 'sae'
|
||||
elif encryption.startswith('owe'):
|
||||
encryption = 'owe'
|
||||
return open_cfg, wpa_cfg, radio_cfg, open_channel, wpa_channel, encryption
|
||||
|
||||
if _uci_wifi_iface('wlan1open') or _uci_wifi_iface('wlan1wpa'):
|
||||
open_cfg, wpa_cfg, radio_cfg, open_channel, wpa_channel, encryption = _iface_state(
|
||||
'wlan1open', 'wlan1wpa', 'radio1')
|
||||
else:
|
||||
open_cfg, wpa_cfg, radio_cfg, open_channel, wpa_channel, encryption = _iface_state(
|
||||
'wlan0open', 'wlan0wpa', 'radio0')
|
||||
pool = _uci_section('pineapd.@ssidpool[0]')
|
||||
radio0_cfg = _uci_wifi_iface('radio0')
|
||||
radio1_cfg = _uci_wifi_iface('radio1')
|
||||
open_cfg = _uci_wifi_iface('wlan0open')
|
||||
wpa_cfg = _uci_wifi_iface('wlan0wpa')
|
||||
r1_open_cfg = _uci_wifi_iface('wlan1open')
|
||||
r1_wpa_cfg = _uci_wifi_iface('wlan1wpa')
|
||||
ent_cfg = _uci_wifi_iface('wlan0ent')
|
||||
status, data = daemon_sock_call('GET', '/api/pineap/hostapd/get_config')
|
||||
host = data if status == 200 and isinstance(data, dict) else {}
|
||||
status2, data2 = daemon_sock_call('GET', '/api/pineap/get_config')
|
||||
pinecfg = data2 if status2 == 200 and isinstance(data2, dict) else {}
|
||||
pool = _uci_section('pineapd.@ssidpool[0]')
|
||||
radio1 = _uci_wifi_iface('radio1')
|
||||
for name in ('wlan1open', 'wlan1wpa'):
|
||||
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 _iface_live(name):
|
||||
if time.time() - _last_reconcile > 30:
|
||||
_last_reconcile = time.time()
|
||||
device_run(['wifi', 'reload'])
|
||||
break
|
||||
ent = _ap_iface_dict(ent_cfg, radio0_cfg)
|
||||
return 200, {
|
||||
'open': {
|
||||
'enabled': open_cfg.get('disabled') == '0',
|
||||
'ssid': open_cfg.get('ssid') or '',
|
||||
'bssid': open_cfg.get('macaddr') or '',
|
||||
'target': pool.get('target') or None,
|
||||
'hidden': open_cfg.get('hidden') == '1',
|
||||
'channel': open_channel,
|
||||
'country': radio_cfg.get('country') or '',
|
||||
'open': _ap_iface_dict(open_cfg, radio0_cfg, pool),
|
||||
'wpa': _ap_iface_dict(wpa_cfg, radio0_cfg),
|
||||
'radio1_open': _ap_iface_dict(r1_open_cfg, radio1_cfg),
|
||||
'radio1_wpa': _ap_iface_dict(r1_wpa_cfg, radio1_cfg),
|
||||
'enterprise': {
|
||||
'enabled': ent.get('enabled', False),
|
||||
'ssid': ent.get('ssid', ''),
|
||||
'enctype': ent.get('enctype') or 'wpa2',
|
||||
'passphrase': ent.get('passphrase', ''),
|
||||
'hidden': ent.get('hidden', False),
|
||||
'channel': ent.get('channel'),
|
||||
'live': _iface_live('wlan0ent'),
|
||||
},
|
||||
'wpa': {
|
||||
'ssid': wpa_cfg.get('ssid') or '',
|
||||
'passphrase': wpa_cfg.get('key') or '',
|
||||
'enctype': encryption,
|
||||
'hidden': wpa_cfg.get('hidden') == '1',
|
||||
'enabled': wpa_cfg.get('disabled') == '0',
|
||||
'channel': wpa_channel,
|
||||
},
|
||||
'enterprise': {'enabled': not host.get('pineape_disabled', True)},
|
||||
'pool': {'disabled': None, 'collecting': bool(pinecfg.get('autossidpool'))},
|
||||
'radio1': {
|
||||
'band': {'2g': BAND_2G, '5g': BAND_5G, '6g': BAND_6G}.get(
|
||||
(radio1 or {}).get('band'), BAND_5G),
|
||||
'channel': (radio1 or {}).get('channel') or 'auto',
|
||||
'htmode': (radio1 or {}).get('htmode') or 'VHT80',
|
||||
'country': (radio1 or {}).get('country') or '',
|
||||
'radios': {
|
||||
'radio0': _radio_dict('radio0'),
|
||||
'radio1': _radio_dict('radio1'),
|
||||
},
|
||||
'pineape': {
|
||||
'enabled': not host.get('pineape_disabled', True),
|
||||
'auth_pass': bool(host.get('pineape_auth_pass')),
|
||||
},
|
||||
}
|
||||
|
||||
@@ -2840,7 +2866,7 @@ def update_pineap_state(mode=None, **flags):
|
||||
|
||||
def h_pineap_mode_get(ctx):
|
||||
state = load_pineap_state()
|
||||
mode = state.get('mode')
|
||||
stored = state.get('mode')
|
||||
_, config = daemon_sock_call('GET', '/api/pineap/get_config')
|
||||
_, hostapd = daemon_sock_call('GET', '/api/pineap/hostapd/get_config')
|
||||
collect = config.get('autossidpool') if isinstance(config, dict) else None
|
||||
@@ -2848,18 +2874,21 @@ def h_pineap_mode_get(ctx):
|
||||
if isinstance(hostapd, dict) and 'pineap_disabled' in hostapd:
|
||||
enabled = not bool(hostapd['pineap_disabled'])
|
||||
|
||||
# On the Pager, the Mimic/PineAP switch is the response engine itself:
|
||||
# Passive intentionally leaves it disabled, while Active enables it.
|
||||
# Treat only a real mismatch with that preset (or disabled collection) as
|
||||
# a custom/Advanced setup.
|
||||
expected_enabled = {'passive': False, 'active': True}.get(mode)
|
||||
engine_mismatch = (enabled is not None and expected_enabled is not None
|
||||
and enabled != expected_enabled)
|
||||
if mode in ('passive', 'active') and (collect is False or engine_mismatch):
|
||||
# The mode is DERIVED from the live device state, never invented.
|
||||
if enabled is not None and collect is not None:
|
||||
live = 'passive' if enabled is False else ('active' if collect else 'advanced')
|
||||
if stored == 'advanced':
|
||||
mode = 'advanced'
|
||||
elif stored == live:
|
||||
mode = stored
|
||||
else:
|
||||
mode = live
|
||||
state = update_pineap_state(mode=live, enabled=enabled, collect=collect,
|
||||
karma=bool(enabled))
|
||||
elif stored in ('passive', 'active', 'advanced'):
|
||||
mode = stored
|
||||
else:
|
||||
mode = 'advanced'
|
||||
state = update_pineap_state(mode='advanced', collect=collect)
|
||||
elif mode not in ('passive', 'active', 'advanced'):
|
||||
mode = 'advanced' if collect is False or enabled is False else 'unknown'
|
||||
|
||||
result = dict(state)
|
||||
result['mode'] = mode
|
||||
|
||||
@@ -111,22 +111,26 @@ class GetApRadio1Test(unittest.TestCase):
|
||||
def test_open_reports_radio1_when_present(self):
|
||||
status, payload = server.h_pineap_wifi_get_ap(ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload['open']['ssid'], 'CorpGuest')
|
||||
self.assertEqual(payload['open']['channel'], 36)
|
||||
self.assertEqual(payload['open']['country'], 'US')
|
||||
self.assertEqual(payload['radio1_open']['ssid'], 'CorpGuest')
|
||||
self.assertEqual(payload['radio1_open']['channel'], 36)
|
||||
self.assertEqual(payload['radio1_open']['country'], 'US')
|
||||
# radio0 cards keep reporting radio0 truth
|
||||
self.assertEqual(payload['open']['ssid'], 'pager-open')
|
||||
|
||||
def test_wpa_reports_radio1_when_present(self):
|
||||
status, payload = server.h_pineap_wifi_get_ap(ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload['wpa']['ssid'], 'Corp')
|
||||
self.assertEqual(payload['wpa']['enctype'], 'sae')
|
||||
self.assertEqual(payload['wpa']['channel'], 1)
|
||||
self.assertEqual(payload['radio1_wpa']['ssid'], 'Corp')
|
||||
self.assertEqual(payload['radio1_wpa']['enctype'], 'sae')
|
||||
self.assertEqual(payload['radio1_wpa']['channel'], 1)
|
||||
self.assertEqual(payload['wpa']['ssid'], 'Service')
|
||||
|
||||
def test_radio1_info(self):
|
||||
status, payload = server.h_pineap_wifi_get_ap(ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload['radio1']['band'], server.BAND_5G)
|
||||
self.assertEqual(payload['radio1']['channel'], 'auto')
|
||||
self.assertEqual(payload['radios']['radio1']['band'], server.BAND_5G)
|
||||
self.assertEqual(payload['radios']['radio1']['channel'], 'auto')
|
||||
self.assertEqual(payload['radios']['radio0']['band'], server.BAND_2G)
|
||||
|
||||
|
||||
class GetApRadio1AbsentTest(unittest.TestCase):
|
||||
|
||||
@@ -109,7 +109,8 @@ class PineapModeTest(unittest.TestCase):
|
||||
server.daemon_sock_call = fake
|
||||
status, payload = server.h_pineap_mode_get(ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload['mode'], 'advanced')
|
||||
# Device truth wins: engine off means the device IS passive now.
|
||||
self.assertEqual(payload['mode'], 'passive')
|
||||
self.assertFalse(payload['enabled'])
|
||||
|
||||
def test_manual_mimic_change_marks_advanced(self):
|
||||
|
||||
@@ -112,7 +112,7 @@ class PineapProxyTest(unittest.TestCase):
|
||||
if sec == 'wireless.wlan0open':
|
||||
return 0, "wireless.wlan0open.disabled='1'\nwireless.wlan0open.ssid='pager-open'\nwireless.wlan0open.macaddr='DE:AD:BE:EF:00:01'\nwireless.wlan0open.hidden='1'\n", ''
|
||||
if sec == 'wireless.radio0':
|
||||
return 0, "wireless.radio0.channel='6'\nwireless.radio0.country='US'\n", ''
|
||||
return 0, "wireless.radio0.band='2g'\nwireless.radio0.channel='6'\nwireless.radio0.country='US'\n", ''
|
||||
if sec.startswith('pineapd.@ssidpool'):
|
||||
return 0, "pineapd.@ssidpool[0].bssid='auto'\npineapd.@ssidpool[0].target='broadcast'\n", ''
|
||||
return 0, '', ''
|
||||
@@ -128,8 +128,12 @@ class PineapProxyTest(unittest.TestCase):
|
||||
server.daemon_sock_call = fake_sock
|
||||
status, payload = server.h_pineap_wifi_get_ap(ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload['wpa'], {'ssid': 'Evil1', 'passphrase': 'sekret', 'enctype': 'psk2',
|
||||
'hidden': False, 'enabled': True, 'channel': 6})
|
||||
self.assertEqual(payload['wpa']['ssid'], 'Evil1')
|
||||
self.assertEqual(payload['wpa']['passphrase'], 'sekret')
|
||||
self.assertEqual(payload['wpa']['enctype'], 'psk2')
|
||||
self.assertEqual(payload['wpa']['hidden'], False)
|
||||
self.assertEqual(payload['wpa']['enabled'], True)
|
||||
self.assertEqual(payload['wpa']['channel'], 6)
|
||||
self.assertEqual(payload['open']['enabled'], False)
|
||||
self.assertEqual(payload['open']['ssid'], 'pager-open')
|
||||
self.assertEqual(payload['open']['bssid'], 'DE:AD:BE:EF:00:01')
|
||||
@@ -137,8 +141,12 @@ class PineapProxyTest(unittest.TestCase):
|
||||
self.assertEqual(payload['open']['channel'], 6)
|
||||
self.assertEqual(payload['open']['country'], 'US')
|
||||
self.assertEqual(payload['open']['target'], 'broadcast')
|
||||
self.assertEqual(payload['enterprise']['enabled'], True)
|
||||
self.assertEqual(payload['enterprise']['enabled'], False)
|
||||
self.assertEqual(payload['enterprise']['ssid'], '')
|
||||
self.assertEqual(payload['pool']['collecting'], True)
|
||||
self.assertEqual(payload['radios']['radio0']['band'], '2.4')
|
||||
self.assertEqual(payload['radios']['radio1']['channel'], 'auto')
|
||||
self.assertEqual(payload['pineape']['enabled'], True)
|
||||
|
||||
def test_wifi_set_ap_open_bssid_channel_and_country(self):
|
||||
sock_calls = []
|
||||
|
||||
Reference in New Issue
Block a user