oversu/VulpesOS
oversu/VulpesOS
Watch
3
2
Fork
You've already forked VulpesOS
 0
Code Issues 27 Pull requests Projects Releases 9 Packages Wiki Activity

Publish Vulpes Preview 15 sources for desktop, Android and Tundra

Browse source 
Share Gaia assets and adapters, keep private profiles and development documents local, and update both public READMEs.
This commit is contained in:
OverSu 2026-09-30 23:13:03 +02:00
commit b38dea0597
25901 changed files with 9982925 additions and 369 deletions
Download patch file Download diff file Expand all files Collapse all files

62
tundra/tests/fixtures/pulse14/cards.txt vendored Normal file
View file

@ -0,0 +1,62 @@
Card #0
Name: droid_card.primary
Driver: module-droid-card.c
Owner Module: 1
Properties:
droid.hw_module = "primary"
device.icon_name = "audio-card"
Profiles:
default: Default profile (sinks: 2, sources: 1, priority: 300, available: yes)
voicecall: Call mode (sinks: 0, sources: 0, priority: 0, available: yes)
voicecall-record: Call mode record (sinks: 0, sources: 0, priority: 0, available: yes)
communication: Communication mode (sinks: 0, sources: 0, priority: 0, available: yes)
ringtone: Ringtone mode (sinks: 0, sources: 0, priority: 0, available: yes)
off: Off (sinks: 0, sources: 0, priority: 0, available: yes)
Active Profile: default
Ports:
output-earpiece: Output to earpiece (type: Unknown, priority: 200, latency offset: 0 usec, available)
Part of profile(s): default
output-speaker: Output to speaker (type: Unknown, priority: 300, latency offset: 0 usec, available)
Part of profile(s): default
output-wired_headset: Output to wired headset (type: Unknown, priority: 100, latency offset: 0 usec, not available)
Part of profile(s): default
output-wired_headphone: Output to wired headphone (type: Unknown, priority: 100, latency offset: 0 usec, not available)
Part of profile(s): default
output-bluetooth_sco: Output to bluetooth sco (type: Unknown, priority: 100, latency offset: 0 usec, available)
Part of profile(s): default
output-sco_headset: Output to sco headset (type: Unknown, priority: 100, latency offset: 0 usec, available)
Part of profile(s): default
output-sco_carkit: Output to sco carkit (type: Unknown, priority: 100, latency offset: 0 usec, available)
Part of profile(s): default
output-a2dp: Output to a2dp (type: Unknown, priority: 100, latency offset: 0 usec, available)
Part of profile(s): default
output-a2dp_headphones: Output to a2dp headphones (type: Unknown, priority: 100, latency offset: 0 usec, available)
Part of profile(s): default
output-a2dp_speaker: Output to a2dp speaker (type: Unknown, priority: 100, latency offset: 0 usec, available)
Part of profile(s): default
output-usb_device: Output to usb device (type: Unknown, priority: 100, latency offset: 0 usec, available)
Part of profile(s): default
output-line: Output to line (type: Unknown, priority: 100, latency offset: 0 usec, available)
Part of profile(s): default
output-speaker_safe: Output to speaker safe (type: Unknown, priority: 200, latency offset: 0 usec, available)
Part of profile(s): default
output-speaker+wired_headphone: Output to speaker+wired headphone (type: Unknown, priority: 100, latency offset: 0 usec, not available)
Part of profile(s): default
output-parking: Parking port (type: Unknown, priority: 50, latency offset: 0 usec, available)
Part of profile(s): default
input-builtin_mic: Input from builtin mic (type: Unknown, priority: 200, latency offset: 0 usec, available)
Part of profile(s): default
input-bluetooth_sco_headset: Input from bluetooth sco headset (type: Unknown, priority: 200, latency offset: 0 usec, available)
Part of profile(s): default
input-wired_headset: Input from wired headset (type: Unknown, priority: 200, latency offset: 0 usec, not available)
Part of profile(s): default
input-voice_call: Input from voice call (type: Unknown, priority: 200, latency offset: 0 usec, available)
Part of profile(s): default
input-back_mic: Input from back mic (type: Unknown, priority: 200, latency offset: 0 usec, available)
Part of profile(s): default
input-usb_device: Input from usb device (type: Unknown, priority: 200, latency offset: 0 usec, available)
Part of profile(s): default
input-default: Input from default (type: Unknown, priority: 200, latency offset: 0 usec, available)
Part of profile(s): default
input-parking: Parking port (type: Unknown, priority: 50, latency offset: 0 usec, available)
Part of profile(s): default

87
tundra/tests/fixtures/pulse14/sinks.txt vendored Normal file
View file

@ -0,0 +1,87 @@
Sink #0
State: SUSPENDED
Name: sink.primary_output
Description: Droid sink primary_output
Driver: module-droid-card.c
Sample Specification: s16le 2ch 48000Hz
Channel Map: front-left,front-right
Owner Module: 1
Mute: no
Volume: front-left: 65536 / 100% / 0.00 dB, front-right: 65536 / 100% / 0.00 dB
balance 0.00
Base Volume: 65536 / 100% / 0.00 dB
Monitor Source: sink.primary_output.monitor
Latency: 0 usec, configured 0 usec
Flags: HARDWARE DECIBEL_VOLUME LATENCY
Properties:
device.description = "Droid sink primary_output"
device.class = "sound"
device.api = "droid-hal"
droid.flags = "AUDIO_OUTPUT_FLAG_PRIMARY|AUDIO_OUTPUT_FLAG_FAST"
droid.hw_module = "primary"
device.icon_name = "audio-card"
droid.output.primary = "true"
droid.output.low_latency = "true"
droid.output.media_latency = "false"
Ports:
output-earpiece: Output to earpiece (type: Unknown, priority: 200, available)
output-speaker: Output to speaker (type: Unknown, priority: 300, available)
output-wired_headset: Output to wired headset (type: Unknown, priority: 100, not available)
output-wired_headphone: Output to wired headphone (type: Unknown, priority: 100, not available)
output-bluetooth_sco: Output to bluetooth sco (type: Unknown, priority: 100, available)
output-sco_headset: Output to sco headset (type: Unknown, priority: 100, available)
output-sco_carkit: Output to sco carkit (type: Unknown, priority: 100, available)
output-a2dp: Output to a2dp (type: Unknown, priority: 100, available)
output-a2dp_headphones: Output to a2dp headphones (type: Unknown, priority: 100, available)
output-a2dp_speaker: Output to a2dp speaker (type: Unknown, priority: 100, available)
output-usb_device: Output to usb device (type: Unknown, priority: 100, available)
output-line: Output to line (type: Unknown, priority: 100, available)
output-speaker_safe: Output to speaker safe (type: Unknown, priority: 200, available)
output-speaker+wired_headphone: Output to speaker+wired headphone (type: Unknown, priority: 100, not available)
output-parking: Parking port (type: Unknown, priority: 50, available)
Active Port: output-speaker
Formats:
pcm
Sink #1
State: SUSPENDED
Name: sink.deep_buffer
Description: Droid sink deep_buffer
Driver: module-droid-card.c
Sample Specification: s16le 2ch 48000Hz
Channel Map: front-left,front-right
Owner Module: 1
Mute: no
Volume: front-left: 65536 / 100% / 0.00 dB, front-right: 65536 / 100% / 0.00 dB
balance 0.00
Base Volume: 65536 / 100% / 0.00 dB
Monitor Source: sink.deep_buffer.monitor
Latency: 0 usec, configured 0 usec
Flags: HARDWARE DECIBEL_VOLUME LATENCY
Properties:
device.description = "Droid sink deep_buffer"
device.class = "sound"
device.api = "droid-hal"
droid.flags = "AUDIO_OUTPUT_FLAG_DEEP_BUFFER"
droid.hw_module = "primary"
device.icon_name = "audio-card"
droid.output.media_latency = "true"
Ports:
output-earpiece: Output to earpiece (type: Unknown, priority: 200, available)
output-speaker: Output to speaker (type: Unknown, priority: 300, available)
output-wired_headset: Output to wired headset (type: Unknown, priority: 100, not available)
output-wired_headphone: Output to wired headphone (type: Unknown, priority: 100, not available)
output-bluetooth_sco: Output to bluetooth sco (type: Unknown, priority: 100, available)
output-sco_headset: Output to sco headset (type: Unknown, priority: 100, available)
output-sco_carkit: Output to sco carkit (type: Unknown, priority: 100, available)
output-a2dp: Output to a2dp (type: Unknown, priority: 100, available)
output-a2dp_headphones: Output to a2dp headphones (type: Unknown, priority: 100, available)
output-a2dp_speaker: Output to a2dp speaker (type: Unknown, priority: 100, available)
output-usb_device: Output to usb device (type: Unknown, priority: 100, available)
output-line: Output to line (type: Unknown, priority: 100, available)
output-speaker_safe: Output to speaker safe (type: Unknown, priority: 200, available)
output-speaker+wired_headphone: Output to speaker+wired headphone (type: Unknown, priority: 100, not available)
output-parking: Parking port (type: Unknown, priority: 50, available)
Active Port: output-speaker
Formats:
pcm

81
tundra/tests/fixtures/pulse14/sources.txt vendored Normal file
View file

@ -0,0 +1,81 @@
Source #0
State: SUSPENDED
Name: sink.primary_output.monitor
Description: Monitor of Droid sink primary_output
Driver: module-droid-card.c
Sample Specification: s16le 2ch 48000Hz
Channel Map: front-left,front-right
Owner Module: 1
Mute: no
Volume: front-left: 65536 / 100% / 0.00 dB, front-right: 65536 / 100% / 0.00 dB
balance 0.00
Base Volume: 65536 / 100% / 0.00 dB
Monitor of Sink: sink.primary_output
Latency: 0 usec, configured 0 usec
Flags: DECIBEL_VOLUME LATENCY
Properties:
device.description = "Monitor of Droid sink primary_output"
device.class = "monitor"
droid.hw_module = "primary"
device.icon_name = "audio-card"
Formats:
pcm
Source #1
State: SUSPENDED
Name: sink.deep_buffer.monitor
Description: Monitor of Droid sink deep_buffer
Driver: module-droid-card.c
Sample Specification: s16le 2ch 48000Hz
Channel Map: front-left,front-right
Owner Module: 1
Mute: no
Volume: front-left: 65536 / 100% / 0.00 dB, front-right: 65536 / 100% / 0.00 dB
balance 0.00
Base Volume: 65536 / 100% / 0.00 dB
Monitor of Sink: sink.deep_buffer
Latency: 0 usec, configured 0 usec
Flags: DECIBEL_VOLUME LATENCY
Properties:
device.description = "Monitor of Droid sink deep_buffer"
device.class = "monitor"
droid.hw_module = "primary"
device.icon_name = "audio-card"
Formats:
pcm
Source #2
State: SUSPENDED
Name: source.droid
Description: Droid source droid
Driver: module-droid-card.c
Sample Specification: s16le 2ch 48000Hz
Channel Map: front-left,front-right
Owner Module: 1
Mute: no
Volume: front-left: 65536 / 100% / 0.00 dB, front-right: 65536 / 100% / 0.00 dB
balance 0.00
Base Volume: 65536 / 100% / 0.00 dB
Monitor of Sink: n/a
Latency: 0 usec, configured 0 usec
Flags: HARDWARE HW_MUTE_CTRL DECIBEL_VOLUME
Properties:
device.description = "Droid source droid"
device.class = "sound"
device.api = "droid-hal"
droid.input.external = "true"
droid.input.builtin = "true"
droid.hw_module = "primary"
device.icon_name = "audio-card"
Ports:
input-builtin_mic: Input from builtin mic (type: Unknown, priority: 200, available)
input-bluetooth_sco_headset: Input from bluetooth sco headset (type: Unknown, priority: 200, available)
input-wired_headset: Input from wired headset (type: Unknown, priority: 200, not available)
input-voice_call: Input from voice call (type: Unknown, priority: 200, available)
input-back_mic: Input from back mic (type: Unknown, priority: 200, available)
input-usb_device: Input from usb device (type: Unknown, priority: 200, available)
input-default: Input from default (type: Unknown, priority: 200, available)
input-parking: Parking port (type: Unknown, priority: 50, available)
Active Port: input-builtin_mic
Formats:
pcm

46
tundra/tests/test_boot_backup.py Normal file
View file

@ -0,0 +1,46 @@
# SPDX-License-Identifier: MPL-2.0
import hashlib
import importlib.util
import io
from pathlib import Path
import unittest
from unittest.mock import patch
SOURCE = Path(__file__).resolve().parents[1] / 'droidian/backup-boot.py'
spec = importlib.util.spec_from_file_location('backup_boot', SOURCE)
backup = importlib.util.module_from_spec(spec)
spec.loader.exec_module(backup)
class BootBackupTests(unittest.TestCase):
def test_copy_is_exact_even_with_more_input(self):
data = b'sargo boot' * 200000
source, destination = io.BytesIO(data + b'next partition'), io.BytesIO()
self.assertEqual(backup.copy_exact(source, destination, len(data)),
hashlib.sha256(data).hexdigest())
self.assertEqual(destination.getvalue(), data)
self.assertEqual(source.read(), b'next partition')
def test_short_partition_read_is_not_reported_as_success(self):
with self.assertRaisesRegex(RuntimeError, 'before its declared size'):
backup.copy_exact(io.BytesIO(b'incomplete'), io.BytesIO(), 1024)
def test_no_privilege_means_no_output_or_device_access(self):
with patch.object(backup.os, 'geteuid', return_value=32011), \
patch.object(backup.tempfile, 'mkdtemp') as mkdir, \
patch.object(backup.os, 'open') as open_device:
with self.assertRaisesRegex(RuntimeError, 'require sudo'):
backup.main()
mkdir.assert_not_called()
open_device.assert_not_called()
def test_other_device_is_rejected_before_output(self):
with patch.object(backup.os, 'geteuid', return_value=0), \
patch.object(backup.pwd, 'getpwnam'), \
patch.object(backup.Path, 'exists', return_value=True), \
patch.object(backup.Path, 'read_bytes', return_value=b'google,bonito\0'), \
patch.object(backup.subprocess, 'check_output', return_value='bonito\n'), \
patch.object(backup.tempfile, 'mkdtemp') as mkdir:
with self.assertRaisesRegex(RuntimeError, 'inventoried Sargo'):
backup.main()
mkdir.assert_not_called()

36
tundra/tests/test_build_sources.py Normal file
View file

@ -0,0 +1,36 @@
# SPDX-License-Identifier: MPL-2.0
import importlib.util
from pathlib import Path
import sys
import tempfile
import unittest
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'tools'))
spec=importlib.util.spec_from_file_location('rescue_test',ROOT/'tools/test-rescue-boot.py')
rescue=importlib.util.module_from_spec(spec);spec.loader.exec_module(rescue)
class SourceSnapshotTests(unittest.TestCase):
def setUp(self):
self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
self.folder=Path(self.temp.name)
self.source=self.folder/'sources/tundra/boot/init.sh'
self.source.parent.mkdir(parents=True)
self.source.write_text('immutable boot source')
self.build={'sourceSnapshot':True,'sources':{'boot/init.sh':rescue.digest(self.source)}}
def test_snapshot_does_not_depend_on_current_checkout(self):
rescue.verify_sources(self.folder,self.build)
def test_tampering_with_saved_source_is_rejected(self):
self.source.write_text('changed')
with self.assertRaisesRegex(ValueError,'source changed'):
rescue.verify_sources(self.folder,self.build)
def test_path_cannot_escape_snapshot(self):
self.build['sources']={'../../outside':'hash'}
with self.assertRaises(ValueError):rescue.verify_sources(self.folder,self.build)
if __name__=='__main__':unittest.main()

34
tundra/tests/test_buttons.py Normal file
View file

@ -0,0 +1,34 @@
import importlib.util
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
s=importlib.util.spec_from_file_location('buttons',Path(__file__).resolve().parents[1]/'runtime/buttons.py')
b=importlib.util.module_from_spec(s);s.loader.exec_module(b)
class Buttons(unittest.TestCase):
def test_only_physical_control_edges(self):
self.assertEqual(b.key_event(1,116,1),'sleep-button-press')
self.assertEqual(b.key_event(1,115,0),'volume-up-button-release')
for event in [(3,116,1),(1,30,1),(1,116,2)]:self.assertIsNone(b.key_event(*event))
def test_backlight_validation_and_wake(self):
with tempfile.TemporaryDirectory() as d,patch.object(b,'BACKLIGHT',Path(d)):
(Path(d)/'max_brightness').write_text('1023')
b.control({'operation':'screen','enabled':False,'level':.5})
self.assertEqual((Path(d)/'brightness').read_text(),'0')
b.control({'operation':'screen','enabled':True,'level':.5})
self.assertEqual((Path(d)/'brightness').read_text(),'512')
for p in [{'operation':'screen','enabled':True,'level':-1},{'operation':'volume','level':float('nan')},{'operation':'exec','command':'true'}]:
with self.assertRaises(ValueError):b.control(p)
def test_vibration_pattern_progresses_without_blocking_and_can_be_cancelled(self):
with tempfile.TemporaryDirectory() as d,patch.object(b,'VIBRATOR',Path(d)),patch.object(b.time,'monotonic',return_value=10):
v=b.Vibration();v.start([200,100,300])
self.assertEqual((Path(d)/'brightness').read_text(),'255')
self.assertEqual(v.deadline,10.2)
v.advance(10.1);self.assertEqual((Path(d)/'brightness').read_text(),'255')
v.advance(10.2);self.assertEqual((Path(d)/'brightness').read_text(),'0')
v.advance(10.31);self.assertEqual((Path(d)/'brightness').read_text(),'255')
v.advance(10.62);self.assertEqual((Path(d)/'brightness').read_text(),'0')
v.start([200]);v.start([]);self.assertIsNone(v.deadline);self.assertEqual((Path(d)/'brightness').read_text(),'0')
for pattern in [None,[-1],[True],[10001],[5000,5001],[1]*33]:
with self.assertRaises(ValueError):v.start(pattern)

62
tundra/tests/test_call_audio.py Normal file
View file

@ -0,0 +1,62 @@
import importlib.util,json,unittest
from pathlib import Path
spec=importlib.util.spec_from_file_location('audio',Path(__file__).resolve().parents[1]/'runtime/call-audio.py')
audio=importlib.util.module_from_spec(spec);spec.loader.exec_module(audio)
class AudioTests(unittest.TestCase):
def rig(self):
commands=[]
devices={'cards':[{'index':0,'owner_module':1,'name':'droid_card.primary','profiles':{'default':{},'voicecall':{}},'active_profile':'default'}],
'sinks':[{'owner_module':1,'name':'sink.primary_output','ports':[{'name':'output-earpiece'},{'name':'output-parking'},{'name':'output-speaker'}],'active_port':'output-speaker','mute':True}],
'sources':[{'name':'source.droid','mute':True}],
'sink-inputs':[{'index':9,'properties':{'media.role':'phone'}}]}
def command(*args):
commands.append(args)
return ''
route=audio.CallAudio(command)
route.devices=lambda kind:devices[kind]
return route,commands,devices
def test_actual_pulse14_listing_and_routing(self):
fixture=Path(__file__).parent/'fixtures/pulse14'
commands=[]
def command(*args):
commands.append(args)
if args[0]=='list':
return (fixture/(args[1]+'.txt')).read_text() if args[1]!='sink-inputs' else ''
return ''
route=audio.CallAudio(command)
self.assertEqual(len(route.devices('cards')),1)
self.assertEqual(route.devices('cards')[0]['owner_module'],1)
self.assertIn('output-earpiece',route.devices('sinks')[0]['ports'])
route.enter();self.assertTrue(route.active)
route.leave();self.assertIsNone(route.saved)
self.assertIn(('set-sink-port','sink.primary_output','output-earpiece'),commands)
def test_active_call_routes_earpiece_then_restores(self):
route,c,_=self.rig()
route.update([{'state':'active'}]);n=len(c)
route.update([{'state':'active'}]);self.assertEqual(len(c),n)
self.assertIn(('set-card-profile','droid_card.primary','voicecall'),c)
self.assertIn(('set-sink-port','sink.primary_output','output-earpiece'),c)
self.assertNotIn(('set-sink-port','sink.primary_output','output-speaker'),c)
self.assertIn(('set-source-mute','source.droid','0'),c)
route.update([])
self.assertIsNone(route.saved)
self.assertIn(('set-card-profile','droid_card.primary','default'),c)
self.assertIn(('set-source-mute','source.droid','1'),c)
def test_controls_route_speaker_mute_and_phone_stream(self):
route,c,_=self.rig();route.enter()
result=route.control({'operation':'set','speaker':True,'muted':True,'volume':0.8})
self.assertTrue(result['speaker']);self.assertTrue(result['muted'])
self.assertIn(('set-sink-port','sink.primary_output','output-speaker'),c)
self.assertIn(('set-source-mute','source.droid','1'),c)
self.assertIn(('set-sink-input-volume','9','80%'),c)
route.leave();self.assertFalse(route.muted);self.assertFalse(route.speaker)
with self.assertRaisesRegex(ValueError,'NOT_ACTIVE'):route.control({'operation':'set','muted':True})
def test_ringing_alone_does_not_open_microphone(self):
route,c,_=self.rig();route.update([{'state':'incoming'}]);self.assertEqual(c,[])
def test_missing_earpiece_never_falls_back_to_speaker(self):
route,c,d=self.rig();d['sinks'][0]['ports']=[{'name':'output-speaker'}]
with self.assertRaisesRegex(RuntimeError,'EARPIECE'):route.update([{'state':'active'}])
self.assertFalse(any(x[0].startswith('set-') for x in c))
if __name__=='__main__':unittest.main()

38
tundra/tests/test_fonts.py Normal file
View file

@ -0,0 +1,38 @@
"""Resolve generic text fonts against the fonts shipped in the native image."""
import importlib.util
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
from xml.sax.saxutils import escape
ROOT = Path(__file__).resolve().parents[1]
@unittest.skipUnless(shutil.which('fc-match'), 'Fontconfig is needed for font resolution')
class NativeFontsTest(unittest.TestCase):
def test_generic_and_explicit_names_resolve_to_shipped_fira(self):
spec = importlib.util.spec_from_file_location('native_fonts', ROOT/'runtime/prepare-native.py')
native = importlib.util.module_from_spec(spec)
spec.loader.exec_module(native)
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root/'opt/vulpes/trial').mkdir(parents=True)
(root/'opt/vulpes/trial/session.py').touch()
native.prepare(root, network=False, hardware=False)
fonts = root/'usr/share/fonts/opentype/vulpes'
config = root/'fonts.conf'
config.write_text('<fontconfig><dir>'+escape(str(fonts))+'</dir><cachedir>'+
escape(str(root/'cache'))+'</cachedir><include>'+
escape(str(root/'etc/fonts/conf.d/00-vulpes-sans.conf'))+
'</include></fontconfig>')
for requested in ('sans-serif', 'FiraSans', 'Fira Sans'):
result = subprocess.check_output(
['fc-match', '-f', '%{family}\n%{file}\n', requested], text=True,
env={**os.environ, 'FONTCONFIG_FILE':str(config), 'LC_ALL':'C'})
family, filename = result.strip().splitlines()
self.assertIn('Fira Sans', family)
self.assertEqual(Path(filename).parent, fonts)
self.assertEqual(Path(filename).name, 'FiraSans-Regular.otf')

58
tundra/tests/test_hardware_status.py Normal file
View file

@ -0,0 +1,58 @@
# SPDX-License-Identifier: MPL-2.0
import importlib.util
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
from types import SimpleNamespace
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('hardware', ROOT/'runtime/hardware-status.py')
hardware = importlib.util.module_from_spec(spec)
spec.loader.exec_module(hardware)
spec = importlib.util.spec_from_file_location('native', ROOT/'runtime/prepare-native.py')
native = importlib.util.module_from_spec(spec)
spec.loader.exec_module(native)
class HardwareTests(unittest.TestCase):
def test_battery_absent_invalid_and_real_charge(self):
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
self.assertIsNone(hardware.battery_state(root))
(root/'capacity').write_text('72')
(root/'status').write_text('Discharging')
self.assertEqual(hardware.battery_state(root),
{'level': .72, 'charging': False, 'status': 'Discharging'})
(root/'status').write_text('Full')
self.assertTrue(hardware.battery_state(root)['charging'])
(root/'capacity').write_text('101')
self.assertIsNone(hardware.battery_state(root))
def test_wifi_reports_missing_and_real_device_states_without_identifiers(self):
for code, state, available in [(20,'unavailable',False), (30,'disconnected',True),
(100,'connected',True), (120,'failed',True)]:
with patch.object(hardware.subprocess, 'run', return_value=SimpleNamespace(returncode=0, stdout=f'{code} (label)')) as run:
self.assertEqual(hardware.wifi_state(), {'available': available, 'state': state})
self.assertEqual(run.call_args.args[0][2], 'GENERAL.STATE')
with patch.object(hardware.subprocess, 'run', side_effect=FileNotFoundError):
self.assertFalse(hardware.wifi_state()['available'])
def test_native_network_configuration_preserves_rescue_and_has_no_connections(self):
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
(root/'opt/vulpes/trial').mkdir(parents=True)
(root/'opt/vulpes/trial/session.py').touch()
(root/'usr/sbin').mkdir(parents=True)
(root/'usr/sbin/NetworkManager').touch()
native.prepare(root)
config = (root/'etc/NetworkManager/NetworkManager.conf').read_text()
self.assertIn('unmanaged-devices=*,except:interface-name:wlan0', config)
self.assertIn('auth-polkit=root-only', config)
self.assertIn('enabled=false', config)
self.assertFalse((root/'etc/NetworkManager/system-connections').exists())
self.assertIn('VULPES_TUNDRA=1', (root/'etc/systemd/system/tundra-display.service').read_text())
if __name__ == '__main__':
unittest.main()

29
tundra/tests/test_helper_dependencies.py Normal file
View file

@ -0,0 +1,29 @@
# SPDX-License-Identifier: MPL-2.0
"""The RAM image must retain PT_INTERP even on a translated build host."""
import os
from pathlib import Path
import stat
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'tools'))
from ramdisk_reference import select_helpers
class HelperDependenciesTests(unittest.TestCase):
def test_french_host_keeps_dynamic_interpreter_and_library(self):
entries = {name: {'mode':stat.S_IFREG|0o755, 'data':name.encode()}
for name in ('bin/busybox','sbin/dropbear','sbin/dmsetup',
'lib/aarch64-linux-gnu/libnss_files.so.2',
'lib/aarch64-linux-gnu/libc.so.6',
'lib/ld-linux-aarch64.so.1')}
def readelf(command, **kwargs):
if Path(command[-1]).read_bytes() != b'sbin/dropbear': return ''
label = 'Requesting program interpreter' if kwargs.get('env',{}).get('LC_ALL') == 'C' else 'Interpréteur de programme requis'
return f'[{label}: /lib/ld-linux-aarch64.so.1]\n(NEEDED) [libc.so.6]\n'
with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, {'LC_ALL':'fr_FR.UTF-8'}), patch('ramdisk_reference.subprocess.check_output', side_effect=readelf):
helpers=select_helpers(entries,Path(temp))
self.assertIn('lib/ld-linux-aarch64.so.1', helpers)
self.assertIn('lib/aarch64-linux-gnu/libc.so.6', helpers)
if __name__=='__main__': unittest.main()

65
tundra/tests/test_native_launch.py Normal file
View file

@ -0,0 +1,65 @@
# SPDX-License-Identifier: MPL-2.0
import importlib.util
import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'tools'))
spec=importlib.util.spec_from_file_location('native_launch',ROOT/'tools/run-sargo-native.py')
launch=importlib.util.module_from_spec(spec);spec.loader.exec_module(launch)
class QualifiedLaunchTests(unittest.TestCase):
def setUp(self):
self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
self.root=Path(self.temp.name);self.folder=self.root/'out/build'
self.folder.mkdir(parents=True)
(self.folder/'proof.json').write_text('{"passed":true}')
(self.folder/'build.json').write_text(json.dumps({'imageSha256':'image-hash'}))
self.report={'passed':True,'flashReady':False,'imageSha256':'image-hash',
'evidenceSha256':{'proof.json':launch.digest(self.folder/'proof.json')}}
self.write_report()
self.root_patch=patch.object(launch,'ROOT',self.root);self.root_patch.start();self.addCleanup(self.root_patch.stop)
def write_report(self):
report=self.folder/'qualification.json';report.write_text(json.dumps(self.report))
(self.root/'out/sargo-native-qualified.json').write_text(json.dumps({
'directory':'out/build','qualificationSha256':launch.digest(report)}))
def test_local_verification_does_not_request_boot(self):
with patch.object(launch.subprocess,'run') as run:
self.assertEqual(launch.qualified_build(),self.folder)
self.assertNotIn('--boot',run.call_args.args[0])
def test_modified_evidence_stops_before_device_tool(self):
(self.folder/'proof.json').write_text('changed')
with patch.object(launch.subprocess,'run') as run:
with self.assertRaisesRegex(ValueError,'evidence changed'):launch.qualified_build()
run.assert_not_called()
def test_failed_qualification_stops_before_device_tool(self):
self.report['passed']=False;self.write_report()
with patch.object(launch.subprocess,'run') as run:
with self.assertRaisesRegex(ValueError,'qualified native'):launch.qualified_build()
run.assert_not_called()
def test_interactive_diagnostics_cannot_be_the_normal_launcher(self):
(self.folder/'build.json').write_text(json.dumps({
'imageSha256':'image-hash', 'interactiveDiagnostics':True}))
with patch.object(launch.subprocess,'run') as run:
with self.assertRaisesRegex(ValueError,'Interactive diagnostic'):
launch.qualified_build()
run.assert_not_called()
def test_pointer_cannot_escape_output_directory(self):
(self.root/'out/sargo-native-qualified.json').write_text(json.dumps({'directory':'../outside'}))
with patch.object(launch.subprocess,'run') as run:
with self.assertRaises(ValueError):launch.qualified_build()
run.assert_not_called()
if __name__=='__main__':unittest.main()

81
tundra/tests/test_persistent_install.py Normal file
View file

@ -0,0 +1,81 @@
# SPDX-License-Identifier: MPL-2.0
import importlib.util
import json
from pathlib import Path
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import patch
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'tools'))
spec=importlib.util.spec_from_file_location('install_sargo',ROOT/'tools/install-sargo.py')
install=importlib.util.module_from_spec(spec);spec.loader.exec_module(install)
class InstallTests(unittest.TestCase):
def setUp(self):
self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
self.folder=Path(self.temp.name)
self.image=self.folder/'boot.img';self.image.write_bytes(b'boot')
self.proof=self.folder/'proof.json';self.proof.write_text('{}')
self.build={'image':'boot.img','imageSha256':install.digest(self.image),
'persistentState':{'uuid':'state'},'watchdogSeconds':None,
'serialSha256':'serial','systemImage':{'imageSha256':'system'}}
self.qualification={'passed':True,'privateInstallReady':True,
'imageSha256':self.build['imageSha256'],
'evidenceSha256':{'proof.json':install.digest(self.proof)},
'persistence':{'passed':True,'distinctBoots':True,'sameGaiaSetting':True,'restored':True}}
self.save()
def save(self):
(self.folder/'build.json').write_text(json.dumps(self.build))
(self.folder/'qualification.json').write_text(json.dumps(self.qualification))
def validate(self):
with patch.object(install,'load',return_value=SimpleNamespace(verify_sources=lambda *args:None)):
return install.checked_build(self.folder)
def test_qualified_image(self):
self.assertEqual(self.validate()[2],self.image)
def test_mutated_image_or_proof_rejected(self):
self.image.write_bytes(b'bad')
with self.assertRaisesRegex(ValueError,'provenance'):self.validate()
self.image.write_bytes(b'boot');self.proof.write_text('bad')
with self.assertRaisesRegex(ValueError,'evidence'):self.validate()
def test_diagnostic_and_timed_images_rejected(self):
self.build['interactiveDiagnostics']=True;self.save()
with self.assertRaises(ValueError):self.validate()
self.build['interactiveDiagnostics']=False;self.build['watchdogSeconds']=600;self.save()
with self.assertRaises(ValueError):self.validate()
def test_fake_persistence_rejected(self):
self.qualification['persistence']['distinctBoots']=False;self.save()
with self.assertRaisesRegex(ValueError,'round trip'):self.validate()
def test_only_boot_a_written(self):
with patch.object(install.subprocess,'run') as run:
install.flash_boot_a('device',self.image)
self.assertEqual(run.call_args.args[0],['fastboot','-s','device','flash','boot_a',str(self.image)])
def test_default_only_reviews(self):
boot=SimpleNamespace(checked_plan=lambda:({'serialSha256':'serial'},None))
with patch.object(sys,'argv',['install','--build',str(self.folder)]), \
patch.object(install,'checked_build',return_value=(self.build,self.qualification,self.image)), \
patch.object(install,'load',return_value=boot),patch.object(install,'flash_boot_a') as flash:
install.main();flash.assert_not_called()
def test_restore_does_not_require_working_candidate(self):
backup=self.folder/'boot_a.img';backup.write_bytes(b'original')
boot=SimpleNamespace(checked_plan=lambda:({'serialSha256':'serial','backupDirectory':str(self.folder)},None),
find_fastboot=lambda *args:'device',check_device=lambda *args:{'product':'sargo'})
with patch.object(sys,'argv',['install','--build',str(self.folder),'--restore']), \
patch.object(install,'load',return_value=boot), \
patch.object(install,'checked_build',side_effect=AssertionError('Candidate cannot gate recovery')), \
patch.object(install,'flash_boot_a') as flash:
install.main();flash.assert_called_once_with('device',backup)
if __name__=='__main__':unittest.main()

52
tundra/tests/test_persistent_state.py Normal file
View file

@ -0,0 +1,52 @@
# SPDX-License-Identifier: MPL-2.0
import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools'))
import persistent_state as state
class PersistentStateTests(unittest.TestCase):
def setUp(self):
self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
self.root=Path(self.temp.name)
self.identifier='2a29bca9-5fcb-48a1-a287-dc975f56b353'
self.image=self.root/(self.identifier+'.ext4')
with self.image.open('wb') as output:output.truncate(512*1024**2)
self.value={'schema':1,'device':'sargo','uuid':self.identifier,
'image':self.image.name,'bytes':512*1024**2,
'imageSha256':'a'*64,'systemImageSha256':'b'*64,
'filesystemCheckPassed':True}
self.path=self.root/'state.json'
def validate(self):
self.path.write_text(json.dumps(self.value))
with patch.object(state,'digest',return_value='a'*64):
return state.validate_manifest(self.path)
def test_valid_prepared_state(self):
self.assertEqual(self.validate()['uuid'],self.identifier)
def test_path_escape_and_shell_input_rejected(self):
for key,value in [('image','../../root.img'),('uuid','$(command)'),('systemImageSha256','; reboot')]:
old=self.value[key];self.value[key]=value
with self.assertRaises(ValueError):self.validate()
self.value[key]=old
def test_size_change_rejected(self):
self.image.write_bytes(b'changed')
with self.assertRaisesRegex(ValueError,'changed'):self.validate()
def test_symlink_rejected(self):
target=self.root/'other';self.image.rename(target);self.image.symlink_to(target)
with self.assertRaisesRegex(ValueError,'changed'):self.validate()
def test_missing_fs_check_rejected(self):
self.value['filesystemCheckPassed']=False
with self.assertRaisesRegex(ValueError,'verification'):self.validate()
if __name__=='__main__':unittest.main()

90
tundra/tests/test_radio_control.py Normal file
View file

@ -0,0 +1,90 @@
import importlib.util
from pathlib import Path
import unittest
from types import SimpleNamespace
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('radio_control', ROOT / 'runtime/radio-control.py')
radio = importlib.util.module_from_spec(spec)
spec.loader.exec_module(radio)
class RadioControlTests(unittest.TestCase):
def test_void_dbus_success_is_not_an_invalid_reply(self):
with patch.object(radio.subprocess, 'run', return_value=SimpleNamespace(
returncode=0, stdout='', stderr='')):
self.assertIsNone(radio.bus('org.ofono', '/ril_0/voicecall01', 'org.ofono.VoiceCall', 'Hangup'))
def test_calls_returns_only_public_call_fields(self):
payload = [[['/ril_0/voicecall01', {
'State': {'type': 's', 'data': 'active'},
'LineIdentification': {'type': 's', 'data': '+12025550123'},
'PrivateModemData': {'type': 's', 'data': 'secret'},
}]]]
with patch.object(radio, 'bus', return_value=payload):
self.assertEqual(radio.request({'version': 1, 'operation': 'calls'}), [
{'path': '/ril_0/voicecall01', 'state': 'active', 'number': '+12025550123'}])
with patch.object(radio, 'bus', return_value=[[]]):
self.assertEqual(radio.request({'version': 1, 'operation': 'calls'}), [])
def test_answer_and_hangup_target_only_the_selected_call(self):
for operation, method in [('answer', 'Answer'), ('hangup', 'Hangup')]:
with patch.object(radio, 'bus', return_value=[]) as bus:
radio.request({'version': 1, 'operation': operation, 'path': '/ril_0/voicecall01'})
bus.assert_called_once_with('org.ofono', '/ril_0/voicecall01', 'org.ofono.VoiceCall', method)
with self.assertRaises(ValueError):
radio.request({'version': 1, 'operation': operation, 'path': '/private'})
def test_modem_rejections_are_not_reported_as_transport_failures(self):
for detail, expected in [
('Call failed: Not registered', 'RADIO_NOT_REGISTERED'),
('org.ofono.Error.InProgress', 'RADIO_BUSY'),
('Call failed: Not available', 'RADIO_NOT_READY'),
('private modem response', 'RADIO_REQUEST_FAILED'),
]:
with self.subTest(detail=detail), patch.object(radio.subprocess, 'run',
return_value=SimpleNamespace(returncode=1, stderr=detail)):
with self.assertRaisesRegex(RuntimeError, '^'+expected+'$'):
radio.bus('org.ofono', '/ril_0', 'org.ofono.Modem', 'GetProperties')
def test_dial_is_allowlisted_and_validated(self):
with patch.object(radio, 'bus', return_value=['/ril_0/voicecall01']) as bus:
result = radio.request({'version': 1, 'operation': 'dial', 'number': '+12025550123'})
self.assertEqual(result['path'], '/ril_0/voicecall01')
bus.assert_called_once_with(
'org.ofono', '/ril_0', 'org.ofono.VoiceCallManager', 'Dial', 'ss', '+12025550123', 'default'
)
def test_sms_does_not_accept_arbitrary_dbus_values(self):
with self.assertRaises(ValueError):
radio.request({'version': 1, 'operation': 'sms', 'number': 'x', 'body': 'hello'})
with self.assertRaises(ValueError):
radio.request({'version': 1, 'operation': 'sms', 'number': '+12025550123', 'body': ''})
def test_sms_refuses_unregistered_network_before_submission(self):
for state in ('searching', 'denied', 'unregistered'):
with self.subTest(state=state), patch.object(radio, 'bus',
return_value=[{'Status': {'data': state}}]) as bus:
with self.assertRaisesRegex(RuntimeError, '^RADIO_NOT_REGISTERED$'):
radio.request({'version': 1, 'operation': 'sms',
'number': '+12025550123', 'body': 'test'})
self.assertEqual(bus.call_count, 1)
def test_modem_queue_is_not_delivery_confirmation(self):
for state in ('registered', 'roaming'):
with self.subTest(state=state), patch.object(radio, 'bus', side_effect=[
[{'Status': {'data': state}}], ['/ril_0/messageabc']]):
result = radio.request({'version': 1, 'operation': 'sms',
'number': '+12025550123', 'body': 'test'})
self.assertEqual(result['state'], 'pending')
self.assertEqual(result['path'], '/ril_0/messageabc')
def test_hangup_only_accepts_owned_voice_paths(self):
with self.assertRaises(ValueError):
radio.request({'version': 1, 'operation': 'hangup', 'path': '/org/freedesktop/secret'})
if __name__ == '__main__':
unittest.main()

131
tundra/tests/test_ramdisk.py Normal file
View file

@ -0,0 +1,131 @@
# SPDX-License-Identifier: MPL-2.0
import importlib.util
from pathlib import Path
import stat
import sys
import unittest
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'tools'))
from ramdisk_reference import read_newc, resolve
from tundra import cpio
spec=importlib.util.spec_from_file_location('hal_policy',ROOT/'runtime/hal_policy.py')
policy=importlib.util.module_from_spec(spec);spec.loader.exec_module(policy)
class BootTests(unittest.TestCase):
def test_cpio_roundtrip_and_absolute_library_symlink(self):
data=cpio([('bin',stat.S_IFDIR|0o755,b'',0,0),('bin/tool',stat.S_IFREG|0o755,b'ELF',0,0),('bin/link',stat.S_IFLNK|0o777,b'/bin/tool',0,0)])
entries=read_newc(data)
self.assertEqual(resolve(entries,'bin/link'),b'ELF')
def test_hardlinked_busybox_payload_is_not_lost(self):
import gzip
data=bytearray(gzip.decompress(cpio([
('bin/busybox',stat.S_IFREG|0o755,b'',0,0),
('bin/last-applet',stat.S_IFREG|0o755,b'ELF payload',0,0)])))
offset=0
for _ in range(2):
fields=[int(data[offset+6+i*8:offset+14+i*8],16) for i in range(13)]
data[offset+6:offset+14]=b'00000007'
data[offset+38:offset+46]=b'00000002'
start=(offset+110+fields[11]+3)&~3
offset=(start+fields[6]+3)&~3
self.assertEqual(resolve(read_newc(bytes(data)),'bin/busybox'),b'ELF payload')
def test_cpio_rejects_invalid_header(self):
with self.assertRaises(ValueError):read_newc(b'x'*111)
def test_hal_keeps_display_but_removes_app_telemetry_and_flash_services(self):
text='''# upstream copyright
service vendor.hwcomposer-2-2 /vendor/bin/hw/composer
class hal animation
onrestart restart surfaceflinger
service vendor.pixelstats_vendor /vendor/bin/pixelstats
class hal
service flash_recovery /system/bin/install-recovery.sh
class main
on boot
start flash_recovery
start vendor.pixelstats_vendor
start vendor.hwcomposer-2-2
mount_all /vendor/etc/fstab.sargo
exec -- /system/bin/sh /vendor/bin/mutate.sh
write /dev/block/by-name/boot_a unsafe
write /sys/class/android_usb/android0/enable 1
write /dev/cpuset/cpus 0-7
'''
output,report=policy.filter_rc(text)
self.assertIn('start vendor.hwcomposer-2-2',output)
self.assertIn('# upstream copyright',output)
self.assertIn('write /dev/cpuset/cpus',output)
for forbidden in ('pixelstats','flash_recovery','mount_all','exec --','/dev/block','android_usb','restart surfaceflinger'):
self.assertNotIn(forbidden,output)
self.assertEqual(report['services'],['vendor.hwcomposer-2-2'])
def test_android_multiline_commands_preserve_their_section(self):
source = 'service zygote /system/bin/app_process \\\n --zygote\non boot\n write /dev/event-log-tags "# owned by logd\n"\n'
output, report = policy.filter_rc(source)
self.assertNotIn('app_process', output)
self.assertIn('write /dev/event-log-tags "# owned by logd\n"', output)
def test_ueventd_permissions_are_not_parsed_as_init_commands(self):
for name in ('ueventd.rc','ueventd.sargo.rc','ueventd.qcom.rc'):
self.assertFalse(policy.is_init_script(Path('/android/vendor')/name))
self.assertTrue(policy.is_init_script(Path('/android/init.sargo.rc')))
def test_trial_service_failure_does_not_request_device_reboot(self):
output,_=policy.filter_rc('service servicemanager /system/bin/servicemanager\n critical\n class core\n')
self.assertNotIn('critical',output)
self.assertIn('class core',output)
def test_display_keeps_its_platform_properties_and_configstore(self):
output,report=policy.filter_rc('on post-fs-data\n load_system_props\nservice vendor.configstore-hal /vendor/bin/hw/android.hardware.configstore@1.1-service\n class hal animation\n user system\n')
self.assertIn('load_system_props',output)
self.assertIn('vendor.configstore-hal',report['services'])
def test_native_config_uses_a_valid_shell_and_denies_block_devices(self):
import tempfile
spec = importlib.util.spec_from_file_location('native_prepare', ROOT/'runtime/prepare-native.py')
native = importlib.util.module_from_spec(spec); spec.loader.exec_module(native)
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root/'opt/vulpes/trial').mkdir(parents=True)
(root/'opt/vulpes/trial/session.py').touch()
native.prepare(root)
shells = (root/'etc/shells').read_text().splitlines()
for line in (root/'etc/passwd').read_text().splitlines():
if line.startswith(('root:', 'droidian:')):
self.assertIn(line.split(':')[-1], shells)
config = (root/'etc/tundra/lxc.conf').read_text()
self.assertIn('lxc.cgroup.devices.deny = b *:* rwm', config)
self.assertIn('lxc.cgroup2.devices.deny = b *:* rwm', config)
self.assertNotIn('host', config)
self.assertIn('lxc.cgroup2.memory.max = 536870912', config)
service=(root/'etc/systemd/system/tundra-display.service').read_text()
self.assertIn('ExecStartPre=+chvt 7', service)
self.assertNotIn('ExecStartPost=', service)
# systemd-user's PAM stack includes these on the reference root.
for name in ('common-account', 'common-session-noninteractive'):
self.assertIn('pam_unix.so', (root/'etc/pam.d'/name).read_text())
self.assertIn('pam_deny.so', (root/'etc/pam.d/other').read_text())
def test_hal_denies_indirect_service_start_and_mount_of_block_device(self):
for line in ('setprop ctl.start zygote','setprop ctl.restart vendor.pixelstats_vendor',
'mount ext4 /dev/mmcblk0p72 /data','mount tmpfs /dev/block/x /data'):
self.assertFalse(policy.command_allowed(line.split()),line)
def test_interactive_boot_checks_are_opt_in(self):
import tempfile
spec = importlib.util.spec_from_file_location('native_prepare', ROOT/'runtime/prepare-native.py')
native = importlib.util.module_from_spec(spec); spec.loader.exec_module(native)
for diagnostics in (False, True):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root/'opt/vulpes/trial').mkdir(parents=True)
(root/'opt/vulpes/trial/session.py').touch()
native.prepare(root, diagnostics=diagnostics)
service = (root/'etc/systemd/system/tundra-display.service').read_text()
self.assertEqual('--diagnostics' in service, diagnostics)
if __name__=='__main__':unittest.main()

99
tundra/tests/test_runtime_reference.py Normal file
View file

@ -0,0 +1,99 @@
# SPDX-License-Identifier: MPL-2.0
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import subprocess
import tarfile
import tempfile
import unittest
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
def load(name, path):
spec = importlib.util.spec_from_file_location(name, ROOT/path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
stage = load('stage_runtime', 'tools/stage-runtime.py')
export = load('export_runtime', 'droidian/export-runtime.py')
display = load('compositor_trial', 'droidian/compositor-trial.py')
class ReferenceTests(unittest.TestCase):
def test_snapshot_excludes_configuration_and_private_trees(self):
for path in ('/etc/shadow', '/home/droidian/.ssh/id_ed25519',
'/var/lib/NetworkManager/secret', '/usr/local/private', '/usr/../etc/shadow'):
self.assertFalse(export.allowed(path), path)
self.assertTrue(export.allowed('/usr/share/doc/phoc/copyright'))
def test_dependency_alternative_and_versioned_virtual_provider(self):
def package(version, depends='', provides=''):
return {'Version': version, 'Architecture': 'arm64', 'Depends': depends,
'Pre-Depends': '', 'Provides': provides}
packages = {'app': package('1', 'missing | interface (>= 2)'),
'old': package('1', provides='interface (= 1)'),
'new': package('3', provides='interface (= 3)')}
def compare(args, **kwargs):
self.assertEqual(args[:2], ['dpkg', '--compare-versions'])
self.assertEqual(args[3:], ['>=', '2'])
return subprocess.CompletedProcess(args, 0 if args[2] == '3' else 1)
with patch.object(export.subprocess, 'run', side_effect=compare):
self.assertEqual(set(export.closure(packages, ['app'])), {'app', 'new'})
def test_unresolved_dependency_is_not_silently_dropped(self):
with self.assertRaisesRegex(ValueError, 'No installed provider'):
export.closure({}, ['missing'])
def test_legacy_lib_files_share_the_runtime_loader_directory(self):
with tempfile.TemporaryDirectory() as temporary:
source = Path(temporary)/'source'; source.mkdir()
data = b'reference library'; name = 'lib/example.so'
with tarfile.open(source/'runtime.tar', 'w') as tar:
info = tarfile.TarInfo(name); info.size = len(data); info.mode = 0o6755
tar.addfile(info, io.BytesIO(data))
manifest = {'archiveSha256': hashlib.sha256((source/'runtime.tar').read_bytes()).hexdigest(),
'files': {'/'+name: {'sha256': hashlib.sha256(data).hexdigest()}},
'packages': {'example': {}}, 'missing': []}
(source/'manifest.json').write_text(json.dumps(manifest))
root = Path(temporary)/'root'; stage.stage(source, root)
self.assertTrue((root/'lib').is_symlink())
self.assertEqual((root/'lib/example.so').read_bytes(), data)
self.assertEqual((root/'lib/example.so').stat().st_mode & 0o6000, 0)
self.assertFalse((root/'etc/machine-id').exists())
def test_archive_paths_cannot_escape_root(self):
for path in ('/usr/bin/x', 'usr/../../etc/shadow', 'etc/shadow'):
with self.assertRaises(ValueError):
stage.destination(path)
def test_gaia_child_drops_all_process_capabilities(self):
path = ROOT/'droidian/session.py'
code = "import runpy; runpy.run_path("+repr(str(path))+")[\"drop_capabilities\"](); print(open('/proc/self/status').read())"
result = subprocess.check_output(['python3', '-c', code], text=True)
values = dict(line.split(':', 1) for line in result.splitlines() if ':' in line)
for key in ('CapInh', 'CapPrm', 'CapEff', 'CapAmb'):
self.assertEqual(int(values[key].strip(), 16), 0, key)
def test_display_restore_continues_after_stop_timeout(self):
calls = []
def command(*args, **kwargs):
calls.append(args)
if args[:2] == ('systemctl', 'stop'):
raise subprocess.TimeoutExpired(args, 45)
return subprocess.CompletedProcess(args, 0, '', '')
with tempfile.TemporaryDirectory() as temporary:
p = Path(temporary)
with patch.object(display, 'SERVICE', p/'service'), \
patch.object(display, 'DROPIN', p/'dropin'), \
patch.object(display, 'CONFIG', p/'config'), \
patch.object(display, 'command', side_effect=command):
report = {}; display.restore(report)
self.assertIn(('systemctl', 'start', 'phosh.service'), calls)
self.assertTrue(report['restoredPhosh'])
self.assertEqual(len(report['cleanupErrors']), 1)

30
tundra/tests/test_slot_finalization.py Normal file
View file

@ -0,0 +1,30 @@
import base64
import importlib.util
from pathlib import Path
import struct
import unittest
spec = importlib.util.spec_from_file_location('slot_finalize', Path(__file__).resolve().parents[1]/'tools/finalize-sargo-slot.py')
slot = importlib.util.module_from_spec(spec)
spec.loader.exec_module(slot)
def snapshot(success=False, corrupt=False, change_b=False):
data=bytearray(256)
struct.pack_into('<Q',data,48,(1<<50)|(slot.SUCCESS if success else 0))
if corrupt:data[32]=1
if change_b:data[128+54]|=0x40
copy={'table':base64.b64encode(data).decode(),'partitions':{
'boot_a':{'index':1,'attributes':struct.unpack_from('<Q',data,48)[0]},
'boot_b':{'index':2,'attributes':0}}}
return {'primary':copy,'secondary':copy}
class SlotFinalizationTests(unittest.TestCase):
def test_accepts_only_success_bit_and_is_idempotent(self):
self.assertEqual(slot.verify_transition(snapshot(),snapshot(True)),['boot_a'])
self.assertEqual(slot.verify_transition(snapshot(True),snapshot(True)),[])
def test_rejects_partition_extent_or_other_slot_change(self):
for after in (snapshot(True,corrupt=True),snapshot(True,change_b=True),snapshot()):
with self.assertRaises(ValueError):slot.verify_transition(snapshot(),after)

43
tundra/tests/test_sms_store.py Normal file
View file

@ -0,0 +1,43 @@
import importlib.util
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
spec=importlib.util.spec_from_file_location('sms_store',Path(__file__).resolve().parents[1]/'runtime/sms-inbox.py')
store=importlib.util.module_from_spec(spec)
spec.loader.exec_module(store)
class SmsStoreTests(unittest.TestCase):
def setUp(self):
self.temp=tempfile.TemporaryDirectory()
self.root=Path(self.temp.name)
transaction=store.transaction
self.override=patch.object(store,'transaction',lambda change,**kw: transaction(change,root=self.root,**kw))
self.override.start()
def tearDown(self):
self.override.stop()
self.temp.cleanup()
def test_terminal_state_survives_message_removal(self):
store.record_outgoing('/ril_0/messageabc','sent')
self.assertEqual(store.outgoing_state('/ril_0/messageabc'),'sent')
store.record_outgoing('/ril_0/messageabc','pending')
self.assertEqual(store.outgoing_state('/ril_0/messageabc'),'sent')
self.assertEqual((self.root/'outgoing.json').stat().st_mode & 0o777,0o600)
def test_outgoing_does_not_overwrite_unacknowledged_incoming(self):
store.transaction(lambda rows:rows.append({'receipt':'receipt'}))
store.record_outgoing('/ril_0/messageabc','failed')
self.assertEqual(store.pending(),[{'receipt':'receipt'}])
store.acknowledge(['receipt'])
self.assertEqual(store.pending(),[])
self.assertEqual(store.outgoing_state('/ril_0/messageabc'),'failed')
def test_latest_terminal_state_replaces_same_path(self):
store.record_outgoing('/ril_0/messageabc','failed')
store.record_outgoing('/ril_0/messageabc','sent')
rows=json.loads((self.root/'outgoing.json').read_text())
self.assertEqual(len(rows),1)
self.assertNotIn('body',rows[0])
self.assertNotIn('number',rows[0])
if __name__=='__main__': unittest.main()

40
tundra/tests/test_ssh_inventory.py Normal file
View file

@ -0,0 +1,40 @@
# SPDX-License-Identifier: MPL-2.0
import importlib.util
from pathlib import Path
import subprocess
import sys
import unittest
from unittest.mock import patch
TOOLS = Path(__file__).resolve().parents[1]/'tools'
sys.path.insert(0, str(TOOLS))
spec = importlib.util.spec_from_file_location('ssh_inventory', TOOLS/'collect-device-ssh.py')
inventory = importlib.util.module_from_spec(spec)
spec.loader.exec_module(inventory)
class SSHInventoryTests(unittest.TestCase):
def test_model_and_compatible_do_not_confuse_sargo_and_bonito(self):
self.assertEqual(inventory.identify('Google Pixel 3a\0', ''), 'sargo')
self.assertEqual(inventory.identify('', 'google,sargo\0qcom,sdm670\0'), 'sargo')
self.assertEqual(inventory.identify('Google Pixel 3a XL', 'google,sargo'), 'wrong-device-pixel-3a-xl')
self.assertEqual(inventory.identify('Qualcomm SDM670', ''), 'unverified')
self.assertEqual(inventory.identify('Google Inc. MSM sdm670 S4 PVT v1.0',
'google,b4s4-sdm670\0qcom,sdm670\0', 'sargo\n'), 'sargo')
self.assertEqual(inventory.identify('Google Pixel 3a', '', 'bonito'), 'wrong-device-pixel-3a-xl')
def test_remote_argument_is_quoted_and_fallback_connection_disabled(self):
base = inventory.ssh_base('/tmp/test-control', '10.15.19.82', 'droidian')
self.assertIn('ProxyCommand=false', base)
result = subprocess.CompletedProcess([], 0, '', '')
with patch.object(inventory.subprocess, 'run', return_value=result) as run:
inventory.capture(base, ['dpkg-query', '-f', '${Version}\n'])
self.assertEqual(run.call_args.args[0][-1], "dpkg-query -f '${Version}\n'")
self.assertNotIn('shell', run.call_args.kwargs)
def test_read_commands_have_no_privileged_or_mutating_entrypoints(self):
allowed = {'id', 'cat', 'uname', 'lsblk', 'ls', 'systemctl', 'dpkg-query', 'getprop'}
for command in inventory.READS.values():
self.assertIn(command[0], allowed)
if command[0] == 'systemctl':
self.assertEqual(command[1], 'list-units')

45
tundra/tests/test_system_image.py Normal file
View file

@ -0,0 +1,45 @@
# SPDX-License-Identifier: MPL-2.0
"""Exercise ownership/format safeguards without writing to a real device."""
import importlib.util
import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('system_image', ROOT/'tools/build-system-image.py')
image = importlib.util.module_from_spec(spec)
spec.loader.exec_module(image)
class SystemImageTests(unittest.TestCase):
def test_pack_refuses_real_ownership_changes_outside_fakeroot(self):
with patch.dict(os.environ, {}, clear=True), patch.object(image.os, 'lchown') as chown:
with self.assertRaisesRegex(RuntimeError, 'fakeroot'):
image.pack(Path('/not-a-real-root'), Path('/not-a-real-image'))
chown.assert_not_called()
def test_pack_keeps_system_owned_by_root_and_session_owned_by_app_user(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)/'root'
for name in ('opt/vulpes/host', 'home/droidian', 'etc'):
(root/name).mkdir(parents=True)
config = root/'etc/os-release'; config.touch()
app = root/'opt/vulpes/host/shell.js'; app.touch()
owners = {}
with patch.dict(os.environ, {'FAKEROOTKEY': 'test'}), \
patch.object(image.os, 'lchown', side_effect=lambda p,u,g: owners.update({p:(u,g)})), \
patch.object(image.subprocess, 'run') as run:
image.pack(root, Path(temporary)/'system.ext4')
self.assertEqual(owners[config], (0,0))
self.assertEqual(owners[app], (32011,32011))
args = run.call_args.args[0]
features = args[args.index('-O')+1]
self.assertTrue(features.startswith('none,'))
self.assertNotIn('orphan_file', features)
self.assertIn('TUNDRA_SYSTEM', args)
if __name__ == '__main__':
unittest.main()

113
tundra/tests/test_temporary_boot.py Normal file
View file

@ -0,0 +1,113 @@
# SPDX-License-Identifier: MPL-2.0
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
TOOLS = Path(__file__).resolve().parents[1]/'tools'
sys.path.insert(0,str(TOOLS))
import boot_backup
spec = importlib.util.spec_from_file_location('temporary_boot',TOOLS/'test-sargo-boot.py')
boot = importlib.util.module_from_spec(spec)
spec.loader.exec_module(boot)
class TemporaryBootTests(unittest.TestCase):
def test_different_phone_rejected_without_fastboot_queries(self):
with patch.object(boot,'getvar') as getvar:
with self.assertRaisesRegex(ValueError,'differs from the inventoried'):
boot.check_device('OTHER',hashlib.sha256(b'EXPECTED').hexdigest())
getvar.assert_not_called()
def test_locked_phone_and_slot_b_rejected(self):
expected = hashlib.sha256(b'SARGO').hexdigest()
for field,value in [('unlocked','no'),('current-slot','b'),('product','bonito'),('partition-size:boot_a','0x1000')]:
values = {'product':'sargo','unlocked':'yes','current-slot':'a','partition-size:boot_a':'0x4000000'}
values[field] = value
with patch.object(boot,'getvar',side_effect=lambda serial,name:values[name]):
with self.assertRaises(ValueError):
boot.check_device('SARGO',expected)
def test_mutating_fastboot_commands_are_not_implemented(self):
with patch.object(boot.subprocess,'run') as run:
for args in [('flash','boot_a','image.img'),('erase','userdata'),('set_active','b'),('flashing','unlock')]:
with self.assertRaises(ValueError):
boot.fastboot('SARGO',*args)
run.assert_not_called()
def test_fastboot_variable_formats(self):
for line in ['(bootloader) current-slot: a','current-slot: a']:
result = subprocess.CompletedProcess([],0,'',line+'\nFinished. Total time: 0.001s\n')
with patch.object(boot,'fastboot',return_value=result):
self.assertEqual(boot.getvar('SARGO','current-slot'),'a')
def test_default_mode_cannot_reboot_or_boot(self):
with patch.object(sys,'argv',['test-sargo-boot.py']), \
patch.object(boot,'checked_plan',return_value=({},Path('image.img'))), \
patch.object(boot,'reboot_from_droidian') as reboot, \
patch.object(boot,'fastboot') as fastboot, \
patch.object(boot,'find_fastboot') as find, \
patch.object(boot.os,'umask'), patch('sys.stdout',new_callable=io.StringIO):
boot.main()
reboot.assert_not_called()
fastboot.assert_not_called()
find.assert_not_called()
def test_collect_resumes_without_reboot_or_fastboot(self):
with tempfile.TemporaryDirectory() as temp:
root=Path(temp);logs=root/'logs/device';logs.mkdir(parents=True)
(root/'VALIDATION.json').write_text(json.dumps({'rawSlotADiagnostic':{}}))
image=root/'image.img';image.write_bytes(b'diagnostic')
sha=hashlib.sha256(image.read_bytes()).hexdigest()
previous={'operation':'fastboot boot','fastbootOutput':'Booting OKAY',
'imageSha256':sha,'serialSha256':'phone'}
(logs/'temporary-boot-original.json').write_text(json.dumps(previous))
plan={'diagnosticImageSha256':sha,'serialSha256':'phone'}
response={'status':'TUNDRA_USB_STATUS protocol=1 readOnly=true\nEND\n',
'version':'Linux version 4.9-124-google-sargo\nEND\n',
'mounts':'proc /proc proc ro 0 0\nEND\n',
'invalid':'ERROR unsupported command\nEND\n'}
with patch.object(sys,'argv',['test-sargo-boot.py','--collect']), \
patch.object(boot,'ROOT',root), \
patch.object(boot,'checked_plan',return_value=(plan,image)), \
patch.object(boot,'find_diagnostic',return_value=Path('/dev/bus/usb/003/011')), \
patch.object(boot,'device_layout',return_value={'interface':0,'in':129,'out':1}), \
patch.object(boot,'exchange',return_value=response), \
patch.object(boot,'reboot_from_droidian') as reboot, \
patch.object(boot,'fastboot') as fastboot, \
patch.object(boot.os,'umask'),patch('sys.stdout',new_callable=io.StringIO):
boot.main()
reboot.assert_not_called()
fastboot.assert_not_called()
reports=[json.loads(p.read_text()) for p in logs.glob('temporary-boot-*.json') if p.name!='temporary-boot-original.json']
self.assertEqual(len(reports),1)
self.assertTrue(reports[0]['passed'])
self.assertEqual(reports[0]['operation'],'usb collect')
self.assertEqual(reports[0]['previousBoot']['report'],'logs/device/temporary-boot-original.json')
validation=json.loads((root/'VALIDATION.json').read_text())
self.assertTrue(validation['rawSlotADiagnostic']['hardwareTested'])
self.assertFalse(validation['sargoHardwareDiagnostic']['gaiaIncluded'])
class RawBackupTests(unittest.TestCase):
def test_corruption_missing_file_and_unexpected_file_rejected(self):
with tempfile.TemporaryDirectory() as temp, patch.object(boot_backup,'SIZES',{'boot_a.img':4}):
folder=Path(temp);part=folder/'boot_a.img';part.write_bytes(b'boot')
m={'device':'sargo','rawPartitions':True,'files':{'boot_a.img':{'bytes':4,'sha256':hashlib.sha256(b'boot').hexdigest()}}}
manifest=folder/'manifest.json';manifest.write_text(json.dumps(m))
self.assertEqual(boot_backup.verify(folder),m)
part.write_bytes(b'bad!')
with self.assertRaisesRegex(ValueError,'checksum'):
boot_backup.verify(folder)
part.unlink()
with self.assertRaisesRegex(ValueError,'file or size'):
boot_backup.verify(folder)
m['files']['../escape']={};manifest.write_text(json.dumps(m))
with self.assertRaisesRegex(ValueError,'exactly'):
boot_backup.verify(folder)

134
tundra/tests/test_tools.py Normal file
View file

@ -0,0 +1,134 @@
# SPDX-License-Identifier: MPL-2.0
import gzip
import importlib.util
import json
from pathlib import Path
import stat
import struct
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
import tundra
spec = importlib.util.spec_from_file_location("collector", tundra.ROOT / "tools/collect-device.py")
collector = importlib.util.module_from_spec(spec)
spec.loader.exec_module(collector)
class PackagingTests(unittest.TestCase):
def test_cpio_roundtrip_and_reproducibility(self):
entries = [("dev", stat.S_IFDIR | 0o755, b"", 0, 0),
("dev/console", stat.S_IFCHR | 0o600, b"", 5, 1),
("init", stat.S_IFREG | 0o755, b"init-payload", 0, 0)]
a = tundra.cpio(entries)
self.assertEqual(a, tundra.cpio(entries))
data = gzip.decompress(a)
cursor = 0
decoded = {}
while True:
self.assertEqual(data[cursor:cursor + 6], b"070701")
fields = [int(data[cursor + 6 + 8 * n:cursor + 14 + 8 * n], 16) for n in range(13)]
cursor += 110
name = data[cursor:cursor + fields[11] - 1].decode()
cursor = (cursor + fields[11] + 3) & ~3
payload = data[cursor:cursor + fields[6]]
cursor = (cursor + fields[6] + 3) & ~3
if name == "TRAILER!!!":
break
decoded[name] = (fields, payload)
self.assertEqual(decoded["init"][1], b"init-payload")
self.assertEqual(decoded["dev/console"][0][9:11], [5, 1])
self.assertEqual(decoded["init"][0][1], stat.S_IFREG | 0o755)
# Independent parser: GNU cpio, never extract a device node onto the host.
p = subprocess.run(["cpio", "-it"], input=data, capture_output=True, check=True)
self.assertIn(b"dev/console", p.stdout)
self.assertIn(b"init", p.stdout)
def test_cpio_rejects_escape(self):
for name in ("/init", "../init", "dev/../../init"):
with self.assertRaises(ValueError):
tundra.cpio([(name, stat.S_IFREG | 0o755, b"", 0, 0)])
def test_manifest_inheritance_and_cycle(self):
with tempfile.TemporaryDirectory() as d:
root = Path(d)
(root / "main.xml").write_text('<manifest><remote name="a" fetch="https://example.org/"/><default remote="a" revision="tag"/><include name="sub.xml"/></manifest>')
(root / "sub.xml").write_text('<manifest><project name="kernel" path="src/kernel"/></manifest>')
self.assertEqual(tundra.flatten_manifest(root, "main.xml"),
[{"path": "src/kernel", "url": "https://example.org/kernel", "revision": "tag"}])
(root / "sub.xml").write_text('<manifest><include name="main.xml"/></manifest>')
with self.assertRaises(ValueError):
tundra.flatten_manifest(root, "main.xml")
def test_sargo_boot_roundtrip(self):
folder = tundra.ROOT / "out/sargo-stage0"
metadata_file = folder / "image.json"
if not metadata_file.exists():
self.skipTest("Run ./tundra prepare-sargo first")
info = json.loads(metadata_file.read_text())
image = folder / info["file"]
self.assertEqual(tundra.digest(image), info["sha256"])
self.assertFalse(info["flashReady"])
with tempfile.TemporaryDirectory() as d:
subprocess.run([sys.executable, str(tundra.checked_source("aosp-mkbootimg") / "unpack_bootimg.py"),
"--boot_img", str(image), "--out", d], check=True, capture_output=True)
self.assertEqual((Path(d) / "ramdisk").read_bytes(), (folder / "initramfs.cpio.gz").read_bytes())
kernel = tundra.ROOT / "downloads/aosp-sargo-kernel" / info["kernelSource"]["commit"] / "Image.lz4-dtb"
self.assertEqual(tundra.digest(Path(d) / "kernel"), tundra.digest(kernel))
self.assertEqual(struct.unpack_from("<H", (folder / "init").read_bytes(), 18)[0], 183)
def test_diagnostic_refuses_host_execution(self):
binary = tundra.ROOT / "out/qemu-x86_64/init"
if not binary.exists():
self.skipTest("Run ./tundra test-vm first")
p = subprocess.run([str(binary)], capture_output=True, timeout=3)
self.assertEqual(p.returncode, 2)
self.assertIn(b"only run as PID 1", p.stdout)
class CollectorTests(unittest.TestCase):
def test_reject_wrong_device(self):
result = lambda x: {"returncode": 0, "stdout": x}
self.assertEqual(collector.identify({"ro.product.device": result("bonito")}), "wrong-device-pixel-3a-xl")
self.assertEqual(collector.identify({"ro.product.device": result("sargo")}), "sargo")
self.assertEqual(collector.identify({"device-model": result("Google Pixel 3a\0")}), "sargo-model-only")
self.assertEqual(collector.identify({"device-model": result("Qualcomm SDM670")}), "unverified")
self.assertEqual(collector.identify({"device-model": {"returncode": 1, "stdout": "Google Pixel 3a"}}), "unverified")
def test_collection_is_read_only_and_selects_transport(self):
commands = []
def capture(command):
commands.append(command)
value = "device" if command[-1] == "get-state" else "sargo" if command[-1] == "ro.product.device" else ""
return {"returncode": 0, "stdout": value, "stderr": ""}
with patch.object(sys, "argv", ["collect", "--serial", "TEST123"]), \
patch.object(collector, "capture", side_effect=capture), \
patch.object(collector, "save") as save, patch.object(collector.os, "umask"):
self.assertEqual(collector.main(), 0)
self.assertTrue(save.call_args.args[1]["readOnly"])
self.assertFalse(save.call_args.args[1]["flashAuthorizedByThisReport"])
for command in commands:
self.assertEqual(command[:3], ["adb", "-s", "TEST123"])
self.assertIn(command[3], ("get-state", "shell"))
if command[3] == "shell":
self.assertIn(command[4], ("getprop", "cat", "uname", "ls"))
self.assertTrue(set(command).isdisjoint({"reboot", "root", "push", "flash", "erase", "mount", "unlock"}))
def test_unauthorized_device_is_not_queried(self):
with patch.object(sys, "argv", ["collect", "--serial", "TEST123"]), \
patch.object(collector, "capture", return_value={"returncode": 1}) as capture, \
patch.object(collector, "save") as save, patch.object(collector.os, "umask"):
with self.assertRaises(SystemExit):
collector.main()
self.assertEqual(capture.call_count, 1)
save.assert_not_called()
if __name__ == "__main__":
unittest.main()

36
tundra/tests/test_usb.py Normal file
View file

@ -0,0 +1,36 @@
# SPDX-License-Identifier: MPL-2.0
import gzip
import json
from pathlib import Path
import struct
import subprocess
import sys
import tempfile
import unittest
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools'))
from tundra import ROOT, checked_source, digest
class USBPackagingTests(unittest.TestCase):
def test_refuses_host_execution(self):
init=ROOT/'out/usb-diagnostic/x86_64/init'
if not init.exists(): self.skipTest('Build USB diagnostic first')
result=subprocess.run([str(init)],capture_output=True,text=True,timeout=5)
self.assertEqual(result.returncode,2)
self.assertIn('must be PID 1',result.stdout)
def test_sargo_usb_independent_unpack(self):
folder=ROOT/'out/usb-diagnostic'
if not (folder/'build.json').exists():self.skipTest('Build USB diagnostic first')
report=json.loads((folder/'build.json').read_text())
image=folder/report['file']
self.assertFalse(report['flashReady'])
self.assertEqual(digest(image),report['imageSha256'])
with tempfile.TemporaryDirectory() as tmp:
result=subprocess.run([sys.executable,str(checked_source('aosp-mkbootimg')/'unpack_bootimg.py'),'--boot_img',str(image),'--out',tmp],check=True,capture_output=True,text=True)
self.assertIn('tundra.usb=1',result.stdout)
ramdisk=(Path(tmp)/'ramdisk').read_bytes()
self.assertEqual(ramdisk,(folder/'aarch64/initramfs.cpio.gz').read_bytes())
elf=subprocess.run(['cpio','-i','--to-stdout','init'],input=gzip.decompress(ramdisk),capture_output=True,check=True).stdout
self.assertEqual(elf[:4],b'\x7fELF')
self.assertEqual(struct.unpack_from('<H',elf,18)[0],183)
self.assertEqual(elf,(folder/'aarch64/init').read_bytes())

64
tundra/tests/test_usb_transport.py Normal file
View file

@ -0,0 +1,64 @@
# SPDX-License-Identifier: MPL-2.0
import ctypes
from pathlib import Path
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools'))
import usb_protocol as usb
# Captured through sysfs from the actual Sargo diagnostic, high-speed config 1.
SARGO = bytes.fromhex('12011002000000406b1d04010904010203010902200001010080fa0904000002ff0000040705810200020007050102000200')
class USBTransportTests(unittest.TestCase):
def test_pixel_remapped_endpoint(self):
self.assertEqual(usb.parse_endpoints(SARGO,1),{'interface':0,'in':0x81,'out':0x01})
def test_original_endpoint_and_different_interface(self):
descriptors=bytearray(SARGO)
descriptors[-5]=2
descriptors[29]=4
self.assertEqual(usb.parse_endpoints(descriptors,1),{'interface':4,'in':0x81,'out':0x02})
def test_invalid_and_ambiguous_descriptors_rejected(self):
both_in=bytearray(SARGO);both_in[-5]=0x82
interrupt=bytearray(SARGO);interrupt[-4]=3
for data in (b'',b'\0\2',SARGO[:-1],both_in,interrupt,SARGO+SARGO[27:]):
with self.subTest(data=data):
with self.assertRaises(ValueError):
usb.parse_endpoints(data,1)
with self.assertRaises(ValueError):
usb.parse_endpoints(SARGO,2)
def test_transfer_uses_discovered_addresses_and_claims_interface(self):
layout={'interface':4,'out':1,'in':0x81}
endpoints=[]
response=b'TUNDRA_USB_STATUS protocol=1 readOnly=true\nEND\n'
def ioctl(fd,code,pointer):
request=ctypes.cast(pointer,ctypes.POINTER(usb.Bulk)).contents
endpoints.append(request.ep)
if len(endpoints)==1:
self.assertEqual(request.ep,1)
self.assertEqual(ctypes.string_at(request.data,request.length),b'status\n')
return request.length
self.assertEqual(request.ep,0x81)
ctypes.memmove(request.data,response,len(response))
return len(response)
with tempfile.TemporaryFile() as stream, \
patch.object(usb,'device_layout',return_value=layout), \
patch.object(usb.Path,'open',return_value=stream), \
patch.object(usb.fcntl,'ioctl') as claim, \
patch.object(usb.ctypes,'CDLL',return_value=SimpleNamespace(ioctl=Mock(side_effect=ioctl))):
self.assertEqual(usb.exchange('/dev/bus/usb/003/011',('status',)),{'status':response.decode()})
self.assertEqual(claim.call_args.args[2],usb.struct.pack('I',4))
self.assertEqual(endpoints,[1,0x81])
def test_invalid_command_rejected_before_device_access(self):
with patch.object(usb,'device_layout') as layout:
with self.assertRaises(ValueError):
usb.exchange('/dev/bus/usb/003/011',('status\nversion',))
layout.assert_not_called()

120
tundra/tests/test_wifi_control.py Normal file
View file

@ -0,0 +1,120 @@
# 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_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_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()