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

Prepare Sargo installations with private per-device workspaces

Browse source 
Bind backups, storage layout, RAM builds, file staging and recovery to each phone. Verify atomic transfers without replacing existing state and record the unresolved Android HAL source provenance.

Validated with 53 native tooling tests, 5 changelog tests and two RAM boots with persistent Gaia state on the development Sargo. No new flash or native binary release.
This commit is contained in:
OverSu 2026-10-04 00:06:13 +02:00
commit d04824ec89
16 changed files with 707 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

10
tundra/tests/test_build_sources.py
View file

@ -1,9 +1,11 @@
# 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'))
@ -32,5 +34,13 @@ class SourceSnapshotTests(unittest.TestCase):
self.build['sources']={'../../outside':'hash'}
with self.assertRaises(ValueError):rescue.verify_sources(self.folder,self.build)
def test_workspace_required_before_any_device_command(self):
(self.folder/'build.json').write_text(json.dumps({'image':'boot.img','deviceWorkspaceSha256':'a'*64}))
with patch.object(sys,'argv',['test-rescue-boot','--build',str(self.folder),'--boot']), \
patch.object(rescue.subprocess,'run') as run:
with self.assertRaisesRegex(ValueError,'requires --device-workspace'):
rescue.main()
run.assert_not_called()
if __name__=='__main__':unittest.main()

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

@ -0,0 +1,113 @@
# SPDX-License-Identifier: MPL-2.0
import hashlib
import importlib.util
import json
from pathlib import Path
import struct
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
import sargo_workspace as workspace
def load(name):
spec = importlib.util.spec_from_file_location(name, TOOLS/(name+'.py'))
value = importlib.util.module_from_spec(spec); spec.loader.exec_module(value)
return value
prepare = load('prepare-sargo-device')
collect = load('collect-sargo-installation')
class WorkspaceTests(unittest.TestCase):
def setUp(self):
temp = tempfile.TemporaryDirectory(); self.addCleanup(temp.cleanup)
self.root = Path(temp.name)
self.source = self.root/'inventory'; (self.source/'backup').mkdir(parents=True)
sizes = {name: 16384 for name in boot_backup.SIZES}
patcher = patch.object(boot_backup, 'SIZES', sizes); patcher.start(); self.addCleanup(patcher.stop)
kernel = b'test kernel'
patcher = patch.object(workspace, 'KERNEL_SHA256', hashlib.sha256(kernel).hexdigest())
patcher.start(); self.addCleanup(patcher.stop)
boot = bytearray(16384); boot[:8] = b'ANDROID!'
struct.pack_into('<10I', boot, 8, len(kernel), 0x8000, 1, 0x1000000, 0, 0, 0x100, 4096, 0, 0)
boot[4096:4096+len(kernel)] = kernel
manifest = {'device': 'sargo', 'rawPartitions': True, 'serialSha256': 'a'*64,
'activeSlot': 'a', 'files': {}}
for name, size in sizes.items():
path = self.source/'backup'/name; path.write_bytes(boot if name=='boot_a.img' else bytes(size))
manifest['files'][name] = {'bytes': size, 'sha256': workspace.digest(path)}
(self.source/'backup/manifest.json').write_text(json.dumps(manifest))
self.layout = {'schema': 2, 'device': '/dev/mmcblk0p72', 'partitionLabel': 'userdata',
'filesystem': 'ext4', 'backingDevice': '259:40',
'table': '0 104448000 linear 259:40 329728', 'size': 53648801280,
'firstMiBSha256': 'b'*64, 'serialSha256': 'a'*64, 'activeSlot': 'a'}
self.save_layout()
def save_layout(self):
(self.source/'layout.json').write_text(json.dumps(self.layout))
def test_private_workspace_is_portable_without_shared_inventory(self):
output = prepare.prepare(self.source, self.root/'workspace')
moved = self.root/'moved'; output.rename(moved)
plan, layout = workspace.checked_workspace(moved)
self.assertEqual(plan['backupDirectory'], str(moved/'backup'))
self.assertEqual(plan['serialSha256'], 'a'*64)
self.assertEqual(layout, self.layout)
def test_different_phone_storage_is_rejected_before_output(self):
self.layout['serialSha256'] = 'c'*64; self.save_layout()
with self.assertRaisesRegex(ValueError, 'same Sargo'):
prepare.prepare(self.source, self.root/'workspace')
self.assertFalse((self.root/'workspace').exists())
def test_other_filesystem_and_slot_are_rejected(self):
for name, value in (('filesystem', 'f2fs'), ('activeSlot', 'b')):
old = self.layout[name]; self.layout[name] = value; self.save_layout()
with self.assertRaises(ValueError):
prepare.prepare(self.source, self.root/'workspace')
self.layout[name] = old
def test_unknown_kernel_is_rejected(self):
with patch.object(workspace, 'KERNEL_SHA256', '0'*64):
with self.assertRaisesRegex(ValueError, 'not been qualified'):
prepare.prepare(self.source, self.root/'workspace')
def test_corruption_and_missing_workspace_record_rejected(self):
output = prepare.prepare(self.source, self.root/'workspace')
(output/'layout.json').write_text('{}')
with self.assertRaisesRegex(ValueError, 'input changed'):
workspace.checked_workspace(output)
def test_existing_and_incomplete_directories_rejected(self):
with self.assertRaisesRegex(ValueError, 'exists'):
prepare.prepare(self.source, self.source)
(self.source/'.incomplete').touch()
with self.assertRaisesRegex(ValueError, 'did not finish'):
prepare.prepare(self.source, self.root/'workspace')
def test_collector_does_not_expose_serial_in_identity(self):
proc = self.root/'proc'; (proc/'device-tree').mkdir(parents=True)
(proc/'device-tree/compatible').write_bytes(b'google,b4s4-sdm670\0')
(proc/'device-tree/model').write_text('Google Inc. MSM sdm670 S4 PVT v1.0\0')
(proc/'cmdline').write_text('androidboot.serialno=PRIVATE androidboot.slot_suffix=_a')
result = collect.identity(proc)
self.assertEqual(result['serialSha256'], hashlib.sha256(b'PRIVATE').hexdigest())
self.assertNotIn('PRIVATE', json.dumps(result))
(proc/'cmdline').write_text('androidboot.serialno=PRIVATE androidboot.slot_suffix=_b')
with self.assertRaises(ValueError):
collect.identity(proc)
(proc/'cmdline').write_text('androidboot.serialno=PRIVATE androidboot.slot_suffix=_a')
(proc/'device-tree/model').write_text('Google Inc. MSM sdm670 B4 PVT v1.0\0')
with self.assertRaisesRegex(ValueError, 'device tree'):
collect.identity(proc)
if __name__ == '__main__':
unittest.main()

10
tundra/tests/test_persistent_install.py
View file

@ -67,6 +67,16 @@ class InstallTests(unittest.TestCase):
patch.object(install,'load',return_value=boot),patch.object(install,'flash_boot_a') as flash:
install.main();flash.assert_not_called()
def test_workspace_build_never_falls_back_to_shared_inventory(self):
self.build['deviceWorkspaceSha256'] = 'a'*64
boot=SimpleNamespace(checked_plan=lambda: self.fail('Shared inventory used'))
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:
with self.assertRaisesRegex(ValueError, 'requires --device-workspace'):
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_backup_plan=lambda:{'serialSha256':'serial','backupDirectory':str(self.folder)},

91
tundra/tests/test_stage_files.py Normal file
View file

@ -0,0 +1,91 @@
# SPDX-License-Identifier: MPL-2.0
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import sys
import tempfile
from types import SimpleNamespace
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('stage_files', TOOLS/'stage-sargo-files.py')
stage = importlib.util.module_from_spec(spec); spec.loader.exec_module(stage)
class ReceiveTests(unittest.TestCase):
def setUp(self):
temp = tempfile.TemporaryDirectory(); self.addCleanup(temp.cleanup)
self.root = Path(temp.name)
paths = {'/proc/device-tree/compatible': b'google,b4s4-sdm670\0',
'/proc/device-tree/model': b'Google Inc. MSM sdm670 S4 PVT v1.0\0',
'/proc/cmdline': b'androidboot.serialno=PHONE androidboot.slot_suffix=_a',
'/sys/class/block/mmcblk0p72/dev': b'259:40', '/dev/mmcblk0p72': bytes(1048576)}
for name, content in paths.items():
target = self.root/name.lstrip('/'); target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(content)
self.mount = self.root/'installed'; self.mount.mkdir()
self.config = {'kind': 'data', 'name': 'new.ext4', 'bytes': 4,
'sha256': hashlib.sha256(b'data').hexdigest(), 'requiredBytes': 4,
'serialSha256': hashlib.sha256(b'PHONE').hexdigest(),
'layout': {'firstMiBSha256': hashlib.sha256(bytes(1048576)).hexdigest(),
'size': 53648801280, 'table': '0 104448000 linear 259:40 329728'},
'mount': str(self.mount), 'checkOnly': False}
def run_receiver(self, data=b'data'):
real_path = Path
def path(name):
name = str(name)
return real_path(self.root/name.lstrip('/')) if name.startswith(('/proc/', '/dev/', '/sys/')) else real_path(name)
def command(args, **kwargs):
if args[0] == 'blockdev': return '53648801280'
if args[0] == 'findmnt': return json.dumps({'filesystems': [{'fstype': 'ext4', 'source': '/dev/mapper/test'}]})
if args[0] == 'dmsetup': return '0 104448000 linear 259:40 329728'
raise AssertionError('Unexpected command: '+repr(args))
with patch('pathlib.Path', side_effect=path), patch('stat.S_ISBLK', return_value=True), \
patch('subprocess.check_output', side_effect=command), \
patch('os.statvfs', return_value=SimpleNamespace(f_bavail=10*1024**3, f_frsize=1)), \
patch.object(sys, 'argv', ['receiver', json.dumps(self.config)]), \
patch.object(sys, 'stdin', SimpleNamespace(buffer=io.BytesIO(data))), \
patch.object(sys, 'stdout', io.StringIO()), patch('os.umask'):
exec(compile(stage.RECEIVE, 'receiver', 'exec'), {})
def test_matching_phone_receives_fresh_file(self):
self.run_receiver()
self.assertEqual((self.mount/'tundra/data/new.ext4').read_bytes(), b'data')
def test_existing_data_never_overwritten(self):
target = self.mount/'tundra/data/new.ext4'; target.parent.mkdir(parents=True)
target.write_bytes(b'personal data')
with self.assertRaisesRegex(ValueError, 'Existing data retained'):
self.run_receiver()
self.assertEqual(target.read_bytes(), b'personal data')
def test_short_and_oversized_transfers_leave_no_file(self):
for data in (b'x', b'too long'):
with self.assertRaises(ValueError): self.run_receiver(data)
self.assertEqual(list((self.mount/'tundra/data').iterdir()), [])
def test_other_phone_refused_before_writing(self):
self.config['serialSha256'] = 'a'*64
with self.assertRaisesRegex(ValueError, 'Wrong phone'):
self.run_receiver()
self.assertFalse((self.mount/'tundra').exists())
def test_storage_identity_and_geometry_changes_refused(self):
for key, value in (('firstMiBSha256', 'a'*64), ('table', '0 104448000 linear 259:40 4096')):
old = self.config['layout'][key]; self.config['layout'][key] = value
with self.assertRaises(ValueError):self.run_receiver()
self.config['layout'][key] = old
self.assertFalse((self.mount/'tundra').exists())
def test_preflight_does_not_create_directories(self):
self.config['checkOnly'] = True
with self.assertRaises(SystemExit):self.run_receiver()
self.assertFalse((self.mount/'tundra').exists())
if __name__ == '__main__':unittest.main()