NODEDC_MISSION_CORE/plugins/insta360-x4/packaging/fetch_sdk.py

#!/usr/bin/env python3
"""Build-time SDK acquisition. Never installed or executed on an operator Node.

Pin every byte from a public mirror; do not execute SDK code to inspect it.
The source lock is provenance, not a claim of vendor authenticity or licensing.
"""

import argparse
import hashlib
import json
import re
import shutil
import struct
import tempfile
import urllib.request
from pathlib import Path, PurePosixPath

ROOT = Path(__file__).resolve().parents[1]
LOCK = Path(__file__).with_name("sdk-lock.json")


def entries(lock):
    if lock["schema"] != "missioncore.insta360.sdk-source/v1":
        raise ValueError("Unsupported SDK lock")
    if not re.fullmatch(r"[0-9a-f]{40}", lock["commit"]):
        raise ValueError("SDK source must be an immutable commit")
    seen = set()
    for entry in lock["files"]:
        path = PurePosixPath(entry["path"])
        if (
            path.is_absolute()
            or ".." in path.parts
            or "\\" in entry["path"]
            or path.as_posix() != entry["path"]
            or entry["path"] in seen
            or not re.fullmatch(r"[0-9a-f]{64}", entry["sha256"])
            or not 0 < entry["bytes"] <= 32 * 1024 * 1024
        ):
            raise ValueError("Invalid SDK payload entry")
        seen.add(entry["path"])
        yield entry


def verified_bytes(data, entry):
    if len(data) != entry["bytes"] or hashlib.sha256(data).hexdigest() != entry["sha256"]:
        raise ValueError("SDK checksum mismatch: " + entry["path"])
    return data


def inspect_elf(path):
    """Parse ELF64 sections without loading the library (never ldd/dlopen)."""
    data = path.read_bytes()
    if (
        len(data) < 64
        or data[:7] != b"\x7fELF\x02\x01\x01"
        or struct.unpack_from("<H", data, 18)[0] != 62
    ):
        raise ValueError("SDK is not Linux x86_64 ELF64")
    offset = struct.unpack_from("<Q", data, 40)[0]
    size, count = struct.unpack_from("<HH", data, 58)
    if size != 64 or not 0 < count < 4096 or offset + count * size > len(data):
        raise ValueError("Invalid ELF section table")
    sections = [struct.unpack_from("<IIQQQQIIQQ", data, offset + i * size) for i in range(count)]
    needed = []
    for section in sections:
        if section[1] != 6:  # SHT_DYNAMIC
            continue
        if section[6] >= count:
            raise ValueError("Invalid ELF string table")
        strings = sections[section[6]]
        table = data[strings[4] : strings[4] + strings[5]]
        if section[4] + section[5] > len(data) or section[5] % 16:
            raise ValueError("Invalid ELF dynamic table")
        for at in range(section[4], section[4] + section[5], 16):
            tag, value = struct.unpack_from("<qQ", data, at)
            if tag == 1:  # DT_NEEDED
                end = table.find(b"\0", value)
                if end < value:
                    raise ValueError("Invalid ELF dependency name")
                needed.append(table[value:end].decode("ascii"))
    versions = sorted(
        set(x.decode() for x in re.findall(rb"(?:GLIBCXX|GLIBC|CXXABI)_[0-9.]+", data))
    )
    return {"architecture": "linux-x86_64", "needed": sorted(needed), "abi_versions": versions}


def verify(folder, lock):
    for entry in entries(lock):
        path = folder / entry["path"]
        if any(p.is_symlink() for p in (path, *path.parents)):
            raise ValueError("SDK payload cannot contain symlinks")
        verified_bytes(path.read_bytes(), entry)
    return inspect_elf(folder / "lib/libCameraSDK.so")


def fetch(destination, source=None):
    lock = json.loads(LOCK.read_text())
    if destination.exists():
        return verify(destination, lock)
    destination.parent.mkdir(parents=True, exist_ok=True)
    stage = Path(tempfile.mkdtemp(prefix=".sdk-", dir=destination.parent))
    try:
        for entry in entries(lock):
            if source:
                path = source / entry["path"]
                if path.is_symlink():
                    raise ValueError("SDK source cannot be a symlink")
                data = path.read_bytes()
            else:
                host = (
                    "https://media.githubusercontent.com/media/"
                    if entry["lfs"]
                    else "https://raw.githubusercontent.com/"
                )
                repo = lock["repository"].removeprefix("https://github.com/")
                url = host + "/".join((repo, lock["commit"], lock["directory"], entry["path"]))
                with urllib.request.urlopen(url, timeout=45) as response:
                    data = response.read(entry["bytes"] + 1)
            target = stage / entry["path"]
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(verified_bytes(data, entry))
            target.chmod(0o644)
        report = verify(stage, lock)
        stage.rename(destination)
        return report
    finally:
        if stage.exists():
            shutil.rmtree(stage)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--destination", type=Path, default=ROOT / "build/sdk")
    parser.add_argument(
        "--source", type=Path, help="Import an already acquired SDK using the same lock"
    )
    parser.add_argument(
        "--check", action="store_true", help="Verify existing payload without network access"
    )
    args = parser.parse_args()
    result = (
        verify(args.destination, json.loads(LOCK.read_text()))
        if args.check
        else fetch(args.destination, args.source)
    )
    print(json.dumps(result, indent=2))