mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-09-01 21:53:42 +08:00
firestar5683
635 lines
28 KiB
Python
Executable File
635 lines
28 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Build StarPilot AGNOS from the exact upstream system image.
|
|
|
|
The output starts with comma's pinned AGNOS system partition and only adds the
|
|
Python packages required by StarPilot's older runtime and C3 support. Setup,
|
|
reset, installer, updater, Magic, NetworkManager, and every existing upstream
|
|
Python package are verified and are never replaced by this tool.
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import lzma
|
|
import os
|
|
import re
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
VERSION_PATH_IN_IMAGE = "/VERSION"
|
|
SITE_PACKAGES_PATH_IN_IMAGE = "/usr/local/venv/lib/python3.12/site-packages"
|
|
STAR_PILOT_DEPENDENCY_NAMES = (
|
|
# C3/runtime compatibility
|
|
"crcmod", "crcmod-1.7.dist-info", "serial", "pyserial-3.5.dist-info",
|
|
"kaitaistruct.py", "kaitaistruct-0.11.dist-info",
|
|
# StarPilot always-on/default features
|
|
"cv2", "opencv_python_headless-4.11.0.86.dist-info", "opencv_python_headless.libs",
|
|
"mapbox_earcut.cpython-312-aarch64-linux-gnu.so", "mapbox_earcut-1.0.3.dist-info",
|
|
"jsonrpc", "json_rpc-1.15.0.dist-info", "xattr", "xattr-1.2.0.dist-info",
|
|
"onnx", "onnx-1.18.0.dist-info", "google", "protobuf-7.35.1.dist-info",
|
|
"typing_extensions.py", "typing_extensions-4.16.0.dist-info",
|
|
# Existing body/web tools still use aiohttp and PyAudio. The aiortc stack is
|
|
# deliberately not copied; WebRTC uses upstream's libdatachannel backend.
|
|
"aiohappyeyeballs", "aiohappyeyeballs-2.7.1.dist-info",
|
|
"aiohttp", "aiohttp-3.12.15.dist-info",
|
|
"aiosignal", "aiosignal-1.4.0.dist-info",
|
|
"attr", "attrs", "attrs-26.1.0.dist-info",
|
|
"frozenlist", "frozenlist-1.8.0.dist-info",
|
|
"multidict", "multidict-6.7.1.dist-info",
|
|
"propcache", "propcache-0.5.2.dist-info",
|
|
"yarl", "yarl-1.24.5.dist-info",
|
|
"pyaudio", "pyaudio-0.2.14.dist-info",
|
|
)
|
|
STAR_PILOT_DEPENDENCY_PATHS = tuple(
|
|
f"{SITE_PACKAGES_PATH_IN_IMAGE}/{name}" for name in STAR_PILOT_DEPENDENCY_NAMES
|
|
)
|
|
C3_DEPENDENCY_PATHS = tuple(
|
|
f"{SITE_PACKAGES_PATH_IN_IMAGE}/{name}"
|
|
for name in ("crcmod", "crcmod-1.7.dist-info", "serial", "pyserial-3.5.dist-info", "kaitaistruct.py", "kaitaistruct-0.11.dist-info")
|
|
)
|
|
ALLOWED_IMAGE_MUTATIONS = frozenset({VERSION_PATH_IN_IMAGE, *STAR_PILOT_DEPENDENCY_PATHS})
|
|
|
|
# Exact system partition pinned by ~/openpilot as of the 19.6 AGNOS release.
|
|
UPSTREAM_VERSION = "19.6"
|
|
UPSTREAM_SYSTEM_URL = (
|
|
"https://commadist.azureedge.net/agnosupdate/"
|
|
"system-5b6ce7965904a157fd3a134ccfcb854f9ca5c1cc2a26b7cb80a4fa4e1cc4aaa3.img.xz"
|
|
)
|
|
UPSTREAM_RAW_SHA256 = "5b6ce7965904a157fd3a134ccfcb854f9ca5c1cc2a26b7cb80a4fa4e1cc4aaa3"
|
|
UPSTREAM_RAW_SIZE = 4_718_592_000
|
|
UPSTREAM_SITE_PACKAGES_COUNT = 213
|
|
|
|
# Compatibility packages are copied from StarPilot's exact, previously
|
|
# deployed and field-tested 19.6.2 image. Existing upstream paths are never
|
|
# overwritten.
|
|
C3_DEPENDENCY_SOURCE_URL = (
|
|
"https://www.dropbox.com/scl/fi/pewhzpqzi3aewuiaffc6m/system10.img.xz"
|
|
"?rlkey=olzrzulhs93zzghnjrskmdwxt&st=exnfk2oz&dl=1"
|
|
)
|
|
C3_DEPENDENCY_SOURCE_RAW_SHA256 = "ab395d4c963a908ab86709f1a6580a62dd24cb34ee71cf5fd5ec29d7d48d0e10"
|
|
C3_DEPENDENCY_SOURCE_RAW_SIZE = 4_718_592_000
|
|
CANDIDATE_SITE_PACKAGES_COUNT = UPSTREAM_SITE_PACKAGES_COUNT + len(STAR_PILOT_DEPENDENCY_PATHS)
|
|
|
|
# Hashes from that exact upstream image. Equality keeps the recovery path stock.
|
|
PROTECTED_PAYLOAD_HASHES = {
|
|
"/etc/NetworkManager/NetworkManager.conf": "779db62d2d4c5f8ce504c5d1f2994d34a9f35296d5efb7f3a48cb1e8a0d4778e",
|
|
"/etc/NetworkManager/conf.d/10-globally-managed-devices.conf": "45e653e2f709c027fad41f2d86b70e008b72c6bf4d34590b4765ebe8fe3ea948",
|
|
"/lib/systemd/system/NetworkManager.service": "fb33a80bf8c78b3af004d4b294c47a0139e37742c1d0d5a6a7663c7d1f4a2b48",
|
|
"/usr/comma/updater": "9df4edbeb5849de03f9c2d691d04646af84a3ef74c2f33be8e73d9281daebe99",
|
|
"/usr/comma/reset": "97ed6413515d0674442c42ae6e20baccf66dd6bb4ec382ee4cf0cc5ebe84e739",
|
|
"/usr/comma/magic.py": "c4416e66b127b31c17d08e6723ad46d12af7683e56626512d66c708b5f347ac9",
|
|
"/usr/comma/installer": "85f6d9e54286a3842920d6967b187478b4e43d6171c331d72d3fb3102106e101",
|
|
"/usr/comma/setup_keys": "934f74ab4b2ac06048418c2857be3a041e192ec03c09979987691c23c91353bd",
|
|
"/usr/comma/setup": "c382ce266653bad781c25e403ddea4af508aa6f3ea2eef3f568d964982fad9d6",
|
|
"/usr/comma/comma.sh": "bcba2b336cf0ca852786f8a58bbce407e0e9fe952c26fc5d903f6d9a34b44b4f",
|
|
}
|
|
|
|
UPSTREAM_REQUIRED_VENV_PATHS = {
|
|
"capnp": f"{SITE_PACKAGES_PATH_IN_IMAGE}/capnp",
|
|
"numpy": f"{SITE_PACKAGES_PATH_IN_IMAGE}/numpy",
|
|
"Crypto": f"{SITE_PACKAGES_PATH_IN_IMAGE}/Crypto",
|
|
"tqdm": f"{SITE_PACKAGES_PATH_IN_IMAGE}/tqdm",
|
|
"raylib": f"{SITE_PACKAGES_PATH_IN_IMAGE}/raylib",
|
|
}
|
|
REQUIRED_VENV_PATHS = {
|
|
**UPSTREAM_REQUIRED_VENV_PATHS,
|
|
"crcmod": f"{SITE_PACKAGES_PATH_IN_IMAGE}/crcmod",
|
|
"serial": f"{SITE_PACKAGES_PATH_IN_IMAGE}/serial",
|
|
"kaitaistruct": f"{SITE_PACKAGES_PATH_IN_IMAGE}/kaitaistruct.py",
|
|
"cv2": f"{SITE_PACKAGES_PATH_IN_IMAGE}/cv2",
|
|
"mapbox_earcut": f"{SITE_PACKAGES_PATH_IN_IMAGE}/mapbox_earcut.cpython-312-aarch64-linux-gnu.so",
|
|
"jsonrpc": f"{SITE_PACKAGES_PATH_IN_IMAGE}/jsonrpc",
|
|
"xattr": f"{SITE_PACKAGES_PATH_IN_IMAGE}/xattr",
|
|
"onnx": f"{SITE_PACKAGES_PATH_IN_IMAGE}/onnx",
|
|
"aiohttp": f"{SITE_PACKAGES_PATH_IN_IMAGE}/aiohttp",
|
|
"pyaudio": f"{SITE_PACKAGES_PATH_IN_IMAGE}/pyaudio",
|
|
}
|
|
|
|
ANDROID_SPARSE_MAGIC = 0xED26FF3A
|
|
CHUNK_TYPE_RAW = 0xCAC1
|
|
CHUNK_TYPE_FILL = 0xCAC2
|
|
CHUNK_TYPE_DONT_CARE = 0xCAC3
|
|
CHUNK_TYPE_CRC32 = 0xCAC4
|
|
XZ_MAGIC = b"\xFD7zXZ\x00"
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Build an upstream-identical StarPilot AGNOS system image")
|
|
parser.add_argument("--manifest", default="system/hardware/tici/agnos.json",
|
|
help="Manifest to copy when writing an optional candidate manifest")
|
|
parser.add_argument("--source-url", default=UPSTREAM_SYSTEM_URL, help="Exact upstream system image URL")
|
|
parser.add_argument("--source-image", help="Use a local exact upstream raw, sparse, or .xz image")
|
|
parser.add_argument("--c3-deps-url", default=C3_DEPENDENCY_SOURCE_URL,
|
|
help="Exact prior StarPilot image containing the compatibility packages")
|
|
parser.add_argument("--c3-deps-image", help="Use a local exact StarPilot dependency source image")
|
|
parser.add_argument("--set-version", required=True, help="StarPilot revision, for example 19.6.5")
|
|
parser.add_argument("--work-dir", default=".cache/agnos_upstream_system")
|
|
parser.add_argument("--output-xz", help="Output .img.xz path")
|
|
parser.add_argument("--new-url", help="Hosted output URL for an optional candidate manifest")
|
|
parser.add_argument("--manifest-out", help="Candidate manifest output path; never overwrites the checked-in manifest")
|
|
parser.add_argument("--force-download", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def find_tool(name: str, extra_candidates: tuple[str, ...] = ()) -> str:
|
|
for candidate in (os.environ.get(name.upper()), name, *extra_candidates):
|
|
if candidate and (shutil.which(candidate) or Path(candidate).is_file()):
|
|
return candidate
|
|
raise RuntimeError(f"{name} not found")
|
|
|
|
|
|
def find_debugfs() -> str:
|
|
return find_tool("debugfs", ("/opt/homebrew/opt/e2fsprogs/sbin/debugfs",))
|
|
|
|
|
|
def find_e2fsck() -> str:
|
|
return find_tool("e2fsck", ("/opt/homebrew/opt/e2fsprogs/sbin/e2fsck",))
|
|
|
|
|
|
def run_cmd(command: list[str], *, allowed_returncodes: frozenset[int] = frozenset({0})) -> subprocess.CompletedProcess[str]:
|
|
result = subprocess.run(command, check=False, capture_output=True, text=True)
|
|
if result.returncode not in allowed_returncodes:
|
|
raise RuntimeError(f"Command failed ({result.returncode}): {' '.join(command)}\n{result.stdout}\n{result.stderr}")
|
|
return result
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
while chunk := stream.read(8 * 1024 * 1024):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def validate_target_version(version: str) -> str:
|
|
clean = version.strip()
|
|
if not re.fullmatch(r"19\.6\.\d+", clean):
|
|
raise RuntimeError("Target version must be a 19.6.x StarPilot revision")
|
|
if int(clean.rsplit(".", 1)[1]) < 1:
|
|
raise RuntimeError("Target version must be newer than upstream 19.6")
|
|
return clean
|
|
|
|
|
|
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 exact upstream AGNOS: {url}", flush=True)
|
|
with urllib.request.urlopen(url) as source, partial.open("wb") as output:
|
|
shutil.copyfileobj(source, output, length=8 * 1024 * 1024)
|
|
partial.replace(destination)
|
|
|
|
|
|
def is_xz_file(path: Path) -> bool:
|
|
with path.open("rb") as stream:
|
|
return stream.read(len(XZ_MAGIC)) == XZ_MAGIC
|
|
|
|
|
|
def decompress_xz(source: Path, destination: Path) -> None:
|
|
partial = destination.with_suffix(destination.suffix + ".part")
|
|
print(f"Decompressing {source}", flush=True)
|
|
with lzma.open(source, "rb") as compressed, partial.open("wb") as output:
|
|
shutil.copyfileobj(compressed, output, length=8 * 1024 * 1024)
|
|
partial.replace(destination)
|
|
|
|
|
|
def is_android_sparse(path: Path) -> bool:
|
|
with path.open("rb") as stream:
|
|
raw = stream.read(4)
|
|
return len(raw) == 4 and struct.unpack("<I", raw)[0] == ANDROID_SPARSE_MAGIC
|
|
|
|
|
|
def unsparse_image(source: Path, destination: Path) -> None:
|
|
print(f"Expanding Android sparse image {source}", flush=True)
|
|
with source.open("rb") as source_file, destination.open("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(
|
|
"<I4H4I", header,
|
|
)
|
|
if magic != ANDROID_SPARSE_MAGIC or major != 1:
|
|
raise RuntimeError("Unsupported Android sparse image")
|
|
source_file.seek(file_header_size)
|
|
for _ in range(total_chunks):
|
|
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 payload_size != output_size:
|
|
raise RuntimeError("Sparse RAW chunk size mismatch")
|
|
remaining = payload_size
|
|
while remaining:
|
|
data = source_file.read(min(8 * 1024 * 1024, remaining))
|
|
if not data:
|
|
raise RuntimeError("Sparse RAW chunk is truncated")
|
|
output.write(data)
|
|
remaining -= len(data)
|
|
elif chunk_type == CHUNK_TYPE_FILL:
|
|
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:
|
|
block = pattern * (block_size // 4)
|
|
for _ in range(chunk_blocks):
|
|
output.write(block)
|
|
elif chunk_type == CHUNK_TYPE_DONT_CARE:
|
|
source_file.seek(payload_size, os.SEEK_CUR)
|
|
output.seek(output_size, os.SEEK_CUR)
|
|
elif chunk_type == CHUNK_TYPE_CRC32:
|
|
source_file.seek(payload_size, os.SEEK_CUR)
|
|
else:
|
|
raise RuntimeError(f"Unknown sparse chunk type: 0x{chunk_type:04x}")
|
|
output.truncate(total_blocks * block_size)
|
|
|
|
|
|
def materialize_upstream_image(source: Path, destination: Path, work_dir: Path) -> None:
|
|
candidate = source
|
|
if is_xz_file(source):
|
|
decompressed = work_dir / "upstream_system.decompressed.img"
|
|
if not decompressed.exists():
|
|
decompress_xz(source, decompressed)
|
|
candidate = decompressed
|
|
if destination.exists():
|
|
return
|
|
if is_android_sparse(candidate):
|
|
unsparse_image(candidate, destination)
|
|
else:
|
|
shutil.copy2(candidate, destination)
|
|
|
|
|
|
def run_debugfs(debugfs: str, image: Path, request: str, *, write: bool = False) -> str:
|
|
command = [debugfs]
|
|
if write:
|
|
command.append("-w")
|
|
command += ["-R", request, str(image)]
|
|
result = run_cmd(command)
|
|
return f"{result.stdout}\n{result.stderr}"
|
|
|
|
|
|
def parse_inode(output: str) -> int:
|
|
match = re.search(r"Inode:\s+(\d+)", output)
|
|
if not match:
|
|
raise RuntimeError(f"Unable to parse inode:\n{output}")
|
|
return int(match.group(1))
|
|
|
|
|
|
def write_version(debugfs: str, image: Path, local_file: Path) -> None:
|
|
expected_mutations = frozenset({VERSION_PATH_IN_IMAGE, *STAR_PILOT_DEPENDENCY_PATHS})
|
|
if ALLOWED_IMAGE_MUTATIONS != expected_mutations:
|
|
raise RuntimeError("AGNOS mutation allowlist contains an unexpected path")
|
|
run_debugfs(debugfs, image, f"rm {VERSION_PATH_IN_IMAGE}", write=True)
|
|
run_debugfs(debugfs, image, f"write {local_file} {VERSION_PATH_IN_IMAGE}", write=True)
|
|
inode = parse_inode(run_debugfs(debugfs, image, f"stat {VERSION_PATH_IN_IMAGE}"))
|
|
for field, value in (("mode", "0100644"), ("uid", "0"), ("gid", "0")):
|
|
run_debugfs(debugfs, image, f"set_inode_field <{inode}> {field} {value}", write=True)
|
|
|
|
|
|
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 image_path_exists(debugfs: str, image: Path, image_path: str) -> bool:
|
|
output = run_debugfs(debugfs, image, f"stat {image_path}")
|
|
return re.search(r"Inode:\s+\d+", output) is not None
|
|
|
|
|
|
def list_image_directory(debugfs: str, image: Path, image_path: str) -> set[str]:
|
|
output = run_debugfs(debugfs, image, f"ls -p {image_path}")
|
|
entries: set[str] = set()
|
|
for line in output.splitlines():
|
|
if line.startswith("/"):
|
|
fields = line.split("/")
|
|
if len(fields) >= 6 and fields[5] not in ("", ".", ".."):
|
|
entries.add(fields[5])
|
|
return entries
|
|
|
|
|
|
def validate_venv_layout(debugfs: str, image: Path, *, expected_count: int,
|
|
required_paths: dict[str, str]) -> dict[str, object]:
|
|
entries = list_image_directory(debugfs, image, SITE_PACKAGES_PATH_IN_IMAGE)
|
|
if len(entries) != expected_count:
|
|
raise RuntimeError(f"Managed venv has {len(entries)} entries; expected {expected_count}")
|
|
missing = [name for name, path in required_paths.items() if not image_path_exists(debugfs, image, path)]
|
|
if missing:
|
|
raise RuntimeError(f"Managed venv is missing required imports: {', '.join(missing)}")
|
|
return {"site_packages_count": len(entries), "required_imports": sorted(required_paths)}
|
|
|
|
|
|
def image_path_is_directory(debugfs: str, image: Path, image_path: str) -> bool:
|
|
output = run_debugfs(debugfs, image, f"stat {image_path}")
|
|
if not re.search(r"Inode:\s+\d+", output):
|
|
raise RuntimeError(f"Image path does not exist: {image_path}")
|
|
return "Type: directory" in output
|
|
|
|
|
|
def extract_starpilot_dependencies(debugfs: str, dependency_image: Path, destination: Path) -> None:
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
for image_path in STAR_PILOT_DEPENDENCY_PATHS:
|
|
if not image_path_exists(debugfs, dependency_image, image_path):
|
|
raise RuntimeError(f"StarPilot dependency source is missing {image_path}")
|
|
if image_path_is_directory(debugfs, dependency_image, image_path):
|
|
run_debugfs(debugfs, dependency_image, f"rdump {image_path} {destination}")
|
|
else:
|
|
run_debugfs(debugfs, dependency_image, f"dump -p {image_path} {destination / Path(image_path).name}")
|
|
extracted = {path.name for path in destination.iterdir()}
|
|
expected = set(STAR_PILOT_DEPENDENCY_NAMES)
|
|
if extracted != expected:
|
|
raise RuntimeError(f"Unexpected StarPilot dependency extraction: {sorted(extracted)}")
|
|
|
|
|
|
def ensure_image_directory(debugfs: str, image: Path, image_path: str) -> None:
|
|
if image_path_exists(debugfs, image, image_path):
|
|
return
|
|
parent = str(Path(image_path).parent)
|
|
if parent not in ("", ".", "/"):
|
|
ensure_image_directory(debugfs, image, parent)
|
|
run_debugfs(debugfs, image, f"mkdir {image_path}", write=True)
|
|
if not image_path_exists(debugfs, image, image_path):
|
|
raise RuntimeError(f"Failed to create image directory {image_path}")
|
|
inode = parse_inode(run_debugfs(debugfs, image, f"stat {image_path}"))
|
|
for field, value in (("mode", "040755"), ("uid", "0"), ("gid", "0")):
|
|
run_debugfs(debugfs, image, f"set_inode_field <{inode}> {field} {value}", write=True)
|
|
|
|
|
|
def add_tree_to_image(debugfs: str, image: Path, source: Path, destination: str) -> None:
|
|
if image_path_exists(debugfs, image, destination):
|
|
raise RuntimeError(f"Refusing to overwrite upstream image path {destination}")
|
|
commands: list[str] = []
|
|
created_directories: set[str] = set()
|
|
|
|
def create_directory(image_path: str) -> None:
|
|
if image_path in created_directories or image_path == SITE_PACKAGES_PATH_IN_IMAGE:
|
|
return
|
|
parent = str(Path(image_path).parent)
|
|
if parent not in ("", ".", "/"):
|
|
create_directory(parent)
|
|
commands.extend((
|
|
f"mkdir {image_path}",
|
|
f"set_inode_field {image_path} mode 040755",
|
|
f"set_inode_field {image_path} uid 0",
|
|
f"set_inode_field {image_path} gid 0",
|
|
))
|
|
created_directories.add(image_path)
|
|
|
|
create_directory(destination)
|
|
for local_path in sorted(source.rglob("*")):
|
|
if "__pycache__" in local_path.parts:
|
|
continue
|
|
relative = local_path.relative_to(source)
|
|
image_path = str(Path(destination) / relative)
|
|
if local_path.is_dir():
|
|
create_directory(image_path)
|
|
continue
|
|
create_directory(str(Path(image_path).parent))
|
|
mode = "0100755" if local_path.stat().st_mode & 0o111 else "0100644"
|
|
commands.extend((
|
|
f"write {local_path} {image_path}",
|
|
f"set_inode_field {image_path} mode {mode}",
|
|
f"set_inode_field {image_path} uid 0",
|
|
f"set_inode_field {image_path} gid 0",
|
|
))
|
|
|
|
with tempfile.NamedTemporaryFile("w", encoding="utf-8") as command_file:
|
|
command_file.write("\n".join(commands) + "\n")
|
|
command_file.flush()
|
|
result = run_cmd([debugfs, "-w", "-f", command_file.name, str(image)])
|
|
output = f"{result.stdout}\n{result.stderr}"
|
|
if "File not found" in output or "Ext2 file already exists" in output:
|
|
raise RuntimeError(f"Failed to add StarPilot dependency tree {destination}:\n{output[-4000:]}")
|
|
if not image_path_exists(debugfs, image, destination):
|
|
raise RuntimeError(f"Failed to add StarPilot dependency tree {destination}")
|
|
|
|
|
|
def add_path_to_image(debugfs: str, image: Path, source: Path, destination: str) -> None:
|
|
if image_path_exists(debugfs, image, destination):
|
|
raise RuntimeError(f"Refusing to overwrite upstream image path {destination}")
|
|
if source.is_dir():
|
|
add_tree_to_image(debugfs, image, source, destination)
|
|
return
|
|
if not source.is_file():
|
|
raise RuntimeError(f"Extracted dependency path is missing: {source}")
|
|
ensure_image_directory(debugfs, image, str(Path(destination).parent))
|
|
run_debugfs(debugfs, image, f"write {source} {destination}", write=True)
|
|
if not image_path_exists(debugfs, image, destination):
|
|
raise RuntimeError(f"Failed to add StarPilot dependency file {destination}")
|
|
inode = parse_inode(run_debugfs(debugfs, image, f"stat {destination}"))
|
|
mode = "0100755" if source.stat().st_mode & 0o111 else "0100644"
|
|
for field, value in (("mode", mode), ("uid", "0"), ("gid", "0")):
|
|
run_debugfs(debugfs, image, f"set_inode_field <{inode}> {field} {value}", write=True)
|
|
|
|
|
|
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("/", "_")
|
|
run_debugfs(debugfs, image, f"dump -p {image_path} {local_path}")
|
|
fingerprints[image_path] = sha256_file(local_path)
|
|
return fingerprints
|
|
|
|
|
|
def validate_protected_payloads(actual: dict[str, str]) -> None:
|
|
differences = {
|
|
path: {"actual": actual.get(path), "expected": expected}
|
|
for path, expected in PROTECTED_PAYLOAD_HASHES.items()
|
|
if actual.get(path) != expected
|
|
}
|
|
if differences:
|
|
raise RuntimeError(f"Image is not exact upstream AGNOS: {json.dumps(differences, sort_keys=True)}")
|
|
|
|
|
|
def validate_ext4(e2fsck: str, image: Path) -> None:
|
|
result = run_cmd([e2fsck, "-fn", str(image)], allowed_returncodes=frozenset({0, 1, 2}))
|
|
output = f"{result.stdout}\n{result.stderr}"
|
|
if "UNEXPECTED INCONSISTENCY" in output or "Filesystem still has errors" in output:
|
|
raise RuntimeError(f"ext4 validation failed:\n{output}")
|
|
|
|
|
|
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 partial.open("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(result.stderr.decode("utf-8", "replace"))
|
|
partial.replace(destination)
|
|
|
|
|
|
def get_system_entry(manifest: list[dict]) -> dict:
|
|
return next(entry for entry in manifest if entry.get("name") == "system")
|
|
|
|
|
|
def update_manifest_system_entry(manifest: list[dict], new_url: str, raw_hash: str, size: int) -> list[dict]:
|
|
updated = json.loads(json.dumps(manifest))
|
|
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 main() -> int:
|
|
args = parse_args()
|
|
target_version = validate_target_version(args.set_version)
|
|
debugfs, e2fsck = find_debugfs(), find_e2fsck()
|
|
work_dir = Path(args.work_dir).resolve()
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
if args.source_image:
|
|
source = Path(args.source_image).resolve()
|
|
else:
|
|
source = work_dir / "upstream_system.img.xz"
|
|
if args.force_download:
|
|
source.unlink(missing_ok=True)
|
|
if not source.exists():
|
|
download(args.source_url, source)
|
|
if not source.is_file():
|
|
raise RuntimeError(f"Source image not found: {source}")
|
|
|
|
upstream_raw = work_dir / "upstream_system.ext4.img"
|
|
materialize_upstream_image(source, upstream_raw, work_dir)
|
|
if upstream_raw.stat().st_size != UPSTREAM_RAW_SIZE:
|
|
raise RuntimeError(f"Upstream raw size mismatch: {upstream_raw.stat().st_size}")
|
|
upstream_hash = sha256_file(upstream_raw)
|
|
if upstream_hash != UPSTREAM_RAW_SHA256:
|
|
raise RuntimeError(f"Upstream raw hash mismatch: {upstream_hash}")
|
|
validate_ext4(e2fsck, upstream_raw)
|
|
if read_image_text(debugfs, upstream_raw, VERSION_PATH_IN_IMAGE) != UPSTREAM_VERSION:
|
|
raise RuntimeError("Source image does not contain upstream /VERSION=19.6")
|
|
upstream_venv = validate_venv_layout(
|
|
debugfs, upstream_raw,
|
|
expected_count=UPSTREAM_SITE_PACKAGES_COUNT,
|
|
required_paths=UPSTREAM_REQUIRED_VENV_PATHS,
|
|
)
|
|
unexpected_dependency_paths = [
|
|
path for path in STAR_PILOT_DEPENDENCY_PATHS if image_path_exists(debugfs, upstream_raw, path)
|
|
]
|
|
if unexpected_dependency_paths:
|
|
raise RuntimeError(f"Dependency allowlist overlaps upstream AGNOS: {unexpected_dependency_paths}")
|
|
upstream_payloads = fingerprint_image_paths(debugfs, upstream_raw, tuple(PROTECTED_PAYLOAD_HASHES), work_dir, "upstream")
|
|
validate_protected_payloads(upstream_payloads)
|
|
|
|
c3_work_dir = work_dir / "c3_dependency_source"
|
|
c3_work_dir.mkdir(parents=True, exist_ok=True)
|
|
if args.c3_deps_image:
|
|
c3_source = Path(args.c3_deps_image).resolve()
|
|
else:
|
|
c3_source = c3_work_dir / "system.img.xz"
|
|
if args.force_download:
|
|
c3_source.unlink(missing_ok=True)
|
|
if not c3_source.exists():
|
|
download(args.c3_deps_url, c3_source)
|
|
if not c3_source.is_file():
|
|
raise RuntimeError(f"C3 dependency source image not found: {c3_source}")
|
|
c3_raw = c3_work_dir / "system.ext4.img"
|
|
materialize_upstream_image(c3_source, c3_raw, c3_work_dir)
|
|
if c3_raw.stat().st_size != C3_DEPENDENCY_SOURCE_RAW_SIZE:
|
|
raise RuntimeError(f"C3 dependency source size mismatch: {c3_raw.stat().st_size}")
|
|
c3_source_hash = sha256_file(c3_raw)
|
|
if c3_source_hash != C3_DEPENDENCY_SOURCE_RAW_SHA256:
|
|
raise RuntimeError(f"C3 dependency source hash mismatch: {c3_source_hash}")
|
|
dependency_packages_dir = work_dir / "starpilot_dependency_packages"
|
|
if dependency_packages_dir.exists():
|
|
shutil.rmtree(dependency_packages_dir)
|
|
extract_starpilot_dependencies(debugfs, c3_raw, dependency_packages_dir)
|
|
|
|
candidate_raw = work_dir / f"starpilot_system_{target_version}.ext4.img"
|
|
candidate_raw.unlink(missing_ok=True)
|
|
shutil.copy2(upstream_raw, candidate_raw)
|
|
version_file = work_dir / "VERSION.starpilot"
|
|
version_file.write_text(target_version + "\n", encoding="utf-8")
|
|
write_version(debugfs, candidate_raw, version_file)
|
|
for image_path in STAR_PILOT_DEPENDENCY_PATHS:
|
|
add_path_to_image(debugfs, candidate_raw, dependency_packages_dir / Path(image_path).name, image_path)
|
|
|
|
if read_image_text(debugfs, candidate_raw, VERSION_PATH_IN_IMAGE) != target_version:
|
|
raise RuntimeError("Failed to write the StarPilot AGNOS version marker")
|
|
candidate_venv = validate_venv_layout(
|
|
debugfs, candidate_raw,
|
|
expected_count=CANDIDATE_SITE_PACKAGES_COUNT,
|
|
required_paths=REQUIRED_VENV_PATHS,
|
|
)
|
|
candidate_payloads = fingerprint_image_paths(debugfs, candidate_raw, tuple(PROTECTED_PAYLOAD_HASHES), work_dir, "candidate")
|
|
validate_protected_payloads(candidate_payloads)
|
|
if candidate_payloads != upstream_payloads:
|
|
raise RuntimeError("Protected upstream payloads changed")
|
|
validate_ext4(e2fsck, candidate_raw)
|
|
|
|
raw_hash = sha256_file(candidate_raw)
|
|
output_xz = Path(args.output_xz).resolve() if args.output_xz else work_dir / f"system-{raw_hash}.img.xz"
|
|
compress_xz(candidate_raw, output_xz)
|
|
metadata = {
|
|
"base_version": UPSTREAM_VERSION,
|
|
"base_raw_sha256": UPSTREAM_RAW_SHA256,
|
|
"target_version": target_version,
|
|
"allowed_image_mutations": sorted(ALLOWED_IMAGE_MUTATIONS),
|
|
"raw_sha256": raw_hash,
|
|
"raw_size": candidate_raw.stat().st_size,
|
|
"xz_sha256": sha256_file(output_xz),
|
|
"xz_size": output_xz.stat().st_size,
|
|
"upstream_venv_validation": upstream_venv,
|
|
"candidate_venv_validation": candidate_venv,
|
|
"c3_dependency_source_raw_sha256": c3_source_hash,
|
|
"starpilot_dependency_paths": list(STAR_PILOT_DEPENDENCY_PATHS),
|
|
"c3_dependency_paths": list(C3_DEPENDENCY_PATHS),
|
|
"protected_payloads": candidate_payloads,
|
|
"factory_reset_stack": "byte-identical to upstream",
|
|
"device_validation_required": True,
|
|
}
|
|
metadata_path = Path(str(output_xz) + ".metadata.json")
|
|
metadata_path.write_text(json.dumps(metadata, indent=2) + "\n", encoding="utf-8")
|
|
|
|
print("Validated upstream-identical StarPilot AGNOS artifact:")
|
|
print(f" target version: {target_version}")
|
|
print(f" raw image: {candidate_raw}")
|
|
print(f" xz image: {output_xz}")
|
|
print(f" raw sha256: {raw_hash}")
|
|
print(f" xz sha256: {metadata['xz_sha256']}")
|
|
print(f" metadata: {metadata_path}")
|
|
print(" only mutations: /VERSION and additive StarPilot runtime/C3 dependencies")
|
|
|
|
if args.new_url:
|
|
manifest_path = Path(args.manifest).resolve()
|
|
output_manifest = Path(args.manifest_out).resolve() if args.manifest_out else work_dir / "agnos.candidate.json"
|
|
if output_manifest == manifest_path:
|
|
raise RuntimeError("Refusing to overwrite the checked-in manifest")
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
output_manifest.write_text(
|
|
json.dumps(update_manifest_system_entry(manifest, args.new_url, raw_hash, candidate_raw.stat().st_size), indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
print(f" candidate manifest: {output_manifest}")
|
|
elif args.manifest_out:
|
|
raise RuntimeError("--manifest-out requires --new-url")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|