1311bed5e9
Restore Gaia settings and media workflows, extend native network adapters, and align the APK payload. Generate French/English changelogs and website entries from the simplified JSON source. Validation: shared JS and Tundra unit tests; desktop workflows and persistence; signed APK payload parity; Sargo RAM and installed startup with boot readback. Android runtime qualification is recorded separately.
142 lines
7.6 KiB
Python
142 lines
7.6 KiB
Python
# SPDX-License-Identifier: MPL-2.0
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import socket
|
|
import stat
|
|
import struct
|
|
import tempfile
|
|
import threading
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('wifi', Path(__file__).resolve().parents[1]/'runtime/wifi-control.py')
|
|
wifi = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(wifi)
|
|
AP = {'id': '/org/freedesktop/NetworkManager/AccessPoint/7',
|
|
'ssid': 'Test: réseau\\', 'security': 'wpa-psk', 'signal': 75}
|
|
|
|
|
|
class WifiTests(unittest.TestCase):
|
|
def test_nmcli_escaped_names_and_enterprise(self):
|
|
raw = AP['id'] + r':Test\: réseau\\:75:WPA2:*' + '\n'
|
|
raw += '/org/freedesktop/NetworkManager/AccessPoint/8:Enterprise:60:WPA2 802.1X:\n'
|
|
with patch.object(wifi, 'nm', return_value=SimpleNamespace(returncode=0, stdout=raw)):
|
|
result = wifi.networks()
|
|
self.assertEqual(result[0]['ssid'], AP['ssid'])
|
|
self.assertTrue(result[0]['connected'])
|
|
self.assertEqual(result[1]['security'], 'unsupported')
|
|
|
|
def test_connection_information_uses_validated_address_and_actual_bitrate(self):
|
|
def nm(*args, **kwargs):
|
|
value='invalid\n192.0.2.4/24\n' if args[1]=='IP4.ADDRESS' else '/org/freedesktop/NetworkManager/Devices/4\n'
|
|
return SimpleNamespace(returncode=0,stdout=value)
|
|
with patch.object(wifi,'nm',side_effect=nm), patch.object(wifi.subprocess,'run',return_value=SimpleNamespace(returncode=0,stdout='{"data":[351000]}')):
|
|
self.assertEqual(wifi.connection_info(),{'ipAddress':'192.0.2.4','linkSpeed':351})
|
|
with patch.object(wifi,'nm',return_value=SimpleNamespace(returncode=0,stdout='/invalid/path')), patch.object(wifi.subprocess,'run') as run:
|
|
self.assertEqual(wifi.connection_info(),{})
|
|
run.assert_not_called()
|
|
|
|
def test_secret_persists_only_after_success(self):
|
|
for fail in (False, True):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
commands = []
|
|
def nm(*args, **kwargs):
|
|
commands.append(args)
|
|
if args[:2] == ('connection', 'load'):
|
|
file = Path(args[2])
|
|
self.assertEqual(stat.S_IMODE(file.stat().st_mode), 0o600)
|
|
self.assertIn('psk=testpassword', file.read_text())
|
|
return SimpleNamespace(returncode=int(fail and args[:2] == ('connection', 'up')))
|
|
with patch.object(wifi, 'networks', return_value=[AP]), patch.object(wifi, 'nm', side_effect=nm):
|
|
request = {'operation':'connect', 'id':AP['id'], 'password':'testpassword'}
|
|
if fail:
|
|
with self.assertRaisesRegex(wifi.WifiError, 'WIFI_CONNECT_FAILED'):
|
|
wifi.dispatch(request, root)
|
|
else:
|
|
self.assertTrue(wifi.dispatch(request, root)['connected'])
|
|
files=list(root.iterdir())
|
|
self.assertEqual(len(files), 0 if fail else 1)
|
|
if not fail:
|
|
self.assertIn('autoconnect=true', files[0].read_text())
|
|
self.assertEqual(wifi.saved_profiles(root)[0]['ssid'], AP['ssid'])
|
|
self.assertNotIn('testpassword', repr(commands))
|
|
|
|
def test_remember_reconnect_failed_password_and_forget(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
with patch.object(wifi, 'networks', return_value=[AP]), patch.object(wifi, 'nm', return_value=SimpleNamespace(returncode=0)):
|
|
wifi.dispatch({'operation':'connect','id':AP['id'],'password':'firstpassword'},root)
|
|
stored = wifi.dispatch({'operation':'known'},root)['networks'][0]
|
|
self.assertNotIn('password', stored)
|
|
wifi.dispatch({'operation':'connect','id':AP['id']},root)
|
|
target = next(root.iterdir())
|
|
before = target.read_text()
|
|
def failing(*args, **kwargs):
|
|
return SimpleNamespace(returncode=int(args[:2]==('connection','up')))
|
|
with patch.object(wifi, 'nm', side_effect=failing):
|
|
with self.assertRaises(wifi.WifiError):
|
|
wifi.dispatch({'operation':'connect','id':AP['id'],'password':'wrongpassword'},root)
|
|
self.assertEqual(target.read_text(),before)
|
|
wifi.dispatch({'operation':'forget','id':stored['id']},root)
|
|
self.assertEqual(list(root.iterdir()),[])
|
|
|
|
def test_rejects_commands_injection_and_unobserved_networks(self):
|
|
with patch.object(wifi, 'nm') as command:
|
|
for request in ({'operation':'execute'}, {'operation':'scan','command':'id'}, []):
|
|
with self.assertRaises(wifi.WifiError):
|
|
wifi.dispatch(request)
|
|
command.assert_not_called()
|
|
with patch.object(wifi, 'networks', return_value=[]):
|
|
with self.assertRaisesRegex(wifi.WifiError, 'WIFI_NETWORK_GONE'):
|
|
wifi.dispatch({'operation':'connect','id':'/tmp/hostile'})
|
|
for password in ('short', 'testpass\n[connection]', 'x'*100):
|
|
with self.assertRaises(wifi.WifiError):
|
|
wifi.profile_text(AP, password, 'test')
|
|
|
|
def test_hidden_network_uses_encoded_ssid_and_private_profile(self):
|
|
with tempfile.TemporaryDirectory() as folder, patch.object(wifi,'nm',return_value=SimpleNamespace(returncode=0)):
|
|
root=Path(folder)
|
|
wifi.dispatch({'operation':'connectHidden','ssid':'Hidden; réseau','security':'wpa-psk','password':'testpassword'},root)
|
|
saved=next(root.iterdir())
|
|
self.assertIn('hidden=true',saved.read_text())
|
|
self.assertEqual(stat.S_IMODE(saved.stat().st_mode),0o600)
|
|
self.assertEqual(wifi.saved_profiles(root)[0]['ssid'],'Hidden; réseau')
|
|
for ssid in ('','x'*33,'é'*17):
|
|
with tempfile.TemporaryDirectory() as folder, self.assertRaises(wifi.WifiError):
|
|
wifi.dispatch({'operation':'connectHidden','ssid':ssid,'security':'open'},Path(folder))
|
|
|
|
def test_socket_round_trip_and_limits(self):
|
|
a, b = socket.socketpair()
|
|
with a, b, patch.object(wifi, 'dispatch', return_value={'networks':[]}) as dispatch:
|
|
# Peer credential enforcement is independent of the CI user's uid.
|
|
class Peer:
|
|
def getsockopt(self, *args): return struct.pack('3i', 123, 32011, 32011)
|
|
def __getattr__(self, name): return getattr(b, name)
|
|
thread = threading.Thread(target=wifi.handle, args=(Peer(),))
|
|
thread.start()
|
|
a.sendall(b'{"operation":"scan"}\n')
|
|
self.assertEqual(wifi.receive(a, 65536), {'result':{'networks':[]}})
|
|
thread.join(2)
|
|
self.assertFalse(thread.is_alive())
|
|
dispatch.assert_called_once_with({'operation':'scan'})
|
|
a, b = socket.socketpair()
|
|
with a, b:
|
|
a.sendall(b'12345\n')
|
|
with self.assertRaises(wifi.WifiError): wifi.receive(b, 4)
|
|
|
|
def test_unauthorized_peer_never_dispatches(self):
|
|
class Peer:
|
|
def settimeout(self, _): pass
|
|
def getsockopt(self, *args): return struct.pack('3i', 123, 65534, 65534)
|
|
def sendall(self, data): self.reply = json.loads(data)
|
|
peer = Peer()
|
|
with patch.object(wifi, 'dispatch') as dispatch:
|
|
wifi.handle(peer)
|
|
dispatch.assert_not_called()
|
|
self.assertEqual(peer.reply, {'error':'PERMISSION_DENIED'})
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|