276 lines
12 KiB
Python
276 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path, PurePosixPath
|
|
import shutil
|
|
import tarfile
|
|
import tempfile
|
|
from typing import Any, BinaryIO
|
|
import zipfile
|
|
|
|
from .versioning import SoftwareVersion
|
|
|
|
PRODUCT_ID = "matrix-screen-controller-walnutpi"
|
|
PACKAGE_FORMAT_VERSION = 1
|
|
MAX_OTA_UPLOAD_BYTES = 256 * 1024 * 1024
|
|
MAX_OTA_PAYLOAD_BYTES = 256 * 1024 * 1024
|
|
MAX_OTA_EXPANDED_BYTES = 1024 * 1024 * 1024
|
|
_MAX_MANIFEST_BYTES = 64 * 1024
|
|
_PACKAGE_MEMBERS = {"manifest.json", "payload.tar.gz"}
|
|
_NATIVE_BUILD_OUTPUTS = {
|
|
"app/display/native/libh618_hub75.so",
|
|
"app/display/native/hub75_benchmark",
|
|
"app/display/native/hub75_native_test",
|
|
"app/display/native/hub75_safeoff",
|
|
}
|
|
_EXCLUDED_PARTS = {".venv", "data", "__pycache__", ".pytest_cache", "node_modules"}
|
|
|
|
|
|
class OtaPackageError(ValueError):
|
|
"""The uploaded archive is not a supported complete release."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OtaPackageInfo:
|
|
path: Path
|
|
target_version: SoftwareVersion
|
|
created_at: str
|
|
release_notes: str
|
|
payload_sha256: str
|
|
payload_bytes: int
|
|
expanded_bytes: int
|
|
|
|
def document(self) -> dict[str, Any]:
|
|
return {
|
|
"target_version": str(self.target_version),
|
|
"created_at": self.created_at,
|
|
"release_notes": self.release_notes,
|
|
"payload_sha256": self.payload_sha256,
|
|
"payload_bytes": self.payload_bytes,
|
|
"expanded_bytes": self.expanded_bytes,
|
|
}
|
|
|
|
|
|
def _validated_manifest(raw: bytes) -> tuple[dict[str, Any], SoftwareVersion]:
|
|
if len(raw) > _MAX_MANIFEST_BYTES:
|
|
raise OtaPackageError("OTA manifest is too large")
|
|
try:
|
|
manifest = json.loads(raw.decode("utf-8"))
|
|
except (UnicodeError, json.JSONDecodeError) as exc:
|
|
raise OtaPackageError("OTA manifest is not valid UTF-8 JSON") from exc
|
|
expected = {
|
|
"format_version",
|
|
"product",
|
|
"software_version",
|
|
"created_at",
|
|
"release_notes",
|
|
"payload_sha256",
|
|
"payload_bytes",
|
|
"expanded_bytes",
|
|
}
|
|
if not isinstance(manifest, dict) or set(manifest) != expected:
|
|
raise OtaPackageError("OTA manifest fields do not match format version 1")
|
|
if manifest["format_version"] != PACKAGE_FORMAT_VERSION:
|
|
raise OtaPackageError("OTA package format is unsupported")
|
|
if manifest["product"] != PRODUCT_ID:
|
|
raise OtaPackageError("OTA package is for a different product")
|
|
try:
|
|
version = SoftwareVersion.parse(manifest["software_version"])
|
|
except ValueError as exc:
|
|
raise OtaPackageError(str(exc)) from exc
|
|
if not isinstance(manifest["created_at"], str) or not manifest["created_at"].strip():
|
|
raise OtaPackageError("OTA package created_at is missing")
|
|
if not isinstance(manifest["release_notes"], str) or len(manifest["release_notes"]) > 4000:
|
|
raise OtaPackageError("OTA release notes are invalid")
|
|
digest = manifest["payload_sha256"]
|
|
if not isinstance(digest, str) or len(digest) != 64 or any(c not in "0123456789abcdef" for c in digest):
|
|
raise OtaPackageError("OTA payload SHA-256 is invalid")
|
|
for key, maximum in (
|
|
("payload_bytes", MAX_OTA_PAYLOAD_BYTES),
|
|
("expanded_bytes", MAX_OTA_EXPANDED_BYTES),
|
|
):
|
|
value = manifest[key]
|
|
if type(value) is not int or value <= 0 or value > maximum:
|
|
raise OtaPackageError(f"OTA {key} is outside the supported limit")
|
|
return manifest, version
|
|
|
|
|
|
def _payload_digest(handle: BinaryIO) -> tuple[str, int]:
|
|
digest = hashlib.sha256()
|
|
total = 0
|
|
while True:
|
|
chunk = handle.read(1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
total += len(chunk)
|
|
if total > MAX_OTA_PAYLOAD_BYTES:
|
|
raise OtaPackageError("OTA payload exceeds the supported limit")
|
|
digest.update(chunk)
|
|
return digest.hexdigest(), total
|
|
|
|
|
|
def inspect_package(path: Path, *, current_version: SoftwareVersion | None = None) -> OtaPackageInfo:
|
|
package_path = Path(path)
|
|
try:
|
|
size = package_path.stat().st_size
|
|
except OSError as exc:
|
|
raise OtaPackageError("OTA package cannot be read") from exc
|
|
if size <= 0 or size > MAX_OTA_UPLOAD_BYTES:
|
|
raise OtaPackageError("OTA package exceeds the 256 MiB upload limit")
|
|
try:
|
|
with zipfile.ZipFile(package_path, "r") as archive:
|
|
entries = archive.infolist()
|
|
if {entry.filename for entry in entries} != _PACKAGE_MEMBERS or len(entries) != 2:
|
|
raise OtaPackageError("OTA package must contain only manifest.json and payload.tar.gz")
|
|
if any(entry.is_dir() or entry.flag_bits & 0x1 for entry in entries):
|
|
raise OtaPackageError("OTA package entries must be unencrypted files")
|
|
manifest_entry = archive.getinfo("manifest.json")
|
|
payload_entry = archive.getinfo("payload.tar.gz")
|
|
if manifest_entry.file_size > _MAX_MANIFEST_BYTES:
|
|
raise OtaPackageError("OTA manifest is too large")
|
|
if payload_entry.file_size > MAX_OTA_PAYLOAD_BYTES:
|
|
raise OtaPackageError("OTA payload exceeds the supported limit")
|
|
manifest, target_version = _validated_manifest(archive.read(manifest_entry))
|
|
with archive.open(payload_entry, "r") as payload:
|
|
digest, payload_bytes = _payload_digest(payload)
|
|
except (OtaPackageError, zipfile.BadZipFile):
|
|
raise
|
|
except (KeyError, OSError, RuntimeError) as exc:
|
|
raise OtaPackageError("OTA package cannot be inspected") from exc
|
|
if payload_bytes != manifest["payload_bytes"] or digest != manifest["payload_sha256"]:
|
|
raise OtaPackageError("OTA payload checksum or size does not match the manifest")
|
|
if current_version is not None and target_version <= current_version:
|
|
raise OtaPackageError(
|
|
f"OTA target {target_version} must be newer than installed version {current_version}"
|
|
)
|
|
return OtaPackageInfo(
|
|
path=package_path,
|
|
target_version=target_version,
|
|
created_at=manifest["created_at"],
|
|
release_notes=manifest["release_notes"],
|
|
payload_sha256=digest,
|
|
payload_bytes=payload_bytes,
|
|
expanded_bytes=manifest["expanded_bytes"],
|
|
)
|
|
|
|
|
|
def _safe_member_path(name: str) -> Path:
|
|
pure = PurePosixPath(name)
|
|
if pure.is_absolute() or not pure.parts or any(part in {"", ".", ".."} for part in pure.parts):
|
|
raise OtaPackageError(f"unsafe OTA payload path: {name}")
|
|
if pure.parts[0] not in {"software", "wheelhouse"}:
|
|
raise OtaPackageError(f"unexpected OTA payload root: {pure.parts[0]}")
|
|
return Path(*pure.parts)
|
|
|
|
|
|
def extract_payload(package: OtaPackageInfo, destination: Path) -> None:
|
|
target = Path(destination)
|
|
target.mkdir(parents=True, exist_ok=False)
|
|
expanded = 0
|
|
try:
|
|
with zipfile.ZipFile(package.path, "r") as archive:
|
|
with archive.open("payload.tar.gz", "r") as payload:
|
|
with tarfile.open(fileobj=payload, mode="r|gz") as tar:
|
|
for member in tar:
|
|
relative = _safe_member_path(member.name)
|
|
output = target / relative
|
|
if member.isdir():
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
if not member.isfile():
|
|
raise OtaPackageError("OTA payload may contain only regular files and directories")
|
|
expanded += member.size
|
|
if expanded > MAX_OTA_EXPANDED_BYTES or expanded > package.expanded_bytes:
|
|
raise OtaPackageError("OTA payload expands beyond its declared limit")
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
source = tar.extractfile(member)
|
|
if source is None:
|
|
raise OtaPackageError(f"OTA payload member cannot be read: {member.name}")
|
|
with output.open("xb") as handle:
|
|
shutil.copyfileobj(source, handle, length=1024 * 1024)
|
|
output.chmod(member.mode & 0o777 or 0o644)
|
|
if expanded != package.expanded_bytes:
|
|
raise OtaPackageError("OTA expanded size does not match the manifest")
|
|
if not (target / "software" / "VERSION").is_file():
|
|
raise OtaPackageError("OTA payload does not contain software/VERSION")
|
|
if not (target / "wheelhouse" / "SHA256SUMS").is_file():
|
|
raise OtaPackageError("OTA payload does not contain the offline wheel manifest")
|
|
except BaseException:
|
|
shutil.rmtree(target, ignore_errors=True)
|
|
raise
|
|
|
|
|
|
def _source_file_allowed(path: Path, source_root: Path) -> bool:
|
|
relative = path.relative_to(source_root)
|
|
if any(part in _EXCLUDED_PARTS for part in relative.parts):
|
|
return False
|
|
if relative.as_posix() in _NATIVE_BUILD_OUTPUTS:
|
|
return False
|
|
if path.suffix in {".pyc", ".pyo"}:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _tar_add_file(tar: tarfile.TarFile, source: Path, archive_name: str) -> int:
|
|
info = tar.gettarinfo(str(source), arcname=archive_name)
|
|
info.uid = info.gid = 0
|
|
info.uname = info.gname = "root"
|
|
info.mtime = 0
|
|
info.mode = 0o755 if source.suffix == ".sh" else 0o644
|
|
with source.open("rb") as handle:
|
|
tar.addfile(info, handle)
|
|
return info.size
|
|
|
|
|
|
def build_package(
|
|
source_root: Path,
|
|
wheelhouse: Path,
|
|
output_path: Path,
|
|
*,
|
|
version: SoftwareVersion,
|
|
release_notes: str,
|
|
created_at: datetime | None = None,
|
|
) -> OtaPackageInfo:
|
|
source_root = Path(source_root).resolve()
|
|
wheelhouse = Path(wheelhouse).resolve()
|
|
if SoftwareVersion.parse((source_root / "VERSION").read_text(encoding="utf-8")) != version:
|
|
raise OtaPackageError("requested package version does not match source VERSION")
|
|
if not (wheelhouse / "SHA256SUMS").is_file():
|
|
raise OtaPackageError("offline wheelhouse SHA256SUMS is missing")
|
|
output = Path(output_path)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
if output.exists():
|
|
raise FileExistsError(output)
|
|
timestamp = (created_at or datetime.now(timezone.utc)).astimezone().isoformat(timespec="seconds")
|
|
with tempfile.TemporaryDirectory(prefix="matrix-ota-build-") as temporary:
|
|
payload_path = Path(temporary) / "payload.tar.gz"
|
|
expanded = 0
|
|
with tarfile.open(payload_path, "w:gz", format=tarfile.PAX_FORMAT) as tar:
|
|
for path in sorted(source_root.rglob("*"), key=lambda item: item.relative_to(source_root).as_posix()):
|
|
if path.is_file() and _source_file_allowed(path, source_root):
|
|
expanded += _tar_add_file(tar, path, f"software/{path.relative_to(source_root).as_posix()}")
|
|
for path in sorted(wheelhouse.rglob("*"), key=lambda item: item.relative_to(wheelhouse).as_posix()):
|
|
if path.is_file():
|
|
expanded += _tar_add_file(tar, path, f"wheelhouse/{path.relative_to(wheelhouse).as_posix()}")
|
|
payload_bytes = payload_path.stat().st_size
|
|
if payload_bytes > MAX_OTA_PAYLOAD_BYTES or expanded > MAX_OTA_EXPANDED_BYTES:
|
|
raise OtaPackageError("generated OTA payload exceeds the supported limit")
|
|
digest = hashlib.sha256(payload_path.read_bytes()).hexdigest()
|
|
manifest = {
|
|
"format_version": PACKAGE_FORMAT_VERSION,
|
|
"product": PRODUCT_ID,
|
|
"software_version": str(version),
|
|
"created_at": timestamp,
|
|
"release_notes": str(release_notes).strip(),
|
|
"payload_sha256": digest,
|
|
"payload_bytes": payload_bytes,
|
|
"expanded_bytes": expanded,
|
|
}
|
|
with zipfile.ZipFile(output, "x", compression=zipfile.ZIP_STORED, allowZip64=True) as archive:
|
|
archive.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2) + "\n")
|
|
archive.write(payload_path, "payload.tar.gz")
|
|
return inspect_package(output)
|