"""Fixed release packaging and verification for the RF-001 RX agent. The module deliberately has no command execution, SDR access, or generic archive extraction. A release contains one exact allowlist of Python files and a content manifest. Approval metadata is authenticated separately with the already provisioned HTTP-channel HMAC secret. """ from __future__ import annotations import ast from dataclasses import dataclass import hashlib import hmac import json import os from pathlib import Path, PurePosixPath import re import shutil import stat from typing import Any, Mapping import zipfile UPDATE_MANIFEST_DOMAIN = b"RF001-UPDATE-MANIFEST-V1\n" CONTENT_MANIFEST_NAME = "RELEASE_CONTENT.json" RELEASE_ID_PREFIX = "rf001-agent-" RELEASE_ID_PATTERN = re.compile(r"^rf001-agent-[0-9]{4}\.[0-9]{2}\.[0-9]{2}\.[0-9]+$") AGENT_VERSION_PATTERN = re.compile(r"^[0-9]{4}\.[0-9]{2}\.[0-9]{2}\.[0-9]+$") SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") MAX_PACKAGE_BYTES = 64 * 1024 * 1024 MAX_UNCOMPRESSED_BYTES = 128 * 1024 * 1024 MAX_ARCHIVE_ENTRIES = 32 MAX_CONTENT_MANIFEST_BYTES = 64 * 1024 RELEASE_SOURCE_FILES: dict[str, str] = { "app/tools/hackrf_process_check.py": "tools/hackrf_process_check.py", "app/tools/rf001_agent_update.py": "tools/rf001_agent_update.py", "app/tools/rf001_capture_runner.py": "tools/rf001_capture_runner.py", "app/tools/rf001_http_control.py": "tools/rf001_http_control.py", "app/tools/rf001_offline_decode.py": "tools/rf001_offline_decode.py", "app/tools/rf001_rx_capture.py": "tools/rf001_rx_capture.py", "app/tools/rf001_rx_http_agent.py": "tools/rf001_rx_http_agent.py", "app/tools/rf001_staircase_job.py": "tools/rf001_staircase_job.py", "app/tools/rf001_staircase_search.py": "tools/rf001_staircase_search.py", } REQUIRED_RELEASE_FILES = frozenset(RELEASE_SOURCE_FILES) class ReleaseValidationError(ValueError): """A release is malformed, unauthenticated, or outside its allowlist.""" @dataclass(frozen=True) class ReleaseInspection: release_id: str agent_version: str protocol_version: int package_sha256: str package_bytes: int content_manifest_sha256: str uncompressed_bytes: int file_count: int def descriptor(self) -> dict[str, object]: return { "schema_version": 1, "release_id": self.release_id, "agent_version": self.agent_version, "protocol_version": self.protocol_version, "package_sha256": self.package_sha256, "package_bytes": self.package_bytes, "content_manifest_sha256": self.content_manifest_sha256, "uncompressed_bytes": self.uncompressed_bytes, "file_count": self.file_count, } def source_agent_version(agent_source: Path) -> str: """Read the single literal AGENT_VERSION from an agent source file.""" try: syntax = ast.parse(agent_source.read_text(encoding="utf-8"), agent_source.name) except (OSError, SyntaxError, UnicodeError) as error: raise ReleaseValidationError("agent source cannot be parsed") from error values: list[ast.expr] = [] for statement in syntax.body: if isinstance(statement, ast.Assign): if any( isinstance(target, ast.Name) and target.id == "AGENT_VERSION" for target in statement.targets ): if not ( len(statement.targets) == 1 and isinstance(statement.targets[0], ast.Name) and statement.targets[0].id == "AGENT_VERSION" ): raise ReleaseValidationError( "agent source AGENT_VERSION assignment is ambiguous" ) values.append(statement.value) elif ( isinstance(statement, ast.AnnAssign) and isinstance(statement.target, ast.Name) and statement.target.id == "AGENT_VERSION" ): if statement.value is None: raise ReleaseValidationError("agent source AGENT_VERSION has no value") values.append(statement.value) if len(values) != 1: raise ReleaseValidationError( "agent source must contain one AGENT_VERSION assignment" ) value = values[0] if not isinstance(value, ast.Constant) or not isinstance(value.value, str): raise ReleaseValidationError("agent source AGENT_VERSION must be a literal string") if AGENT_VERSION_PATTERN.fullmatch(value.value) is None: raise ReleaseValidationError("agent source AGENT_VERSION is invalid") return value.value def validate_release_identity( source_root: Path, *, release_id: str, requested_agent_version: str, ) -> str: """Require release id, requested version, and packaged source to agree.""" if RELEASE_ID_PATTERN.fullmatch(release_id) is None: raise ReleaseValidationError("release_id is invalid") if AGENT_VERSION_PATTERN.fullmatch(requested_agent_version) is None: raise ReleaseValidationError("agent_version is invalid") if release_id != f"{RELEASE_ID_PREFIX}{requested_agent_version}": raise ReleaseValidationError("release_id does not match agent_version") agent_source = ( source_root / RELEASE_SOURCE_FILES["app/tools/rf001_rx_http_agent.py"] ).resolve() actual_agent_version = source_agent_version(agent_source) if actual_agent_version != requested_agent_version: raise ReleaseValidationError( "source AGENT_VERSION does not match requested agent_version" ) return actual_agent_version def canonical_json_bytes(value: Mapping[str, Any]) -> bytes: return json.dumps( value, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ).encode("utf-8") def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: while block := stream.read(4 * 1024 * 1024): digest.update(block) return digest.hexdigest() def sign_update_manifest(token: bytes, manifest: Mapping[str, Any]) -> str: return hmac.new( token, UPDATE_MANIFEST_DOMAIN + canonical_json_bytes(manifest), hashlib.sha256, ).hexdigest() def verify_update_manifest_signature( token: bytes, manifest: Mapping[str, Any], signature: str, ) -> None: if SHA256_PATTERN.fullmatch(signature) is None: raise ReleaseValidationError("update manifest signature is invalid") expected = sign_update_manifest(token, manifest) if not hmac.compare_digest(expected, signature): raise ReleaseValidationError("update manifest signature mismatch") def _validate_member_name(name: str) -> PurePosixPath: if not name or "\\" in name or name.startswith("/"): raise ReleaseValidationError(f"unsafe archive path: {name!r}") path = PurePosixPath(name) if path.is_absolute() or any(part in ("", ".", "..") for part in path.parts): raise ReleaseValidationError(f"unsafe archive path: {name!r}") if any(":" in part for part in path.parts): raise ReleaseValidationError(f"drive-like archive path: {name!r}") return path def _is_link(info: zipfile.ZipInfo) -> bool: unix_mode = (info.external_attr >> 16) & 0xFFFF return stat.S_IFMT(unix_mode) == stat.S_IFLNK def _load_content_manifest( archive: zipfile.ZipFile, info: zipfile.ZipInfo, ) -> tuple[dict[str, Any], bytes]: if info.file_size > MAX_CONTENT_MANIFEST_BYTES: raise ReleaseValidationError("release content manifest is too large") raw = archive.read(info) try: decoded = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as error: raise ReleaseValidationError("release content manifest is invalid") from error if not isinstance(decoded, dict): raise ReleaseValidationError("release content manifest must be an object") if raw != canonical_json_bytes(decoded): raise ReleaseValidationError("release content manifest is not canonical JSON") return decoded, raw def _validate_content_manifest(manifest: Mapping[str, Any]) -> dict[str, dict[str, Any]]: if manifest.get("schema_version") != 1: raise ReleaseValidationError("unsupported release content schema") release_id = manifest.get("release_id") agent_version = manifest.get("agent_version") protocol_version = manifest.get("protocol_version") if not isinstance(release_id, str) or RELEASE_ID_PATTERN.fullmatch(release_id) is None: raise ReleaseValidationError("release_id is invalid") if not isinstance(agent_version, str) or AGENT_VERSION_PATTERN.fullmatch(agent_version) is None: raise ReleaseValidationError("agent_version is invalid") if type(protocol_version) is not int or protocol_version <= 0: raise ReleaseValidationError("protocol_version is invalid") rows = manifest.get("files") if not isinstance(rows, list): raise ReleaseValidationError("release files list is invalid") files: dict[str, dict[str, Any]] = {} for row in rows: if not isinstance(row, dict): raise ReleaseValidationError("release file entry is invalid") name = row.get("path") digest = row.get("sha256") size = row.get("bytes") if not isinstance(name, str): raise ReleaseValidationError("release file path is invalid") _validate_member_name(name) key = name.casefold() if key in (existing.casefold() for existing in files): raise ReleaseValidationError("duplicate release file path") if SHA256_PATTERN.fullmatch(str(digest)) is None: raise ReleaseValidationError(f"release file hash is invalid: {name}") if type(size) is not int or size < 0: raise ReleaseValidationError(f"release file size is invalid: {name}") files[name] = {"path": name, "sha256": digest, "bytes": size} if frozenset(files) != REQUIRED_RELEASE_FILES: unexpected = sorted(set(files) - REQUIRED_RELEASE_FILES) missing = sorted(REQUIRED_RELEASE_FILES - set(files)) raise ReleaseValidationError( f"release file allowlist mismatch; unexpected={unexpected}; missing={missing}" ) return files def inspect_release_archive( package_path: Path, *, expected_package_sha256: str | None = None, expected_package_bytes: int | None = None, expected_content_manifest_sha256: str | None = None, extraction_root: Path | None = None, ) -> ReleaseInspection: package_path = package_path.resolve() package_bytes = package_path.stat().st_size if package_bytes <= 0 or package_bytes > MAX_PACKAGE_BYTES: raise ReleaseValidationError("release package size is outside the limit") if expected_package_bytes is not None and package_bytes != expected_package_bytes: raise ReleaseValidationError("release package size mismatch") package_sha256 = sha256_file(package_path) if expected_package_sha256 is not None and package_sha256 != expected_package_sha256: raise ReleaseValidationError("release package SHA256 mismatch") with zipfile.ZipFile(package_path, "r") as archive: entries = archive.infolist() if not entries or len(entries) > MAX_ARCHIVE_ENTRIES: raise ReleaseValidationError("release archive entry count is invalid") seen: set[str] = set() regular: dict[str, zipfile.ZipInfo] = {} uncompressed_bytes = 0 for info in entries: path = _validate_member_name(info.filename) normalized = path.as_posix() key = normalized.casefold() if key in seen: raise ReleaseValidationError("ambiguous duplicate archive entry") seen.add(key) if info.flag_bits & 0x1: raise ReleaseValidationError("encrypted archive entries are prohibited") if _is_link(info): raise ReleaseValidationError("archive links are prohibited") if info.is_dir(): raise ReleaseValidationError("explicit archive directory entries are prohibited") uncompressed_bytes += info.file_size if uncompressed_bytes > MAX_UNCOMPRESSED_BYTES: raise ReleaseValidationError("release uncompressed size exceeds the limit") regular[normalized] = info if CONTENT_MANIFEST_NAME not in regular: raise ReleaseValidationError("release content manifest is absent") content_manifest, content_raw = _load_content_manifest( archive, regular[CONTENT_MANIFEST_NAME], ) content_sha256 = hashlib.sha256(content_raw).hexdigest() if ( expected_content_manifest_sha256 is not None and content_sha256 != expected_content_manifest_sha256 ): raise ReleaseValidationError("release content manifest SHA256 mismatch") files = _validate_content_manifest(content_manifest) expected_names = set(REQUIRED_RELEASE_FILES) | {CONTENT_MANIFEST_NAME} if set(regular) != expected_names: raise ReleaseValidationError("release archive contents are not allowlisted") for name, metadata in files.items(): info = regular[name] if info.file_size != metadata["bytes"]: raise ReleaseValidationError(f"release file size mismatch: {name}") digest = hashlib.sha256() with archive.open(info, "r") as stream: while block := stream.read(1024 * 1024): digest.update(block) if digest.hexdigest() != metadata["sha256"]: raise ReleaseValidationError(f"release file SHA256 mismatch: {name}") if extraction_root is not None: extraction_root = extraction_root.resolve() extraction_root.mkdir(parents=True, exist_ok=False) for name in sorted(REQUIRED_RELEASE_FILES): target = extraction_root.joinpath(*PurePosixPath(name).parts) target.parent.mkdir(parents=True, exist_ok=True) resolved = target.resolve() try: resolved.relative_to(extraction_root) except ValueError as error: raise ReleaseValidationError("archive path escaped extraction root") from error with archive.open(regular[name], "r") as source: with target.open("xb") as destination: shutil.copyfileobj(source, destination, length=1024 * 1024) manifest_target = extraction_root / CONTENT_MANIFEST_NAME with manifest_target.open("xb") as stream: stream.write(content_raw) return ReleaseInspection( release_id=str(content_manifest["release_id"]), agent_version=str(content_manifest["agent_version"]), protocol_version=int(content_manifest["protocol_version"]), package_sha256=package_sha256, package_bytes=package_bytes, content_manifest_sha256=content_sha256, uncompressed_bytes=uncompressed_bytes, file_count=len(files), ) def build_release_archive( source_root: Path, package_path: Path, *, release_id: str, agent_version: str, protocol_version: int, ) -> ReleaseInspection: if protocol_version <= 0: raise ReleaseValidationError("protocol_version is invalid") source_root = source_root.resolve() validate_release_identity( source_root, release_id=release_id, requested_agent_version=agent_version, ) rows = [] sources: dict[str, Path] = {} for archive_name, relative_source in RELEASE_SOURCE_FILES.items(): source = (source_root / relative_source).resolve() if not source.is_file(): raise FileNotFoundError(f"release source is absent: {source}") sources[archive_name] = source rows.append( { "path": archive_name, "bytes": source.stat().st_size, "sha256": sha256_file(source), } ) content = { "schema_version": 1, "release_id": release_id, "agent_version": agent_version, "protocol_version": protocol_version, "files": sorted(rows, key=lambda row: str(row["path"])), } content_raw = canonical_json_bytes(content) package_path.parent.mkdir(parents=True, exist_ok=True) if package_path.exists(): raise FileExistsError(f"release package already exists: {package_path}") timestamp = (2026, 9, 29, 0, 0, 0) with zipfile.ZipFile( package_path, "x", compression=zipfile.ZIP_DEFLATED, compresslevel=9, ) as archive: manifest_info = zipfile.ZipInfo(CONTENT_MANIFEST_NAME, timestamp) manifest_info.compress_type = zipfile.ZIP_DEFLATED manifest_info.external_attr = 0o100600 << 16 archive.writestr(manifest_info, content_raw) for name in sorted(sources): info = zipfile.ZipInfo(name, timestamp) info.compress_type = zipfile.ZIP_DEFLATED info.external_attr = 0o100600 << 16 archive.writestr(info, sources[name].read_bytes()) return inspect_release_archive(package_path) def atomic_write_json(path: Path, value: Mapping[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_bytes(canonical_json_bytes(value) + b"\n") os.replace(temporary, path)