diff --git a/launch_chffrplus.sh b/launch_chffrplus.sh index 8844dd53d..6bea0c8ac 100755 --- a/launch_chffrplus.sh +++ b/launch_chffrplus.sh @@ -29,20 +29,6 @@ function agnos_init { # set success flag for current boot slot sudo abctl --set_success - # Restore SSH access after a user-triggered reset if keys were backed up to /cache. - SSH_BACKUP_DIR="/cache/reset_backup" - if [ -d "$SSH_BACKUP_DIR" ]; then - sudo mkdir -p /data/params/d - for key in GithubSshKeys SshEnabled; do - if [ -f "$SSH_BACKUP_DIR/$key" ]; then - sudo cp "$SSH_BACKUP_DIR/$key" "/data/params/d/$key" - fi - done - sudo chown comma:comma /data/params/d/GithubSshKeys /data/params/d/SshEnabled 2>/dev/null || true - sudo chmod 600 /data/params/d/GithubSshKeys /data/params/d/SshEnabled 2>/dev/null || true - sudo rm -rf "$SSH_BACKUP_DIR" - fi - # TODO: do this without udev in AGNOS # udev does this, but sometimes we startup faster sudo chgrp gpu /dev/adsprpc-smd /dev/ion /dev/kgsl-3d0 diff --git a/launch_env.sh b/launch_env.sh index 5693bbf91..bd150d77a 100755 --- a/launch_env.sh +++ b/launch_env.sh @@ -21,7 +21,7 @@ fi export QCOM_PRIORITY=12 if [ -z "$AGNOS_VERSION" ]; then - export AGNOS_VERSION="19.6.2" + export AGNOS_VERSION="19.6.3" fi if [ -z "$AGNOS_ACCEPTED_VERSIONS" ]; then diff --git a/starpilot/system/the_galaxy/factory_reset.py b/starpilot/system/the_galaxy/factory_reset.py index d807d6e8c..fd7a4c587 100644 --- a/starpilot/system/the_galaxy/factory_reset.py +++ b/starpilot/system/the_galaxy/factory_reset.py @@ -1,6 +1,8 @@ import subprocess import time +from openpilot.system.hardware import HARDWARE + _DELETE_TIMEOUT_S = 1800 _DELETE_RETRY_ATTEMPTS = 20 @@ -8,6 +10,11 @@ _DELETE_RETRY_DELAY_S = 0.25 _DIRECTORY_NOT_EMPTY_ERROR = "directory not empty" +def request_system_factory_reset(): + """Request AGNOS's stock userdata reset and reboot flow.""" + HARDWARE.uninstall() + + def remove_path(path): # Managed processes can recreate entries under /data/params while rm is finishing. for attempt in range(_DELETE_RETRY_ATTEMPTS): diff --git a/starpilot/system/the_galaxy/tests/test_factory_reset.py b/starpilot/system/the_galaxy/tests/test_factory_reset.py index c5e61eb4d..ff37c480b 100644 --- a/starpilot/system/the_galaxy/tests/test_factory_reset.py +++ b/starpilot/system/the_galaxy/tests/test_factory_reset.py @@ -16,6 +16,24 @@ def _load_factory_reset_module(): factory_reset = _load_factory_reset_module() +def test_request_system_factory_reset_uses_agnos_uninstall(monkeypatch): + calls = [] + monkeypatch.setattr(factory_reset.HARDWARE, "uninstall", lambda: calls.append(True)) + + factory_reset.request_system_factory_reset() + + assert calls == [True] + + +def test_galaxy_factory_reset_worker_does_not_partially_delete_userdata(): + source = (Path(__file__).resolve().parents[1] / "the_galaxy.py").read_text(encoding="utf-8") + worker = source.split("def _factory_reset_worker():", 1)[1].split("\ndef ", 1)[0] + + assert "_request_system_factory_reset()" in worker + assert "_run_factory_reset_delete" not in worker + assert "_finish_update_and_reboot" not in worker + + def test_remove_path_retries_directory_not_empty(monkeypatch): results = iter( [ diff --git a/starpilot/system/the_galaxy/the_galaxy.py b/starpilot/system/the_galaxy/the_galaxy.py index 15c6eaea6..80af72d0a 100644 --- a/starpilot/system/the_galaxy/the_galaxy.py +++ b/starpilot/system/the_galaxy/the_galaxy.py @@ -98,7 +98,10 @@ from openpilot.starpilot.common.testing_grounds import ( TESTING_GROUNDS_STATE_PATH as SHARED_TESTING_GROUNDS_STATE_PATH, ) from openpilot.starpilot.navigation.destination_store import normalize_destination_payload, update_recent_destinations -from openpilot.starpilot.system.the_galaxy.factory_reset import remove_path as _run_factory_reset_delete +from openpilot.starpilot.system.the_galaxy.factory_reset import ( + remove_path as _run_factory_reset_delete, + request_system_factory_reset as _request_system_factory_reset, +) from openpilot.starpilot.system.the_galaxy import flm_workspace, utilities from openpilot.starpilot.system.the_galaxy.update_recovery import inspect_interrupted_update, public_recovery_status, recover_interrupted_update @@ -1437,22 +1440,6 @@ _fast_update_state = { } _ROUTE_DELETE_LOCK = threading.Lock() -_FACTORY_RESET_WIPE_PATHS = [ - "/data/params", - "/cache/starpilot/params", - "/cache/params", - "/data/media/0/realdata", - "/data/media/0/realdata_HD", - "/data/media/0/realdata_konik", - "/data/models", - "/data/toggle_backups", - "/data/backups", - "/data/themes", - "/data/media/0/osm/offline", - "/cache/use_HD", - "/cache/use_konik", -] - _PLOTS_POLL_INTERVAL_S = 0.75 _PLOTS_BOOT_STABILIZATION_WINDOW_S = 45.0 _PLOTS_BOOT_POLL_INTERVAL_S = 1.0 @@ -2543,34 +2530,18 @@ def _factory_reset_worker(): started_at = time.time() try: - _set_fast_update_progress(1, "Preparing factory reset", 10.0, "Cleaning up legacy device state...") + _set_fast_update_progress(1, "Preparing factory reset", 10.0, "Requesting AGNOS system reset...") _set_fast_update_state( running=True, stage="factory-resetting", - message="Factory reset started. Wiping device state...", + message="Factory reset requested. AGNOS will erase userdata after reboot...", lastError="", lastMode="factory-reset", startedAt=started_at, finishedAt=0.0, ) - _set_fast_update_progress(1, "Preparing factory reset", 100.0, "Factory reset initialized.") - - total_paths = max(1, len(_FACTORY_RESET_WIPE_PATHS)) - for index, path in enumerate(_FACTORY_RESET_WIPE_PATHS, start=1): - step_percent = ((index - 1) / total_paths) * 100.0 - _set_fast_update_progress(2, "Wiping device data", step_percent, f"Removing {path}...") - _run_factory_reset_delete(path) - _set_fast_update_progress(2, "Wiping device data", (index / total_paths) * 100.0, f"Removed {path}.") - - _set_fast_update_progress(3, "Resetting factory state", 100.0, "Legacy device state removed.") - - _set_fast_update_progress(4, "Finalizing reset", 50.0, "Syncing filesystem before reboot...") - subprocess.run(["sync"], capture_output=True, text=True, timeout=60, check=False) - _set_fast_update_progress(4, "Finalizing reset", 100.0, "Filesystem sync complete.") - - _finish_update_and_reboot( - "Factory reset complete. Device is rebooting now. Please wait for reconnection." - ) + _set_fast_update_progress(1, "Preparing factory reset", 100.0, "AGNOS reset trigger initialized.") + _request_system_factory_reset() except Exception as exception: _set_fast_update_error_state("Factory reset failed.", exception) diff --git a/system/hardware/tici/agnos.json b/system/hardware/tici/agnos.json index 3e2c65740..d6da5576d 100644 --- a/system/hardware/tici/agnos.json +++ b/system/hardware/tici/agnos.json @@ -67,13 +67,13 @@ }, { "name": "system", - "url": "https://www.dropbox.com/scl/fi/pewhzpqzi3aewuiaffc6m/system10.img.xz?rlkey=olzrzulhs93zzghnjrskmdwxt&st=exnfk2oz&dl=1", - "hash": "ab395d4c963a908ab86709f1a6580a62dd24cb34ee71cf5fd5ec29d7d48d0e10", - "hash_raw": "ab395d4c963a908ab86709f1a6580a62dd24cb34ee71cf5fd5ec29d7d48d0e10", + "url": "https://cloud.floonetwork.io/s/w4tQdX5kS9Rgeiw/download", + "hash": "8478e080cf5fbfdcc84d6910f7cd1cb6b2a5a2939eb9e56f2425f87ee33d146d", + "hash_raw": "8478e080cf5fbfdcc84d6910f7cd1cb6b2a5a2939eb9e56f2425f87ee33d146d", "size": 4718592000, "sparse": false, "full_check": false, "has_ab": true, - "ondevice_hash": "ab395d4c963a908ab86709f1a6580a62dd24cb34ee71cf5fd5ec29d7d48d0e10" + "ondevice_hash": "8478e080cf5fbfdcc84d6910f7cd1cb6b2a5a2939eb9e56f2425f87ee33d146d" } ] diff --git a/tools/agnos/patch_system_reset_image.py b/tools/agnos/patch_system_reset_image.py index 9b9f1c595..b88f90fac 100644 --- a/tools/agnos/patch_system_reset_image.py +++ b/tools/agnos/patch_system_reset_image.py @@ -1,50 +1,36 @@ #!/usr/bin/env python3 +"""Build StarPilot's AGNOS system image from an official upstream image. + +The system payload must remain upstream-identical except for /VERSION. C3 +compatibility stays outside this system image in the manifest's known-good C3 +bootloader/boot partitions and in StarPilot's runtime. Reset, setup, networking, +and updater behavior remain owned by upstream AGNOS. +""" + import argparse import hashlib import json import os -import re import shutil -import subprocess import struct -import tempfile +import subprocess import urllib.request -import zipfile -from io import BytesIO from pathlib import Path -RESET_PATH_IN_IMAGE = "/usr/comma/reset" -COMMA_SH_PATH_IN_IMAGE = "/usr/comma/comma.sh" -MAGIC_PATH_IN_IMAGE = "/usr/comma/magic.py" -SETUP_PATH_IN_IMAGE = "/usr/comma/setup" -UPDATER_PATH_IN_IMAGE = "/usr/comma/updater" -BG_PATH_IN_IMAGE = "/usr/comma/bg.jpg" -WESTON_SERVICE_PATH_IN_IMAGE = "/lib/systemd/system/weston.service" -RESET_ENTRY_IN_ZIPAPP = "openpilot/system/ui/reset.py" -MICI_RESET_ENTRY_IN_ZIPAPP = "openpilot/system/ui/mici_reset.py" -TICI_RESET_ENTRY_IN_ZIPAPP = "openpilot/system/ui/tici_reset.py" -APPLICATION_ENTRY_IN_ZIPAPP = "openpilot/system/ui/lib/application.py" -WIFI_MANAGER_ENTRY_IN_SETUP_ZIPAPP = "openpilot/system/ui/lib/wifi_manager.py" -SETUP_ENTRY_IN_SETUP_ZIPAPP = "openpilot/system/ui/setup.py" -TICI_SETUP_ENTRY_IN_SETUP_ZIPAPP = "openpilot/system/ui/tici_setup.py" -MICI_SETUP_ENTRY_IN_SETUP_ZIPAPP = "openpilot/system/ui/mici_setup.py" -UPDATER_ENTRY_IN_ZIPAPP = "openpilot/system/ui/updater.py" VERSION_PATH_IN_IMAGE = "/VERSION" -PYTHON_SITE_PACKAGES_PATH_IN_IMAGE = "/usr/local/venv/lib/python3.12/site-packages" -AMDGPU_FIRMWARE_PATH_IN_IMAGE = "/lib/firmware/amdgpu" -PATCH_MARKER = "STARPILOT_C4_RESET_LAYOUT_V1" -APP_PATCH_MARKER = "STARPILOT_C4_RESET_APP_DIMENSIONS_V1" -SETUP_WIFI_PATCH_MARKER = "JEEPNY_AVAILABLE = True" -SETUP_BRANDING_PATCH_MARKER = "STARPILOT_SETUP_BRANDING_V1" -SETUP_SSH_RESTORE_PATCH_MARKER = "STARPILOT_SETUP_SSH_RESTORE_V1" -WESTON_BG_PATCH_MARKER = "STARPILOT_WESTON_BG_ORIENTATION_V2" -COMMA_SH_DISPLAY_WAIT_PATCH_MARKER = "STARPILOT_DISPLAY_READY_WAIT_V1" -JEEPNY_VERSION = "0.9.0" -JEEPNY_WHEEL_URL = "https://files.pythonhosted.org/packages/b2/a3/e137168c9c44d18eff0376253da9f1e9234d0239e0ee230d2fee6cea8e55/jeepney-0.9.0-py3-none-any.whl" -JEEPNY_WHEEL_SHA256 = "97e5714520c16fc0a45695e5365a2e11b81ea79bba796e26f9f1d178cb182683" -JEEPNY_PACKAGE_DIR = "jeepney" -JEEPNY_DIST_INFO_DIR = f"jeepney-{JEEPNY_VERSION}.dist-info" +ALLOWED_IMAGE_MUTATIONS = frozenset({VERSION_PATH_IN_IMAGE}) + +UPSTREAM_PROTECTED_PAYLOADS = ( + "/usr/comma/comma.sh", + "/usr/comma/reset", + "/usr/comma/setup", + "/usr/comma/updater", + "/etc/NetworkManager/NetworkManager.conf", + "/etc/NetworkManager/conf.d/10-globally-managed-devices.conf", + "/lib/systemd/system/NetworkManager.service", +) + ANDROID_SPARSE_MAGIC = 0xED26FF3A CHUNK_TYPE_RAW = 0xCAC1 CHUNK_TYPE_FILL = 0xCAC2 @@ -52,1312 +38,419 @@ CHUNK_TYPE_DONT_CARE = 0xCAC3 CHUNK_TYPE_CRC32 = 0xCAC4 XZ_MAGIC = b"\xFD7zXZ\x00" -AMDGPU_FIRMWARE_SHA256 = { - "gc_12_0_0_imu.bin.zst": "aa15e5b3156bffc45e0c50bccbcd364fbd3f958531b695b7487a803d780b8328", - "gc_12_0_0_me.bin.zst": "d7eba5197f2580f32b8256b1d9cb68e723e9e644293a34446a7913e3c093cba5", - "gc_12_0_0_mec.bin.zst": "1931593440b8f9423580d9e2cdc5b34e7c682cdffe1ca4b74b0c2f6a0420236d", - "gc_12_0_0_pfp.bin.zst": "16bfd64c10fe73b5e760055069a60e5841dba16c0ed4edb56c20d675e23901f6", - "gc_12_0_0_rlc.bin.zst": "6436b582734a413456fff3d3c7195e71cc9e78a7ed31ee21c83ffd6fae1ad186", - "psp_14_0_2_sos.bin.zst": "7b538448b57d4f9dd06b2eea90d4f86a16e65e3027cdecee8db71c2c5f1fa243", - "sdma_7_0_0.bin.zst": "beaafb53993a106edd392392d5896245ae2a957c6d0f495d0002eec72ad8ad38", - "smu_14_0_2.bin.zst": "6951995d1d606f4dc60c895f19d34ed18aa40e62129f83d8510c45e8aa9ae2fc", -} - -DEFAULT_SYNC_COMMA_FILES = [ - "/usr/comma/bg.jpg", - "/usr/comma/comma.sh", - "/usr/comma/debug.py", - "/usr/comma/fs_setup.sh", - "/usr/comma/installer", - "/usr/comma/magic.py", - "/usr/comma/power_drop_monitor.py", - "/usr/comma/power_monitor.py", - "/usr/comma/reset", - "/usr/comma/screen_calibration.py", - "/usr/comma/setup", - "/usr/comma/setup_keys", - "/usr/comma/updater", -] - -INODE_MODE_TYPE_PREFIX = { - "regular": "100", - "directory": "040", - "symlink": "120", - "character": "20", - "block": "60", - "fifo": "10", - "socket": "140", -} - -def patch_application_script(original: bytes) -> bytes: - if APP_PATCH_MARKER.encode("utf-8") in original: - return original - - text = original.decode("utf-8", "replace") - replacement = ( - f"# {APP_PATCH_MARKER}\n" - "_dt = HARDWARE.get_device_type()\n" - "if _dt in ('tici', 'tizi'):\n" - " gui_app = GuiApplication(2160, 1080)\n" - "else:\n" - " gui_app = GuiApplication(536, 240)\n" - ) - - fixed = text.replace("gui_app = GuiApplication(2160, 1080)", replacement) - if fixed == text: - fixed = re.sub( - r"^gui_app\s*=\s*GuiApplication\([^\n]+\)\s*$", - replacement.rstrip("\n"), - text, - count=1, - flags=re.MULTILINE, - ) - - if fixed == text: - # Newer upstream application.py is already device-aware via GuiApplication defaults. - # Keep it as-is and stamp a marker so verification can still pass. - if text.startswith("#!"): - first_nl = text.find("\n") - if first_nl != -1: - fixed = text[:first_nl + 1] + f"# {APP_PATCH_MARKER} (no-op)\n" + text[first_nl + 1:] - else: - fixed = text + f"\n# {APP_PATCH_MARKER} (no-op)\n" - else: - fixed = f"# {APP_PATCH_MARKER} (no-op)\n" + text - - return fixed.encode("utf-8") - - -def patch_setup_wifi_manager() -> bytes: - """ - Replace setup zipapp's wifi_manager.py with repo version that gracefully handles - missing jeepney (fallback to nmcli/fake), preventing setup boot-logo hangs. - """ - repo_root = Path(__file__).resolve().parents[2] - src = repo_root / "system/ui/lib/wifi_manager.py" - if not src.is_file(): - raise RuntimeError(f"Unable to find repo wifi_manager source: {src}") - data = src.read_bytes() - if SETUP_WIFI_PATCH_MARKER.encode("utf-8") not in data: - raise RuntimeError("Repo wifi_manager.py does not appear to include jeepney fallback marker") - return data - - -def patched_weston_bg_python() -> str: - som_id_path = "/sys/devices/platform/vendor/vendor:gpio-som-id/som_id" - return ( - "from PIL import Image; " - f"som=open(\"{som_id_path}\").read().strip(); " - "img=Image.open(\"/usr/comma/bg.jpg\").convert(\"RGB\"); " - "img=img.rotate(180) if som == \"1\" else img; " - "mask=img.convert(\"L\").point(lambda p: 255 if p > 16 else 0); " - "bbox=mask.getbbox(); " - "logo=img.crop(bbox) if bbox else img; " - # Pillow positive degrees are counter-clockwise; -90 pre-rotates the source clockwise. - "logo=logo.rotate(-90, expand=True); " - "resample=Image.Resampling.LANCZOS if hasattr(Image, \"Resampling\") else Image.LANCZOS; " - "logo=logo.resize((max(1, logo.width//3), max(1, logo.height//3)), resample); " - "canvas=Image.new(\"RGB\", img.size, (0, 0, 0)); " - "canvas.paste(logo, ((img.width - logo.width)//2, (img.height - logo.height)//2)); " - "canvas.save(\"/tmp/bg.jpg\")" - ) - - -def patched_weston_bg_exec_line() -> str: - python_cmd = patched_weston_bg_python().replace('"', '\\"') - return f"ExecStartPre=/bin/bash -c \"/usr/local/venv/bin/python -c '{python_cmd}'\"" - - -def patch_comma_sh_display_wait(original: bytes) -> bytes: - text = original.decode("utf-8") - if COMMA_SH_DISPLAY_WAIT_PATCH_MARKER in text: - return original - - old = """echo "waiting for magic" -for i in {1..200}; do - if systemctl is-active --quiet magic && [ -S /tmp/drmfd.sock ]; then - break - fi - sleep 0.1 -done - -if systemctl is-active --quiet magic && [ -S /tmp/drmfd.sock ]; then - echo "magic ready after ${SECONDS}s" -else - echo "timed out waiting for magic, ${SECONDS}s" -fi -""" - new = f"""# {COMMA_SH_DISPLAY_WAIT_PATCH_MARKER} -if systemctl cat magic.service >/dev/null 2>&1; then - echo "waiting for magic" - for i in {{1..200}}; do - if systemctl is-active --quiet magic && [ -S /tmp/drmfd.sock ]; then - break - fi - sleep 0.1 - done - - if systemctl is-active --quiet magic && [ -S /tmp/drmfd.sock ]; then - echo "magic ready after ${{SECONDS}}s" - else - echo "timed out waiting for magic, ${{SECONDS}}s" - fi -else - echo "magic unavailable; waiting for weston" - for i in {{1..200}}; do - if systemctl is-active --quiet weston-ready && [ -S /var/tmp/weston/wayland-0 ]; then - break - fi - sleep 0.1 - done - - if systemctl is-active --quiet weston-ready && [ -S /var/tmp/weston/wayland-0 ]; then - echo "weston ready after ${{SECONDS}}s" - else - echo "timed out waiting for weston, ${{SECONDS}}s" - fi -fi -""" - - if old not in text: - raise RuntimeError("Unable to find comma.sh display readiness wait") - return text.replace(old, new, 1).encode("utf-8") - - -def patch_weston_service(original: bytes) -> bytes: - text = original.decode("utf-8") - if WESTON_BG_PATCH_MARKER in text: - return original - - old = ( - "ExecStartPre=/bin/bash -c \"/usr/local/venv/bin/python -c 'from PIL import Image; " - "img=Image.open(\\\"/usr/comma/bg.jpg\\\"); " - "(img.rotate(180) if open(\\\"/sys/devices/platform/vendor/vendor:gpio-som-id/som_id\\\").read().strip() == \\\"1\\\" else img).save(\\\"/tmp/bg.jpg\\\")'\"" - ) - new = ( - f"# {WESTON_BG_PATCH_MARKER}: displayed boot logo was 90 degrees counter-clockwise.\n" - f"{patched_weston_bg_exec_line()}" - ) - - if old not in text: - raise RuntimeError("Unable to find weston.service background image generation line") - return text.replace(old, new, 1).encode("utf-8") - - -def patch_setup_branding_script(original: bytes, entry_name: str) -> bytes: - text = original.decode("utf-8") - if SETUP_BRANDING_PATCH_MARKER in text: - return text.encode("utf-8") - - text = text.replace( - 'OPENPILOT_URL = "https://openpilot.comma.ai"', - 'NETWORK_CHECK_URL = "https://openpilot.comma.ai"\n' - 'DEFAULT_INSTALLER_URL = "https://installer.comma.ai/firestar5683/StarPilot"\n' - f'# {SETUP_BRANDING_PATCH_MARKER}', - ) - text = text.replace("urllib.request.Request(OPENPILOT_URL, method=\"HEAD\")", - "urllib.request.Request(NETWORK_CHECK_URL, method=\"HEAD\")") - text = text.replace("urllib.request.urlopen(OPENPILOT_URL, timeout=2)", - "urllib.request.urlopen(NETWORK_CHECK_URL, timeout=2)") - text = text.replace("self.download(OPENPILOT_URL)", "self.download(DEFAULT_INSTALLER_URL)") - - if entry_name == MICI_SETUP_ENTRY_IN_SETUP_ZIPAPP: - text = text.replace('LargerSlider("slide to use\\nopenpilot"', 'LargerSlider("slide to use\\nstarpilot"') - text = text.replace('LargerSlider("slide to install\\nopenpilot"', 'LargerSlider("slide to install\\nstarpilot"') - text = text.replace('BigPillButton("install openpilot"', 'BigPillButton("install StarPilot"') - text = text.replace('set_text("install openpilot"', 'set_text("install StarPilot"') - elif entry_name == TICI_SETUP_ENTRY_IN_SETUP_ZIPAPP: - text = text.replace('ButtonRadio("openpilot"', 'ButtonRadio("StarPilot"') - - if SETUP_BRANDING_PATCH_MARKER not in text: - raise RuntimeError(f"Failed to patch setup branding for {entry_name}") - - return text.encode("utf-8") - - -def patch_setup_module(relative_path: str) -> bytes: - """ - Replace setup zipapp module with repo version so setup behavior stays in sync. - """ - repo_root = Path(__file__).resolve().parents[2] - src = repo_root / relative_path - if not src.is_file(): - raise RuntimeError(f"Unable to find repo setup source: {src}") - return src.read_bytes() - - -def patch_setup_script_with_ssh_restore(relative_path: str) -> bytes: - """ - Apply SSH-key restore logic directly into setup scripts for AGNOS images. - This keeps repo runtime behavior unchanged while making image reset flows - resilient to setup/install failures. - """ - text = patch_setup_module(relative_path).decode("utf-8") - if SETUP_SSH_RESTORE_PATCH_MARKER in text: - return text.encode("utf-8") - - restore_block = f""" -# {SETUP_SSH_RESTORE_PATCH_MARKER} -def _restore_ssh_after_reset(): - backup_dir = "/cache/reset_backup" - params_dir = "/data/params/d" - if not os.path.isdir(backup_dir): - return - - restored = False - try: - os.makedirs(params_dir, exist_ok=True) - for key in ("GithubSshKeys", "SshEnabled"): - src = f"{{backup_dir}}/{{key}}" - dst = f"{{params_dir}}/{{key}}" - if not os.path.isfile(src): - continue - shutil.copyfile(src, dst) - os.chmod(dst, 0o600) - restored = True - - if restored: - os.system("sudo chown -R comma:comma /data/params >/dev/null 2>&1 || true") - os.system("sudo /usr/comma/set_ssh.sh >/tmp/setup_ssh_restore.log 2>&1 || true") - finally: - os.system(f"sudo rm -rf {{backup_dir}} >/dev/null 2>&1 || true") -""" - - if "def main():" not in text: - raise RuntimeError(f"Unable to patch setup script without main(): {relative_path}") - - text = text.replace("\ndef main():", f"{restore_block}\n\ndef main():", 1) - text = text.replace(" try:\n gui_app.init_window(", " try:\n _restore_ssh_after_reset()\n gui_app.init_window(", 1) - return text.encode("utf-8") - - -def get_setup_replacements() -> dict[str, bytes]: - """ - Keep the reference setup bundle intact and patch only the networking backend. - - The reference AGNOS setup bundle already contains the correct small-screen - selector and matching mici UI modules. Replacing those modules with repo-head - versions caused bootstrap incompatibilities. The only setup-side changes we - still need are the jeepney fallback plus the StarPilot branding/url strings. - """ - return { - WIFI_MANAGER_ENTRY_IN_SETUP_ZIPAPP: patch_setup_wifi_manager(), - } - - -def patch_updater_module() -> bytes: - """ - Replace only the bundled updater selector with the repo version. - - The selector itself carries the small-screen fallback logic; the rest of the - reference updater zipapp stays unchanged. - """ - repo_root = Path(__file__).resolve().parents[2] - src = repo_root / "system/ui/updater.py" - if not src.is_file(): - raise RuntimeError(f"Unable to find repo updater source: {src}") - return src.read_bytes() - - -def patch_reset_script() -> bytes: - """ - Use repo reset.py so AGNOS reset stays in sync with upstream selector logic. - """ - repo_root = Path(__file__).resolve().parents[2] - src = repo_root / "system/ui/reset.py" - if not src.is_file(): - raise RuntimeError(f"Unable to find repo reset source: {src}") - data = src.read_text(encoding="utf-8") - if PATCH_MARKER not in data: - if data.startswith("#!"): - first_nl = data.find("\n") - if first_nl != -1: - data = data[:first_nl + 1] + f"# {PATCH_MARKER}\n" + data[first_nl + 1:] - else: - data = data + f"\n# {PATCH_MARKER}\n" - else: - data = f"# {PATCH_MARKER}\n" + data - return data.encode("utf-8") - def parse_args() -> argparse.Namespace: - p = argparse.ArgumentParser(description="Patch AGNOS system image with StarPilot reset and hardware support") - p.add_argument("--manifest", default="system/hardware/tici/agnos.json", help="Path to AGNOS manifest JSON") - p.add_argument("--work-dir", default=".cache/agnos_reset_patch", help="Working directory") - p.add_argument("--source-url", default=None, help="Override source raw system image URL") - p.add_argument("--source-image", default=None, help="Use existing local raw system image file instead of download") - p.add_argument("--reference-manifest", default=None, help="Optional AGNOS manifest used to source /usr/comma installer payloads") - p.add_argument("--reference-source-url", default=None, help="Override reference AGNOS system image URL for /usr/comma file sync") - p.add_argument("--reference-image", default=None, help="Use existing local reference system image file for /usr/comma file sync") - p.add_argument("--sync-comma-files", default=",".join(DEFAULT_SYNC_COMMA_FILES), - help="Comma-separated file list to sync from reference image (e.g. /usr/comma/installer,/usr/comma/setup)") - p.add_argument("--disable-comma-file-sync", action="store_true", - help="Disable syncing /usr/comma files from a reference image") - p.add_argument("--disable-usbgpu-firmware", action="store_true", - help="Do not install the AMD firmware required by the external GPU") - p.add_argument("--output-xz", default=None, help="Output .img.xz path") - p.add_argument("--new-url", default=None, help="Hosted URL for patched image; used for manifest output") - p.add_argument("--manifest-out", default=None, help="Write updated manifest JSON here") - p.add_argument("--in-place-manifest", action="store_true", help="Update manifest file in place") - p.add_argument("--force-download", action="store_true", help="Force redownload source image") - p.add_argument("--set-version", default=None, help="Override /VERSION inside patched image (e.g. 12.8.1)") - return p.parse_args() + parser = argparse.ArgumentParser( + description="Create a stock AGNOS system image with only StarPilot's version marker changed", + ) + parser.add_argument("--manifest", default="system/hardware/tici/agnos.json", + help="StarPilot manifest to update after the image is hosted") + parser.add_argument("--upstream-manifest", default=None, + help="Official openpilot AGNOS manifest used as the system-image source") + parser.add_argument("--source-url", default=None, + help="Override the official system image URL") + parser.add_argument("--source-image", default=None, + help="Use an already-downloaded official system image") + parser.add_argument("--expected-source-version", default=None, + help="Fail unless the official image has this /VERSION") + parser.add_argument("--set-version", required=True, + help="StarPilot AGNOS version written to /VERSION") + parser.add_argument("--work-dir", default=".cache/agnos_stock_c3", + help="Directory for downloaded and generated images") + parser.add_argument("--output-xz", default=None, help="Output .img.xz path") + parser.add_argument("--new-url", default=None, + help="Hosted output URL; enables manifest generation") + parser.add_argument("--manifest-out", default=None, + help="Write the updated StarPilot manifest here") + parser.add_argument("--in-place-manifest", action="store_true", + help="Update --manifest in place") + parser.add_argument("--force-download", action="store_true") + return parser.parse_args() def find_debugfs() -> str: - candidates = [ + candidates = ( os.environ.get("DEBUGFS"), "debugfs", "/opt/homebrew/opt/e2fsprogs/sbin/debugfs", - ] - for c in candidates: - if c and shutil.which(c): - return c - if c and Path(c).is_file(): - return c + ) + for candidate in candidates: + if candidate and (shutil.which(candidate) or Path(candidate).is_file()): + return candidate raise RuntimeError("debugfs not found. Install e2fsprogs and retry.") def load_manifest(path: Path) -> list[dict]: - return json.loads(path.read_text()) + return json.loads(path.read_text(encoding="utf-8")) def get_system_entry(manifest: list[dict]) -> dict: - for e in manifest: - if e.get("name") == "system": - return e + for entry in manifest: + if entry.get("name") == "system": + return entry raise RuntimeError("No system entry found in manifest") +def find_default_upstream_manifest(primary_manifest_path: Path) -> Path | None: + repo_root = Path(__file__).resolve().parents[2] + candidates = ( + repo_root.parent / "openpilot/openpilot/common/hardware/comma/agnos.json", + repo_root.parent / "openpilot/openpilot/system/hardware/comma/agnos.json", + repo_root.parent / "openpilot/openpilot/system/hardware/tici/agnos.json", + repo_root.parent / "openpilot/system/hardware/tici/agnos.json", + ) + primary = primary_manifest_path.resolve() + for candidate in candidates: + if candidate.is_file() and candidate.resolve() != primary: + return candidate.resolve() + return None + + +def resolve_upstream_manifest(explicit_path: str | None, primary_manifest_path: Path) -> Path: + upstream = Path(explicit_path).resolve() if explicit_path else find_default_upstream_manifest(primary_manifest_path) + if upstream is None or not upstream.is_file(): + raise RuntimeError("Official upstream AGNOS manifest not found; pass --upstream-manifest") + if upstream.resolve() == primary_manifest_path.resolve(): + raise RuntimeError("Refusing to use StarPilot's manifest as the stock source") + return upstream + + def pick_source_url(system_entry: dict, override: str | None) -> str: if override: return override url = system_entry.get("url") - if isinstance(url, str): - return url - alt = system_entry.get("alt") - if isinstance(alt, dict) and isinstance(alt.get("url"), str): - return alt["url"] - raise RuntimeError("No source URL found for system image") + if not isinstance(url, str) or not url: + raise RuntimeError("Official system manifest entry has no URL") + return url -def find_default_reference_manifest(primary_manifest_path: Path) -> Path | None: - # Expected tree layout for local development: - # /starpilot/system/hardware/tici/agnos.json - # /openpilot/system/hardware/tici/agnos.json - repo_root = primary_manifest_path - for _ in range(4): - if repo_root.parent == repo_root: - break - repo_root = repo_root.parent - - candidates = [ - repo_root.parent / "openpilot/openpilot/system/hardware/tici/agnos.json", - repo_root.parent / "openpilot/system/hardware/tici/agnos.json", - repo_root / "openpilot/system/hardware/tici/agnos.json", - ] - - for candidate in candidates: - if candidate.is_file() and candidate.resolve() != primary_manifest_path.resolve(): - return candidate.resolve() - return None +def expected_upstream_version(starpilot_version: str, explicit_version: str | None) -> str: + if explicit_version: + return explicit_version.strip() + parts = starpilot_version.strip().split(".") + if len(parts) < 3: + raise RuntimeError("Pass --expected-source-version when --set-version has no StarPilot revision suffix") + return ".".join(parts[:-1]) -def parse_sync_file_list(raw: str) -> list[str]: - out: list[str] = [] - seen: set[str] = set() - for token in raw.replace(";", ",").split(","): - item = token.strip() - if not item: - continue - if not item.startswith("/"): - if "/" in item: - item = f"/{item.lstrip('/')}" - else: - item = f"/usr/comma/{item}" - if item not in seen: - seen.add(item) - out.append(item) - return out +def download(url: str, destination: Path) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + partial = destination.with_suffix(destination.suffix + ".part") + print(f"Downloading official AGNOS system image: {url}", flush=True) + with urllib.request.urlopen(url) as response, open(partial, "wb") as output: + shutil.copyfileobj(response, output, length=8 * 1024 * 1024) + partial.replace(destination) -def download(url: str, dst: Path) -> None: - dst.parent.mkdir(parents=True, exist_ok=True) - tmp = dst.with_suffix(dst.suffix + ".part") - print(f"Downloading {url} -> {dst}", flush=True) - with urllib.request.urlopen(url) as src, open(tmp, "wb") as out: - shutil.copyfileobj(src, out, length=1024 * 1024) - tmp.replace(dst) - - -def download_with_sha256(url: str, dst: Path, expected_sha256: str) -> None: - if not dst.exists(): - download(url, dst) - - actual_sha256 = sha256_file(dst) - if actual_sha256 != expected_sha256: - dst.unlink(missing_ok=True) - download(url, dst) - actual_sha256 = sha256_file(dst) - - if actual_sha256 != expected_sha256: - raise RuntimeError(f"Downloaded file hash mismatch for {dst}: got {actual_sha256}, expected {expected_sha256}") - - -def run_cmd(cmd: list[str], check: bool = True, capture: bool = True) -> subprocess.CompletedProcess[str]: - proc = subprocess.run(cmd, text=True, capture_output=capture) - if check and proc.returncode != 0: - raise RuntimeError(f"Command failed ({proc.returncode}): {' '.join(cmd)}\n{proc.stdout}\n{proc.stderr}") - return proc +def run_cmd(cmd: list[str], capture: bool = True) -> subprocess.CompletedProcess[str]: + result = subprocess.run(cmd, check=False, capture_output=capture, text=True) + if result.returncode != 0: + output = f"{result.stdout}\n{result.stderr}" if capture else "" + raise RuntimeError(f"Command failed ({result.returncode}): {' '.join(cmd)}\n{output}") + return result def is_xz_file(path: Path) -> bool: - with open(path, "rb") as f: - header = f.read(len(XZ_MAGIC)) - return header == XZ_MAGIC + with open(path, "rb") as stream: + return stream.read(len(XZ_MAGIC)) == XZ_MAGIC -def decompress_xz(src: Path, dst: Path) -> None: - dst.parent.mkdir(parents=True, exist_ok=True) - tmp = dst.with_suffix(dst.suffix + ".part") - print(f"Decompressing XZ image {src} -> {dst}", flush=True) - with open(tmp, "wb") as out: - proc = subprocess.run(["xz", "-d", "-c", str(src)], stdout=out, stderr=subprocess.PIPE, text=True) - if proc.returncode != 0: - tmp.unlink(missing_ok=True) - raise RuntimeError(f"xz decompression failed:\n{proc.stderr}") - tmp.replace(dst) +def decompress_xz(source: Path, destination: Path) -> None: + partial = destination.with_suffix(destination.suffix + ".part") + print(f"Decompressing {source}", flush=True) + with open(partial, "wb") as output: + result = subprocess.run(["xz", "-T0", "-dc", str(source)], stdout=output, stderr=subprocess.PIPE) + if result.returncode != 0: + partial.unlink(missing_ok=True) + raise RuntimeError(f"xz failed: {result.stderr.decode('utf-8', 'replace')}") + partial.replace(destination) def is_android_sparse(path: Path) -> bool: - with open(path, "rb") as f: - header = f.read(4) - if len(header) != 4: - return False - return int.from_bytes(header, "little") == ANDROID_SPARSE_MAGIC + with open(path, "rb") as stream: + raw = stream.read(4) + return len(raw) == 4 and struct.unpack(" None: - print(f"Unsparsing Android image {src_sparse} -> {dst_raw}", flush=True) - with open(src_sparse, "rb") as f_in, open(dst_raw, "wb") as f_out: - file_hdr = f_in.read(28) - if len(file_hdr) != 28: - raise RuntimeError("Invalid sparse image header length") - magic, major, minor, file_hdr_sz, chunk_hdr_sz, blk_sz, total_blks, total_chunks, _checksum = struct.unpack(" 28: - f_in.read(file_hdr_sz - 28) - if chunk_hdr_sz < 12: - raise RuntimeError(f"Invalid chunk header size: {chunk_hdr_sz}") +def unsparse_image(source: Path, destination: Path) -> None: + print(f"Converting Android sparse image {source}", flush=True) + with open(source, "rb") as source_file, open(destination, "wb") as output: + header = source_file.read(28) + if len(header) != 28: + raise RuntimeError("Sparse image header is truncated") + magic, major, _minor, file_header_size, chunk_header_size, block_size, total_blocks, total_chunks, _checksum = struct.unpack( + " 28: + source_file.read(file_header_size - 28) for _ in range(total_chunks): - chunk_hdr = f_in.read(12) - if len(chunk_hdr) != 12: - raise RuntimeError("Unexpected EOF in chunk header") - chunk_type, _reserved, chunk_sz, total_sz = struct.unpack("<2H2I", chunk_hdr) - if chunk_hdr_sz > 12: - f_in.read(chunk_hdr_sz - 12) - - data_sz = total_sz - chunk_hdr_sz - out_chunk_bytes = chunk_sz * blk_sz + chunk_header = source_file.read(chunk_header_size) + if len(chunk_header) != chunk_header_size: + raise RuntimeError("Sparse chunk header is truncated") + chunk_type, _reserved, chunk_blocks, total_size = struct.unpack("<2H2I", chunk_header[:12]) + payload_size = total_size - chunk_header_size + output_size = chunk_blocks * block_size if chunk_type == CHUNK_TYPE_RAW: - if data_sz != out_chunk_bytes: - raise RuntimeError(f"RAW chunk size mismatch: data={data_sz} out={out_chunk_bytes}") - remaining = data_sz - while remaining > 0: - chunk = f_in.read(min(8 * 1024 * 1024, remaining)) + if payload_size != output_size: + raise RuntimeError("Sparse RAW chunk size mismatch") + remaining = payload_size + while remaining: + chunk = source_file.read(min(8 * 1024 * 1024, remaining)) if not chunk: - raise RuntimeError("Unexpected EOF in RAW chunk") - f_out.write(chunk) + raise RuntimeError("Sparse RAW chunk is truncated") + output.write(chunk) remaining -= len(chunk) elif chunk_type == CHUNK_TYPE_FILL: - if data_sz != 4: - raise RuntimeError(f"FILL chunk expected 4 bytes, got {data_sz}") - pattern = f_in.read(4) - if len(pattern) != 4: - raise RuntimeError("Unexpected EOF in FILL chunk") - # Write as sparse hole if fill is zero for speed. - if pattern == b"\x00\x00\x00\x00": - f_out.seek(out_chunk_bytes, os.SEEK_CUR) + if payload_size != 4: + raise RuntimeError("Sparse FILL chunk has invalid size") + pattern = source_file.read(4) + if pattern == b"\0\0\0\0": + output.seek(output_size, os.SEEK_CUR) else: - unit = pattern * (blk_sz // 4) - for _ in range(chunk_sz): - f_out.write(unit) + unit = pattern * (block_size // 4) + for _ in range(chunk_blocks): + output.write(unit) elif chunk_type == CHUNK_TYPE_DONT_CARE: - if data_sz > 0: - f_in.read(data_sz) - f_out.seek(out_chunk_bytes, os.SEEK_CUR) + if payload_size: + source_file.read(payload_size) + output.seek(output_size, os.SEEK_CUR) elif chunk_type == CHUNK_TYPE_CRC32: - if data_sz != 4: - raise RuntimeError(f"CRC32 chunk expected 4 bytes, got {data_sz}") - f_in.read(4) + if payload_size != 4: + raise RuntimeError("Sparse CRC32 chunk has invalid size") + source_file.read(4) else: raise RuntimeError(f"Unknown sparse chunk type: 0x{chunk_type:04x}") - f_out.truncate(total_blks * blk_sz) + output.truncate(total_blocks * block_size) -def materialize_ext4_image(source_img: Path, raw_img: Path, work_dir: Path, label: str, force: bool = False) -> None: - source_for_sparse = source_img - - if is_xz_file(source_img): - decompressed = work_dir / f"{label}.decompressed.img" - if force and decompressed.exists(): - decompressed.unlink() +def materialize_ext4_image(source: Path, destination: Path, work_dir: Path, force: bool = False) -> None: + candidate = source + if is_xz_file(source): + decompressed = work_dir / "official_system.decompressed.img" + if force: + decompressed.unlink(missing_ok=True) if not decompressed.exists(): - decompress_xz(source_img, decompressed) - source_for_sparse = decompressed + decompress_xz(source, decompressed) + candidate = decompressed - if force and raw_img.exists(): - raw_img.unlink() - - if raw_img.exists(): + if force: + destination.unlink(missing_ok=True) + if destination.exists(): return - - if is_android_sparse(source_for_sparse): - unsparse_image(source_for_sparse, raw_img) + if is_android_sparse(candidate): + unsparse_image(candidate, destination) else: - shutil.copy2(source_for_sparse, raw_img) + shutil.copy2(candidate, destination) def run_debugfs(debugfs: str, image: Path, request: str, write: bool = False) -> str: - cmd = [debugfs] + command = [debugfs] if write: - cmd.append("-w") - cmd += ["-R", request, str(image)] - proc = run_cmd(cmd, check=True, capture=True) - return f"{proc.stdout}\n{proc.stderr}" - - -def split_shebang(data: bytes) -> tuple[bytes, bytes]: - if data.startswith(b"#!"): - idx = data.find(b"\n") - if idx != -1: - return data[:idx + 1], data[idx + 1:] - return b"", data - - -def patch_reset_zipapp(original: bytes) -> bytes: - shebang, zip_payload = split_shebang(original) - - src_io = BytesIO(zip_payload) - dst_io = BytesIO() - changed = False - - replacement_reset = patch_reset_script() - reset_replacements = { - RESET_ENTRY_IN_ZIPAPP: replacement_reset, - } - - with zipfile.ZipFile(src_io, "r") as src, zipfile.ZipFile(dst_io, "w", compression=zipfile.ZIP_DEFLATED) as dst: - if APPLICATION_ENTRY_IN_ZIPAPP not in src.namelist(): - raise RuntimeError(f"{APPLICATION_ENTRY_IN_ZIPAPP} not found in reset zipapp") - - seen_entries: set[str] = set() - for info in src.infolist(): - seen_entries.add(info.filename) - payload = src.read(info.filename) - if info.filename in reset_replacements: - replacement_payload = reset_replacements[info.filename] - if payload != replacement_payload: - payload = replacement_payload - changed = True - elif info.filename == APPLICATION_ENTRY_IN_ZIPAPP: - patched_payload = patch_application_script(payload) - if patched_payload != payload: - payload = patched_payload - changed = True - - new_info = zipfile.ZipInfo(info.filename, info.date_time) - new_info.compress_type = zipfile.ZIP_DEFLATED - new_info.external_attr = info.external_attr - new_info.create_system = info.create_system - dst.writestr(new_info, payload) - - # Some reference images may not include reset.py; add missing entry explicitly. - default_external_attr = 0o100644 << 16 - for entry, payload in reset_replacements.items(): - if entry in seen_entries: - continue - new_info = zipfile.ZipInfo(entry) - new_info.compress_type = zipfile.ZIP_DEFLATED - new_info.external_attr = default_external_attr - new_info.create_system = 3 - dst.writestr(new_info, payload) - changed = True - - if not changed: - return original - return shebang + dst_io.getvalue() - - -def patch_updater_zipapp(original: bytes) -> bytes: - shebang, zip_payload = split_shebang(original) - - replacement = patch_updater_module() - src_io = BytesIO(zip_payload) - dst_io = BytesIO() - changed = False - found_updater = False - - with zipfile.ZipFile(src_io, "r") as src, zipfile.ZipFile(dst_io, "w", compression=zipfile.ZIP_DEFLATED) as dst: - for info in src.infolist(): - payload = src.read(info.filename) - if info.filename == UPDATER_ENTRY_IN_ZIPAPP: - found_updater = True - if payload != replacement: - payload = replacement - changed = True - - new_info = zipfile.ZipInfo(info.filename, info.date_time) - new_info.compress_type = zipfile.ZIP_DEFLATED - new_info.external_attr = info.external_attr - new_info.create_system = info.create_system - dst.writestr(new_info, payload) - - if not found_updater: - new_info = zipfile.ZipInfo(UPDATER_ENTRY_IN_ZIPAPP) - new_info.compress_type = zipfile.ZIP_DEFLATED - new_info.external_attr = 0o100644 << 16 - new_info.create_system = 3 - dst.writestr(new_info, replacement) - changed = True - - if not changed: - return original - return shebang + dst_io.getvalue() - - -def patch_setup_zipapp(original: bytes) -> bytes: - shebang, zip_payload = split_shebang(original) - - src_io = BytesIO(zip_payload) - dst_io = BytesIO() - changed = False - - replacements = get_setup_replacements() - - with zipfile.ZipFile(src_io, "r") as src, zipfile.ZipFile(dst_io, "w", compression=zipfile.ZIP_DEFLATED) as dst: - seen_entries: set[str] = set() - for info in src.infolist(): - seen_entries.add(info.filename) - payload = src.read(info.filename) - if info.filename in replacements and payload != replacements[info.filename]: - payload = replacements[info.filename] - changed = True - elif info.filename in (MICI_SETUP_ENTRY_IN_SETUP_ZIPAPP, TICI_SETUP_ENTRY_IN_SETUP_ZIPAPP): - patched_payload = patch_setup_branding_script(payload, info.filename) - if patched_payload != payload: - payload = patched_payload - changed = True - - new_info = zipfile.ZipInfo(info.filename, info.date_time) - new_info.compress_type = zipfile.ZIP_DEFLATED - new_info.external_attr = info.external_attr - new_info.create_system = info.create_system - dst.writestr(new_info, payload) - - default_external_attr = 0o100644 << 16 - for entry, payload in replacements.items(): - if entry in seen_entries: - continue - new_info = zipfile.ZipInfo(entry) - new_info.compress_type = zipfile.ZIP_DEFLATED - new_info.external_attr = default_external_attr - new_info.create_system = 3 - dst.writestr(new_info, payload) - changed = True - - if not changed: - return original - return shebang + dst_io.getvalue() - - -def zipapp_has_markers(data: bytes) -> bool: - _shebang, zip_payload = split_shebang(data) - with zipfile.ZipFile(BytesIO(zip_payload), "r") as z: - reset_script = z.read(RESET_ENTRY_IN_ZIPAPP) - tici_reset_script = z.read(TICI_RESET_ENTRY_IN_ZIPAPP) - mici_reset_script = z.read(MICI_RESET_ENTRY_IN_ZIPAPP) - app_script = z.read(APPLICATION_ENTRY_IN_ZIPAPP) - return ( - PATCH_MARKER.encode() in reset_script - and b"_device_tree_device_type" in reset_script - and b"gui_app.big_ui()" not in reset_script - and b"mici_setup" not in mici_reset_script - and b"jeepney" not in mici_reset_script - and b"mici_setup" not in tici_reset_script - and b"jeepney" not in tici_reset_script - and APP_PATCH_MARKER.encode() in app_script - ) - - -def setup_zipapp_has_expected_content(data: bytes) -> bool: - _shebang, zip_payload = split_shebang(data) - replacements = get_setup_replacements() - with zipfile.ZipFile(BytesIO(zip_payload), "r") as z: - try: - wifi_manager = z.read(WIFI_MANAGER_ENTRY_IN_SETUP_ZIPAPP) - if SETUP_WIFI_PATCH_MARKER.encode() not in wifi_manager: - return False - for entry, payload in replacements.items(): - if z.read(entry) != payload: - return False - for entry in (MICI_SETUP_ENTRY_IN_SETUP_ZIPAPP, TICI_SETUP_ENTRY_IN_SETUP_ZIPAPP): - setup_script = z.read(entry) - if SETUP_BRANDING_PATCH_MARKER.encode() not in setup_script: - return False - if b"installer.comma.ai/firestar5683/StarPilot" not in setup_script: - return False - mici_setup = z.read(MICI_SETUP_ENTRY_IN_SETUP_ZIPAPP) - if b"install openpilot" in mici_setup or b"slide to install\\nopenpilot" in mici_setup: - return False - except KeyError: - return False - return True - - -def updater_zipapp_has_expected_content(data: bytes) -> bool: - _shebang, zip_payload = split_shebang(data) - with zipfile.ZipFile(BytesIO(zip_payload), "r") as z: - try: - updater_script = z.read(UPDATER_ENTRY_IN_ZIPAPP) - except KeyError: - return False - return updater_script == patch_updater_module() - - -def weston_service_has_expected_content(data: bytes) -> bool: - return ( - WESTON_BG_PATCH_MARKER.encode("utf-8") in data - and b"displayed boot logo was 90 degrees counter-clockwise" in data - and b"logo=img.crop(bbox) if bbox else img" in data - and b"logo=logo.rotate(-90, expand=True)" in data - and b"logo=logo.resize((max(1, logo.width//3), max(1, logo.height//3)), resample)" in data - and b"canvas.save(\\\"/tmp/bg.jpg\\\")" in data - ) - - -def comma_sh_has_expected_display_wait(data: bytes) -> bool: - return ( - COMMA_SH_DISPLAY_WAIT_PATCH_MARKER.encode("utf-8") in data - and b"systemctl cat magic.service" in data - and b"systemctl is-active --quiet weston-ready" in data - and b"[ -S /var/tmp/weston/wayland-0 ]" in data - ) + command.append("-w") + command += ["-R", request, str(image)] + result = run_cmd(command) + return f"{result.stdout}\n{result.stderr}" def parse_inode(debugfs_output: str) -> int: - m = re.search(r"Inode:\s+(\d+)", debugfs_output) - if not m: - raise RuntimeError(f"Unable to parse inode from debugfs stat output:\n{debugfs_output}") - return int(m.group(1)) + for line in debugfs_output.splitlines(): + if "Inode:" in line: + value = line.split("Inode:", 1)[1].strip().split()[0] + return int(value) + raise RuntimeError(f"Unable to parse inode from debugfs output:\n{debugfs_output}") -def format_debugfs_mode(mode_octal: str) -> str: - try: - mode = int(mode_octal, 8) - except ValueError as e: - raise RuntimeError(f"Invalid octal inode mode: {mode_octal}") from e - if not 0 <= mode <= 0xFFFF: - raise RuntimeError(f"Inode mode exceeds ext4 field width: {mode_octal}") - return f"0{mode:o}" +def write_allowed_file(debugfs: str, image: Path, image_path: str, local_file: Path, + mode_octal: str = "100644", uid: int = 0, gid: int = 0) -> None: + if image_path not in ALLOWED_IMAGE_MUTATIONS: + raise RuntimeError(f"Refusing non-C3 AGNOS system mutation: {image_path}") - -def verify_inode_metadata(debugfs: str, image: Path, image_path: str, expected_type: str, - mode_octal: str, uid: int, gid: int) -> None: - stat_out = run_debugfs(debugfs, image, f"stat {image_path}", write=False) - file_type, perms_octal, actual_uid, actual_gid = parse_debugfs_stat(stat_out) - expected_perms = int(mode_octal, 8) & 0o7777 - actual_perms = int(perms_octal, 8) - if (file_type, actual_perms, actual_uid, actual_gid) != (expected_type, expected_perms, uid, gid): - raise RuntimeError( - f"Metadata verification failed for {image_path}: " - f"got type={file_type} mode={actual_perms:04o} uid={actual_uid} gid={actual_gid}, " - f"expected type={expected_type} mode={expected_perms:04o} uid={uid} gid={gid}" - ) - - -def write_regular_file_to_image(debugfs: str, image: Path, image_path: str, local_file: Path, mode_octal: str, uid: int = 0, gid: int = 0) -> None: try: run_debugfs(debugfs, image, f"rm {image_path}", write=True) - except Exception as e: - err = str(e).lower() - if "file not found" not in err and "no such file" not in err: + except RuntimeError as error: + if "file not found" not in str(error).lower() and "no such file" not in str(error).lower(): raise run_debugfs(debugfs, image, f"write {local_file} {image_path}", write=True) - stat_out = run_debugfs(debugfs, image, f"stat {image_path}", write=False) - inode = parse_inode(stat_out) - run_debugfs(debugfs, image, f"set_inode_field <{inode}> mode {format_debugfs_mode(mode_octal)}", write=True) - run_debugfs(debugfs, image, f"set_inode_field <{inode}> uid {uid}", write=True) - run_debugfs(debugfs, image, f"set_inode_field <{inode}> gid {gid}", write=True) - verify_inode_metadata(debugfs, image, image_path, "regular", mode_octal, uid, gid) - - -def ensure_directory_in_image(debugfs: str, image: Path, image_path: str, mode_octal: str = "040755", uid: int = 0, gid: int = 0) -> None: - try: - run_debugfs(debugfs, image, f"mkdir {image_path}", write=True) - except Exception as e: - err = str(e).lower() - if "already exists" not in err and "file exists" not in err: - raise - - stat_out = run_debugfs(debugfs, image, f"stat {image_path}", write=False) - inode = parse_inode(stat_out) - run_debugfs(debugfs, image, f"set_inode_field <{inode}> mode {format_debugfs_mode(mode_octal)}", write=True) - run_debugfs(debugfs, image, f"set_inode_field <{inode}> uid {uid}", write=True) - run_debugfs(debugfs, image, f"set_inode_field <{inode}> gid {gid}", write=True) - verify_inode_metadata(debugfs, image, image_path, "directory", mode_octal, uid, gid) - - -def extract_wheel_subset(wheel_path: Path, extract_dir: Path, roots: set[str]) -> dict[Path, str]: - if extract_dir.exists(): - shutil.rmtree(extract_dir) - extract_dir.mkdir(parents=True) - file_modes: dict[Path, str] = {} - - with zipfile.ZipFile(wheel_path, "r") as wheel: - for info in wheel.infolist(): - parts = Path(info.filename).parts - if not parts or parts[0] not in roots: - continue - if any(part == ".." for part in parts): - raise RuntimeError(f"Unsafe wheel entry path: {info.filename}") - - local_path = extract_dir.joinpath(*parts) - if info.is_dir(): - local_path.mkdir(parents=True, exist_ok=True) - continue - - local_path.parent.mkdir(parents=True, exist_ok=True) - with wheel.open(info, "r") as src, open(local_path, "wb") as dst: - shutil.copyfileobj(src, dst) - - perms = (info.external_attr >> 16) & 0o777 - if not perms: - perms = 0o644 - os.chmod(local_path, perms) - file_modes[local_path] = f"100{perms:03o}" - - if not (extract_dir / JEEPNY_PACKAGE_DIR / "__init__.py").is_file(): - raise RuntimeError("jeepney wheel extraction did not produce jeepney/__init__.py") - if not (extract_dir / JEEPNY_DIST_INFO_DIR / "METADATA").is_file(): - raise RuntimeError("jeepney wheel extraction did not produce dist-info/METADATA") - - return file_modes - - -def install_python_package_tree(debugfs: str, image: Path, source_dir: Path, image_root: str, file_modes: dict[Path, str]) -> None: - dirs = sorted((p for p in source_dir.rglob("*") if p.is_dir()), key=lambda p: len(p.relative_to(source_dir).parts)) - for local_dir in dirs: - rel = local_dir.relative_to(source_dir).as_posix() - ensure_directory_in_image(debugfs, image, f"{image_root}/{rel}", "040755", 0, 0) - - files = sorted((p for p in source_dir.rglob("*") if p.is_file()), key=lambda p: p.relative_to(source_dir).as_posix()) - for local_file in files: - rel = local_file.relative_to(source_dir).as_posix() - mode_octal = file_modes.get(local_file, "100644") - write_regular_file_to_image(debugfs, image, f"{image_root}/{rel}", local_file, mode_octal, 0, 0) - - -def install_jeepney_into_image(debugfs: str, image: Path, work_dir: Path) -> None: - wheel_dir = work_dir / "python_wheels" - wheel_path = wheel_dir / f"jeepney-{JEEPNY_VERSION}-py3-none-any.whl" - download_with_sha256(JEEPNY_WHEEL_URL, wheel_path, JEEPNY_WHEEL_SHA256) - - extract_dir = work_dir / "jeepney_wheel" - file_modes = extract_wheel_subset(wheel_path, extract_dir, {JEEPNY_PACKAGE_DIR, JEEPNY_DIST_INFO_DIR}) - - print(f"Installing jeepney {JEEPNY_VERSION} into AGNOS Python venv", flush=True) - install_python_package_tree(debugfs, image, extract_dir, PYTHON_SITE_PACKAGES_PATH_IN_IMAGE, file_modes) - - -def image_has_jeepney(debugfs: str, image: Path, work_dir: Path) -> bool: - verify_dir = work_dir / "jeepney_verify" - verify_dir.mkdir(parents=True, exist_ok=True) - init_file = verify_dir / "__init__.py" - wrappers_file = verify_dir / "wrappers.py" - metadata_file = verify_dir / "METADATA" - init_file.unlink(missing_ok=True) - wrappers_file.unlink(missing_ok=True) - metadata_file.unlink(missing_ok=True) - try: - for image_path in ( - f"{PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_PACKAGE_DIR}/__init__.py", - f"{PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_PACKAGE_DIR}/wrappers.py", - f"{PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_DIST_INFO_DIR}/METADATA", - ): - file_type, _mode, _uid, _gid = parse_debugfs_stat(run_debugfs(debugfs, image, f"stat {image_path}", write=False)) - if file_type != "regular": - raise RuntimeError(f"{image_path} has inode type {file_type}, expected regular") - - run_debugfs(debugfs, image, f"dump -p {PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_PACKAGE_DIR}/__init__.py {init_file}", write=False) - run_debugfs(debugfs, image, f"dump -p {PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_PACKAGE_DIR}/wrappers.py {wrappers_file}", write=False) - run_debugfs(debugfs, image, f"dump -p {PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_DIST_INFO_DIR}/METADATA {metadata_file}", write=False) - except Exception: - return False - - return ( - b"from .wrappers import *" in init_file.read_bytes() - and b"class DBusAddress" in wrappers_file.read_bytes() - and f"Version: {JEEPNY_VERSION}".encode("utf-8") in metadata_file.read_bytes() - ) - - -def parse_debugfs_stat(debugfs_output: str) -> tuple[str, str, int, int]: - type_match = re.search(r"Type:\s+([A-Za-z]+)", debugfs_output) - mode_match = re.search(r"Mode:\s+([0-7]+)", debugfs_output) - user_match = re.search(r"User:\s+(\d+)", debugfs_output) - group_match = re.search(r"Group:\s+(\d+)", debugfs_output) - if not type_match or not mode_match or not user_match or not group_match: - raise RuntimeError(f"Unable to parse debugfs stat output:\n{debugfs_output}") - return type_match.group(1).lower(), mode_match.group(1), int(user_match.group(1)), int(group_match.group(1)) - - -def inode_mode_from_type_and_perms(file_type: str, perms_octal: str) -> str: - prefix = INODE_MODE_TYPE_PREFIX.get(file_type) - if prefix is None: - raise RuntimeError(f"Unsupported inode type '{file_type}' for mode conversion") - perms = perms_octal.strip() - if not perms: - raise RuntimeError("Empty permissions value in inode stat") - return f"{prefix}{int(perms, 8):03o}" - - -def sync_files_from_reference_image(debugfs: str, reference_img: Path, patched_img: Path, sync_paths: list[str], work_dir: Path) -> list[str]: - sync_dir = work_dir / "reference_sync" - sync_dir.mkdir(parents=True, exist_ok=True) - synced: list[str] = [] - - for image_path in sync_paths: - source_tmp = sync_dir / f"source{image_path.replace('/', '_')}" - verify_tmp = sync_dir / f"verify{image_path.replace('/', '_')}" - - stat_out = run_debugfs(debugfs, reference_img, f"stat {image_path}", write=False) - file_type, perms_octal, uid, gid = parse_debugfs_stat(stat_out) - mode_octal = inode_mode_from_type_and_perms(file_type, perms_octal) - - run_debugfs(debugfs, reference_img, f"dump -p {image_path} {source_tmp}", write=False) - print(f"Syncing {image_path} from reference image (mode={mode_octal}, uid={uid}, gid={gid})", flush=True) - write_regular_file_to_image(debugfs, patched_img, image_path, source_tmp, mode_octal, uid, gid) - - run_debugfs(debugfs, patched_img, f"dump -p {image_path} {verify_tmp}", write=False) - if sha256_file(source_tmp) != sha256_file(verify_tmp): - raise RuntimeError(f"Verification failed after syncing {image_path}") - synced.append(image_path) - - return synced + inode = parse_inode(run_debugfs(debugfs, image, f"stat {image_path}")) + for field, value in (("mode", f"0{int(mode_octal, 8):o}"), ("uid", str(uid)), ("gid", str(gid))): + run_debugfs(debugfs, image, f"set_inode_field <{inode}> {field} {value}", write=True) def sha256_file(path: Path) -> str: - h = hashlib.sha256() - with open(path, "rb") as f: - while True: - chunk = f.read(1024 * 1024) - if not chunk: - break - h.update(chunk) - return h.hexdigest() - - -def sha256_zstd_payload(path: Path) -> str: - try: - import zstandard - except ImportError as e: - raise RuntimeError("zstandard is required to verify the external-GPU firmware") from e - digest = hashlib.sha256() - with open(path, "rb") as compressed: - with zstandard.ZstdDecompressor().stream_reader(compressed) as source: - while chunk := source.read(1024 * 1024): - digest.update(chunk) + with open(path, "rb") as stream: + while chunk := stream.read(1024 * 1024): + digest.update(chunk) return digest.hexdigest() -def install_amdgpu_firmware_from_reference(debugfs: str, reference_img: Path, patched_img: Path, work_dir: Path) -> None: - ensure_directory_in_image(debugfs, patched_img, AMDGPU_FIRMWARE_PATH_IN_IMAGE, "040755", 0, 0) - firmware_dir = work_dir / "amdgpu_firmware" - firmware_dir.mkdir(parents=True, exist_ok=True) - - for filename, expected_payload_hash in AMDGPU_FIRMWARE_SHA256.items(): - image_path = f"{AMDGPU_FIRMWARE_PATH_IN_IMAGE}/{filename}" - local_file = firmware_dir / filename - verify_file = firmware_dir / f"{filename}.verify" - local_file.unlink(missing_ok=True) - verify_file.unlink(missing_ok=True) - - stat_out = run_debugfs(debugfs, reference_img, f"stat {image_path}", write=False) - file_type, perms_octal, uid, gid = parse_debugfs_stat(stat_out) - if file_type != "regular": - raise RuntimeError(f"Reference firmware {image_path} is {file_type}, expected regular") - run_debugfs(debugfs, reference_img, f"dump -p {image_path} {local_file}", write=False) - if sha256_zstd_payload(local_file) != expected_payload_hash: - raise RuntimeError(f"Reference firmware payload hash mismatch for {filename}") - - mode_octal = inode_mode_from_type_and_perms(file_type, perms_octal) - write_regular_file_to_image(debugfs, patched_img, image_path, local_file, mode_octal, uid, gid) - run_debugfs(debugfs, patched_img, f"dump -p {image_path} {verify_file}", write=False) - if sha256_file(local_file) != sha256_file(verify_file): - raise RuntimeError(f"Compressed firmware verification failed for {filename}") - if sha256_zstd_payload(verify_file) != expected_payload_hash: - raise RuntimeError(f"Installed firmware payload hash mismatch for {filename}") - - print(f"Installed and verified {len(AMDGPU_FIRMWARE_SHA256)} AMD firmware files", flush=True) +def verify_official_raw_hash(system_entry: dict, official_raw: Path) -> str: + expected = system_entry.get("hash_raw") + if not isinstance(expected, str) or not expected: + raise RuntimeError("Official system manifest entry has no hash_raw") + actual = sha256_file(official_raw) + if actual != expected.lower(): + raise RuntimeError(f"Official system image hash mismatch: got {actual}, expected {expected.lower()}") + return actual -def compress_xz(src: Path, dst: Path) -> None: - dst.parent.mkdir(parents=True, exist_ok=True) - tmp = dst.with_suffix(dst.suffix + ".part") - print(f"Compressing {src} -> {dst}", flush=True) - with open(tmp, "wb") as out: - proc = subprocess.run(["xz", "-T0", "-6", "-c", str(src)], stdout=out, stderr=subprocess.PIPE, text=True) - if proc.returncode != 0: - tmp.unlink(missing_ok=True) - raise RuntimeError(f"xz failed:\n{proc.stderr}") - tmp.replace(dst) +def fingerprint_image_paths(debugfs: str, image: Path, paths: tuple[str, ...], work_dir: Path, + label: str) -> dict[str, str]: + output_dir = work_dir / f"fingerprints_{label}" + output_dir.mkdir(parents=True, exist_ok=True) + fingerprints: dict[str, str] = {} + for image_path in paths: + local_path = output_dir / image_path.strip("/").replace("/", "_") + local_path.unlink(missing_ok=True) + run_debugfs(debugfs, image, f"dump -p {image_path} {local_path}") + fingerprints[image_path] = sha256_file(local_path) + return fingerprints -def update_manifest_system_entry(manifest: list[dict], new_url: str, new_hash_raw: str, size: int) -> list[dict]: +def read_image_text(debugfs: str, image: Path, image_path: str) -> str: + output = run_debugfs(debugfs, image, f"cat {image_path}") + lines = [line.strip() for line in output.splitlines() if line.strip() and not line.startswith("debugfs ")] + return lines[0] if lines else "" + + +def compress_xz(source: Path, destination: Path) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + partial = destination.with_suffix(destination.suffix + ".part") + print(f"Compressing {source} -> {destination}", flush=True) + with open(partial, "wb") as output: + result = subprocess.run(["xz", "-T0", "-6", "-c", str(source)], stdout=output, stderr=subprocess.PIPE) + if result.returncode != 0: + partial.unlink(missing_ok=True) + raise RuntimeError(f"xz failed: {result.stderr.decode('utf-8', 'replace')}") + partial.replace(destination) + + +def write_artifact_metadata(destination: Path, *, upstream_manifest: Path, upstream_version: str, + starpilot_version: str, raw_hash: str, raw_size: int, + protected_hashes: dict[str, str]) -> Path: + metadata_path = Path(str(destination) + ".metadata.json") + metadata = { + "upstream_manifest": str(upstream_manifest), + "upstream_version": upstream_version, + "starpilot_version": starpilot_version, + "raw_sha256": raw_hash, + "raw_size": raw_size, + "xz_sha256": sha256_file(destination), + "xz_size": destination.stat().st_size, + "protected_upstream_payloads": protected_hashes, + } + metadata_path.write_text(json.dumps(metadata, indent=2) + "\n", encoding="utf-8") + return metadata_path + + +def update_manifest_system_entry(manifest: list[dict], new_url: str, raw_hash: str, size: int) -> list[dict]: updated = json.loads(json.dumps(manifest)) - system_entry = get_system_entry(updated) - old_url = system_entry.get("url") - old_hash = system_entry.get("hash") - old_hash_raw = system_entry.get("hash_raw") - old_size = system_entry.get("size") - - system_entry["url"] = new_url - system_entry["hash"] = new_hash_raw - system_entry["hash_raw"] = new_hash_raw - system_entry["size"] = size - system_entry["sparse"] = False - system_entry["full_check"] = False - - if isinstance(old_url, str) and isinstance(old_hash, str) and isinstance(old_hash_raw, str) and isinstance(old_size, int): - system_entry["alt"] = { - "url": old_url, - "hash": old_hash, - "hash_raw": old_hash_raw, - "size": old_size, - } - + entry = get_system_entry(updated) + entry.update({ + "url": new_url, + "hash": raw_hash, + "hash_raw": raw_hash, + "size": size, + "sparse": False, + "full_check": False, + "has_ab": True, + "ondevice_hash": raw_hash, + }) + entry.pop("alt", None) + entry.pop("casync_caibx", None) + entry.pop("casync_store", None) return updated -def resolve_reference_source_image(args: argparse.Namespace, primary_manifest_path: Path, work_dir: Path) -> Path: - if args.reference_image: - ref_image = Path(args.reference_image).resolve() - if not ref_image.is_file(): - raise RuntimeError(f"Reference image not found: {ref_image}") - return ref_image - - reference_manifest_path: Path | None = None - if args.reference_manifest: - reference_manifest_path = Path(args.reference_manifest).resolve() - if not reference_manifest_path.is_file(): - raise RuntimeError(f"Reference manifest not found: {reference_manifest_path}") - else: - reference_manifest_path = find_default_reference_manifest(primary_manifest_path) - - if args.reference_source_url: - reference_url = args.reference_source_url - elif reference_manifest_path is not None: - reference_manifest = load_manifest(reference_manifest_path) - reference_entry = get_system_entry(reference_manifest) - reference_url = pick_source_url(reference_entry, None) - print(f"Using reference AGNOS manifest: {reference_manifest_path}", flush=True) - else: - raise RuntimeError( - "No reference image source found. Set --reference-image, --reference-source-url, or --reference-manifest." - ) - - reference_download = work_dir / "reference_system.img" - if args.force_download and reference_download.exists(): - reference_download.unlink() - if not reference_download.exists(): - download(reference_url, reference_download) - return reference_download - - def main() -> int: args = parse_args() debugfs = find_debugfs() - - manifest_path = Path(args.manifest).resolve() - manifest = load_manifest(manifest_path) - system_entry = get_system_entry(manifest) + target_manifest_path = Path(args.manifest).resolve() + target_manifest = load_manifest(target_manifest_path) + upstream_manifest_path = resolve_upstream_manifest(args.upstream_manifest, target_manifest_path) + upstream_manifest = load_manifest(upstream_manifest_path) + upstream_system_entry = get_system_entry(upstream_manifest) work_dir = Path(args.work_dir).resolve() work_dir.mkdir(parents=True, exist_ok=True) - if args.source_image: - downloaded_img = Path(args.source_image).resolve() - if not downloaded_img.is_file(): - raise RuntimeError(f"Source image not found: {downloaded_img}") + source = Path(args.source_image).resolve() + if not source.is_file(): + raise RuntimeError(f"Official source image not found: {source}") else: - source_url = pick_source_url(system_entry, args.source_url) - downloaded_img = work_dir / "base_system.img" - if args.force_download and downloaded_img.exists(): - downloaded_img.unlink() - if not downloaded_img.exists(): - download(source_url, downloaded_img) + source = work_dir / "official_system.payload" + if args.force_download: + source.unlink(missing_ok=True) + if not source.exists(): + download(pick_source_url(upstream_system_entry, args.source_url), source) - raw_img = work_dir / "base_system.ext4.img" - materialize_ext4_image(downloaded_img, raw_img, work_dir, "base_system", force=args.force_download) + official_raw = work_dir / "official_system.ext4.img" + materialize_ext4_image(source, official_raw, work_dir, force=args.force_download) + verify_official_raw_hash(upstream_system_entry, official_raw) + source_version = read_image_text(debugfs, official_raw, VERSION_PATH_IN_IMAGE) + required_source_version = expected_upstream_version(args.set_version, args.expected_source_version) + if source_version != required_source_version: + raise RuntimeError( + f"Official source /VERSION is {source_version!r}, expected {required_source_version!r}", + ) - patched_img = work_dir / "patched_system.ext4.img" - if patched_img.exists(): - patched_img.unlink() - print(f"Copying source image -> {patched_img}", flush=True) - shutil.copy2(raw_img, patched_img) + patched_raw = work_dir / "starpilot_system.ext4.img" + patched_raw.unlink(missing_ok=True) + shutil.copy2(official_raw, patched_raw) - sync_paths = [] if args.disable_comma_file_sync else parse_sync_file_list(args.sync_comma_files) - reference_raw = None - if sync_paths or not args.disable_usbgpu_firmware: - reference_source_img = resolve_reference_source_image(args, manifest_path, work_dir) - reference_raw = work_dir / "reference_system.ext4.img" - materialize_ext4_image(reference_source_img, reference_raw, work_dir, "reference_system", force=args.force_download) + before = fingerprint_image_paths(debugfs, official_raw, UPSTREAM_PROTECTED_PAYLOADS, work_dir, "official") + version_file = work_dir / "VERSION.starpilot" + version_file.write_text(args.set_version.strip() + "\n", encoding="utf-8") + write_allowed_file(debugfs, patched_raw, VERSION_PATH_IN_IMAGE, version_file) + after = fingerprint_image_paths(debugfs, patched_raw, UPSTREAM_PROTECTED_PAYLOADS, work_dir, "starpilot") + if before != after: + changed = sorted(path for path in before if before[path] != after.get(path)) + raise RuntimeError(f"Upstream recovery payload changed unexpectedly: {changed}") + if read_image_text(debugfs, patched_raw, VERSION_PATH_IN_IMAGE) != args.set_version.strip(): + raise RuntimeError("Failed to write StarPilot AGNOS version marker") - if sync_paths: - assert reference_raw is not None - print(f"Syncing /usr/comma payload files from reference image: {reference_raw}", flush=True) - synced_files = sync_files_from_reference_image(debugfs, reference_raw, patched_img, sync_paths, work_dir) - print(f"Synced {len(synced_files)} /usr/comma files from reference image", flush=True) + raw_hash = sha256_file(patched_raw) + raw_size = patched_raw.stat().st_size + output_xz = Path(args.output_xz).resolve() if args.output_xz else work_dir / f"system-{raw_hash}.img.xz" + compress_xz(patched_raw, output_xz) + metadata_path = write_artifact_metadata( + output_xz, + upstream_manifest=upstream_manifest_path, + upstream_version=source_version, + starpilot_version=args.set_version.strip(), + raw_hash=raw_hash, + raw_size=raw_size, + protected_hashes=after, + ) - if not args.disable_usbgpu_firmware: - assert reference_raw is not None - print(f"Installing external-GPU firmware from reference image: {reference_raw}", flush=True) - install_amdgpu_firmware_from_reference(debugfs, reference_raw, patched_img, work_dir) - - preserved_paths = { - RESET_PATH_IN_IMAGE: "comma_reset", - SETUP_PATH_IN_IMAGE: "comma_setup", - COMMA_SH_PATH_IN_IMAGE: "comma_sh", - MAGIC_PATH_IN_IMAGE: "comma_magic", - BG_PATH_IN_IMAGE: "comma_bg", - } - expected_hashes: dict[str, str] = {} - for image_path, label in preserved_paths.items(): - preserved_file = work_dir / f"{label}.preserved" - print(f"Recording existing {image_path} for preservation", flush=True) - run_debugfs(debugfs, patched_img, f"dump -p {image_path} {preserved_file}", write=False) - expected_hashes[image_path] = sha256_file(preserved_file) - - original_updater = work_dir / "comma_updater.orig" - patched_updater = work_dir / "comma_updater.patched" - verify_updater = work_dir / "comma_updater.verify" - original_weston = work_dir / "weston_service.orig" - patched_weston = work_dir / "weston_service.patched" - verify_weston = work_dir / "weston_service.verify" - patched_comma_sh = work_dir / "comma_sh.patched" - - original_comma_sh = work_dir / "comma_sh.preserved" - comma_sh_patched_data = patch_comma_sh_display_wait(original_comma_sh.read_bytes()) - patched_comma_sh.write_bytes(comma_sh_patched_data) - - print("Writing patched /usr/comma/comma.sh display readiness wait", flush=True) - write_regular_file_to_image(debugfs, patched_img, COMMA_SH_PATH_IN_IMAGE, patched_comma_sh, "100775", 0, 0) - expected_hashes[COMMA_SH_PATH_IN_IMAGE] = sha256_file(patched_comma_sh) - - print("Extracting weston.service from image", flush=True) - run_debugfs(debugfs, patched_img, f"dump -p {WESTON_SERVICE_PATH_IN_IMAGE} {original_weston}", write=False) - - weston_original_data = original_weston.read_bytes() - weston_patched_data = patch_weston_service(weston_original_data) - if weston_patched_data == weston_original_data: - print("weston.service already contains the expected boot-logo patch; continuing", flush=True) - patched_weston.write_bytes(weston_patched_data) - - print("Writing patched weston.service back into image", flush=True) - write_regular_file_to_image(debugfs, patched_img, WESTON_SERVICE_PATH_IN_IMAGE, patched_weston, "100644", 0, 0) - - run_debugfs(debugfs, patched_img, f"dump -p {WESTON_SERVICE_PATH_IN_IMAGE} {verify_weston}", write=False) - verify_weston_data = verify_weston.read_bytes() - if not weston_service_has_expected_content(verify_weston_data): - raise RuntimeError("weston.service verification failed after writing weston.service file into image") - - print("Extracting /usr/comma/updater from image", flush=True) - run_debugfs(debugfs, patched_img, f"dump -p {UPDATER_PATH_IN_IMAGE} {original_updater}", write=False) - - updater_original_data = original_updater.read_bytes() - updater_patched_data = patch_updater_zipapp(updater_original_data) - if updater_patched_data == updater_original_data: - print("Updater zipapp already contains the expected selector patch; continuing", flush=True) - patched_updater.write_bytes(updater_patched_data) - - print("Writing patched /usr/comma/updater back into image", flush=True) - write_regular_file_to_image(debugfs, patched_img, UPDATER_PATH_IN_IMAGE, patched_updater, "100775", 0, 0) - - run_debugfs(debugfs, patched_img, f"dump -p {UPDATER_PATH_IN_IMAGE} {verify_updater}", write=False) - verify_updater_data = verify_updater.read_bytes() - if not updater_zipapp_has_expected_content(verify_updater_data): - raise RuntimeError("Updater zipapp verification failed after writing updater file into image") - - install_jeepney_into_image(debugfs, patched_img, work_dir) - if not image_has_jeepney(debugfs, patched_img, work_dir): - raise RuntimeError("jeepney verification failed after installing package into image") - - for image_path, label in preserved_paths.items(): - verify_file = work_dir / f"{label}.verify" - run_debugfs(debugfs, patched_img, f"dump -p {image_path} {verify_file}", write=False) - if image_path == COMMA_SH_PATH_IN_IMAGE and not comma_sh_has_expected_display_wait(verify_file.read_bytes()): - raise RuntimeError("comma.sh display readiness verification failed") - if sha256_file(verify_file) != expected_hashes[image_path]: - raise RuntimeError(f"{image_path} does not match the expected generated payload") - - if args.set_version: - version_file = work_dir / "VERSION.patched" - version_file.write_text(args.set_version.strip() + "\n", encoding="utf-8") - print(f"Writing {VERSION_PATH_IN_IMAGE}={args.set_version.strip()}", flush=True) - write_regular_file_to_image(debugfs, patched_img, VERSION_PATH_IN_IMAGE, version_file, "100644", 0, 0) - version_raw = run_debugfs(debugfs, patched_img, f"cat {VERSION_PATH_IN_IMAGE}", write=False) - version_lines = [ln.strip() for ln in version_raw.splitlines() if ln.strip() and not ln.startswith("debugfs ")] - version_verify = version_lines[0] if version_lines else "" - if version_verify != args.set_version.strip(): - raise RuntimeError(f"/VERSION mismatch after patch: got '{version_verify}', expected '{args.set_version.strip()}'") - - raw_hash = sha256_file(patched_img) - raw_size = patched_img.stat().st_size - - default_name = f"system-{raw_hash}.img.xz" - output_xz = Path(args.output_xz).resolve() if args.output_xz else (work_dir / default_name) - compress_xz(patched_img, output_xz) - - print("") - print("Patched AGNOS system artifact ready:") - print(f" raw image: {patched_img}") - print(f" xz image: {output_xz}") - print(f" raw sha256/hash_raw: {raw_hash}") - print(f" size: {raw_size}") - print("") + print("Stock-plus-C3 AGNOS system artifact ready:") + print(f" upstream manifest: {upstream_manifest_path}") + print(f" upstream version: {source_version}") + print(f" StarPilot version: {args.set_version.strip()}") + print(f" raw image: {patched_raw}") + print(f" xz image: {output_xz}") + print(f" raw sha256: {raw_hash}") + print(f" raw size: {raw_size}") + print(f" metadata: {metadata_path}") + print(" protected payloads: byte-identical to upstream") if args.new_url: - new_manifest = update_manifest_system_entry(manifest, args.new_url, raw_hash, raw_size) - out_path: Path + updated_manifest = update_manifest_system_entry(target_manifest, args.new_url, raw_hash, raw_size) if args.in_place_manifest: - out_path = manifest_path + output_manifest = target_manifest_path elif args.manifest_out: - out_path = Path(args.manifest_out).resolve() + output_manifest = Path(args.manifest_out).resolve() else: - out_path = work_dir / "agnos.patched.json" - out_path.write_text(json.dumps(new_manifest, indent=2) + "\n") - print(f"Updated manifest written: {out_path}") + output_manifest = work_dir / "agnos.starpilot.json" + output_manifest.write_text(json.dumps(updated_manifest, indent=2) + "\n", encoding="utf-8") + print(f" updated manifest: {output_manifest}") else: - print("No --new-url provided. Manifest not updated.") - print("Set system entry values to:") - print(json.dumps({ - "url": "", - "hash": raw_hash, - "hash_raw": raw_hash, - "size": raw_size, - "sparse": False, - "full_check": False, - "has_ab": True, - }, indent=2)) - + print("No --new-url supplied; the checked-in manifest was not changed.") return 0 diff --git a/tools/agnos/test_patch_system_reset_image.py b/tools/agnos/test_patch_system_reset_image.py index c2742978a..49c0d7703 100644 --- a/tools/agnos/test_patch_system_reset_image.py +++ b/tools/agnos/test_patch_system_reset_image.py @@ -1,106 +1,88 @@ +import importlib.util +import json from pathlib import Path -import runpy import pytest -from tools.agnos.patch_system_reset_image import ( - AMDGPU_FIRMWARE_SHA256, - COMMA_SH_DISPLAY_WAIT_PATCH_MARKER, - comma_sh_has_expected_display_wait, - find_default_reference_manifest, - format_debugfs_mode, - patch_comma_sh_display_wait, - patch_setup_branding_script, - sha256_zstd_payload, -) + +def _load_patch_module(): + path = Path(__file__).resolve().parent / "patch_system_reset_image.py" + spec = importlib.util.spec_from_file_location("patch_system_reset_image_under_test", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module -ORIGINAL_DISPLAY_WAIT = b'''#!/usr/bin/env bash -echo "waiting for magic" -for i in {1..200}; do - if systemctl is-active --quiet magic && [ -S /tmp/drmfd.sock ]; then - break - fi - sleep 0.1 -done - -if systemctl is-active --quiet magic && [ -S /tmp/drmfd.sock ]; then - echo "magic ready after ${SECONDS}s" -else - echo "timed out waiting for magic, ${SECONDS}s" -fi - -exec /data/continue.sh -''' +patch_image = _load_patch_module() +ALLOWED_IMAGE_MUTATIONS = patch_image.ALLOWED_IMAGE_MUTATIONS +UPSTREAM_PROTECTED_PAYLOADS = patch_image.UPSTREAM_PROTECTED_PAYLOADS +VERSION_PATH_IN_IMAGE = patch_image.VERSION_PATH_IN_IMAGE +resolve_upstream_manifest = patch_image.resolve_upstream_manifest +update_manifest_system_entry = patch_image.update_manifest_system_entry +write_allowed_file = patch_image.write_allowed_file +expected_upstream_version = patch_image.expected_upstream_version +verify_official_raw_hash = patch_image.verify_official_raw_hash -def test_patch_comma_sh_display_wait_uses_available_display_service(): - patched = patch_comma_sh_display_wait(ORIGINAL_DISPLAY_WAIT) - - assert COMMA_SH_DISPLAY_WAIT_PATCH_MARKER.encode() in patched - assert b"systemctl cat magic.service" in patched - assert b"systemctl is-active --quiet magic" in patched - assert b"systemctl is-active --quiet weston-ready" in patched - assert b"[ -S /var/tmp/weston/wayland-0 ]" in patched - assert comma_sh_has_expected_display_wait(patched) - assert patch_comma_sh_display_wait(patched) == patched +def test_only_version_is_mutable(): + assert ALLOWED_IMAGE_MUTATIONS == {VERSION_PATH_IN_IMAGE} + assert set(UPSTREAM_PROTECTED_PAYLOADS) >= { + "/usr/comma/comma.sh", + "/usr/comma/reset", + "/usr/comma/setup", + "/usr/comma/updater", + "/etc/NetworkManager/NetworkManager.conf", + "/etc/NetworkManager/conf.d/10-globally-managed-devices.conf", + "/lib/systemd/system/NetworkManager.service", + } -def test_patch_comma_sh_display_wait_rejects_unknown_layout(): - with pytest.raises(RuntimeError, match="display readiness wait"): - patch_comma_sh_display_wait(b"#!/usr/bin/env bash\nexec /data/continue.sh\n") +def test_starpilot_revision_maps_to_exact_upstream_version(): + assert expected_upstream_version("19.6.3", None) == "19.6" + assert expected_upstream_version("19.6.3", "19.6-test") == "19.6-test" + with pytest.raises(RuntimeError, match="revision suffix"): + expected_upstream_version("19.6", None) -@pytest.mark.parametrize("slider_text", ["slide to use", "slide to install"]) -def test_patch_mici_setup_branding_handles_old_and_new_labels(slider_text): - original = f'''OPENPILOT_URL = "https://openpilot.comma.ai" -self._openpilot_slider = LargerSlider("{slider_text}\\nopenpilot", callback) -self._continue_button = BigPillButton("install openpilot", green=True) -self._continue_button.set_text("install openpilot" if not custom_software else "choose software") -'''.encode() +def test_official_source_hash_is_enforced(tmp_path): + image = tmp_path / "official.img" + image.write_bytes(b"official") + digest = patch_image.sha256_file(image) - patched = patch_setup_branding_script(original, "openpilot/system/ui/mici_setup.py") - - assert b"installer.comma.ai/firestar5683/StarPilot" in patched - assert f"{slider_text}\\nstarpilot".encode() in patched - assert b"install StarPilot" in patched - assert b"install openpilot" not in patched + assert verify_official_raw_hash({"hash_raw": digest}, image) == digest + with pytest.raises(RuntimeError, match="hash mismatch"): + verify_official_raw_hash({"hash_raw": "0" * 64}, image) -@pytest.mark.parametrize(("mode", "expected"), [ - ("100775", "0100775"), - ("100644", "0100644"), - ("040755", "040755"), - ("120777", "0120777"), -]) -def test_format_debugfs_mode(mode, expected): - assert format_debugfs_mode(mode) == expected +def test_write_rejects_non_version_path(tmp_path): + with pytest.raises(RuntimeError, match="Refusing non-C3 AGNOS system mutation"): + write_allowed_file("debugfs", tmp_path / "system.img", "/usr/comma/setup", tmp_path / "setup") -def test_external_gpu_firmware_matches_tinygrad_requirements(): - firmware_metadata = Path(__file__).resolve().parents[2] / "tinygrad/runtime/autogen/am/fw.py" - hashes = runpy.run_path(firmware_metadata)["hashes"] - expected = {filename.removesuffix(".zst"): digest for filename, digest in AMDGPU_FIRMWARE_SHA256.items()} - assert all(hashes[filename] == digest for filename, digest in expected.items()) - - -def test_zstd_payload_hash(tmp_path): - import hashlib - import zstandard - - payload = b"external GPU firmware payload" - compressed = tmp_path / "firmware.bin.zst" - compressed.write_bytes(zstandard.ZstdCompressor().compress(payload)) - - assert sha256_zstd_payload(compressed) == hashlib.sha256(payload).hexdigest() - - -def test_default_reference_manifest_uses_sibling_openpilot(tmp_path): - primary = tmp_path / "starpilot/system/hardware/tici/agnos.json" - reference = tmp_path / "openpilot/openpilot/system/hardware/tici/agnos.json" - primary.parent.mkdir(parents=True) - reference.parent.mkdir(parents=True) +def test_upstream_manifest_cannot_be_primary(tmp_path): + primary = tmp_path / "agnos.json" primary.write_text("[]") - reference.write_text("[]") + with pytest.raises(RuntimeError, match="Refusing to use StarPilot"): + resolve_upstream_manifest(str(primary), primary) - assert find_default_reference_manifest(primary) == reference.resolve() + +def test_update_manifest_changes_only_system_entry(): + original = [ + {"name": "boot", "url": "custom-boot", "hash": "boot-hash"}, + {"name": "system", "url": "official", "hash": "old", "alt": {"url": "old-alt"}}, + ] + updated = update_manifest_system_entry(original, "hosted", "new-hash", 123) + + assert updated[0] == original[0] + assert updated[1] == { + "name": "system", + "url": "hosted", + "hash": "new-hash", + "hash_raw": "new-hash", + "size": 123, + "sparse": False, + "full_check": False, + "has_ab": True, + "ondevice_hash": "new-hash", + } + assert json.dumps(original)