#!/usr/bin/env python3 """Check whether a production PetaLinux build is reusable; never validates target media.""" import argparse import hashlib import json import os import posixpath import shutil import sys import tarfile import tempfile REQUIRED = ("BOOT.BIN", "image.ub", "rootfs.tar.bz2", "build_inputs.json", "summary.json", "MANIFEST.md") def digest(path): h = hashlib.sha256() with open(path, "rb") as stream: for part in iter(lambda: stream.read(1024 * 1024), b""): h.update(part) return h.hexdigest().upper() def fingerprint(obj): data = dict(obj) data.pop("fingerprint_sha256", None) encoded = json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(encoded.encode("utf-8")).hexdigest().upper() def load_json(path): with open(path, "r", encoding="utf-8") as stream: return json.load(stream) def checksum_entries(path): entries = {} with open(path, "r", encoding="utf-8") as stream: for lineno, raw in enumerate(stream, 1): line = raw.strip() if not line: continue parts = line.split(None, 1) if len(parts) != 2 or len(parts[0]) != 64 or any(c not in "0123456789abcdefABCDEF" for c in parts[0]): raise ValueError("invalid SHA256SUMS line {}".format(lineno)) name = parts[1].lstrip("* ").replace("\\", "/") chunks = name.split("/") if not name or os.path.isabs(name) or ".." in chunks or name == "SHA256SUMS" or name in entries: raise ValueError("unsafe or duplicate SHA256SUMS entry: {}".format(name)) entries[name] = parts[0].upper() return entries def check(current_path, delivery): root = os.path.abspath(delivery) errors = [] for name in REQUIRED + ("SHA256SUMS",): if not os.path.isfile(os.path.join(root, name)): errors.append("missing artifact: " + name) if errors: return {"reusable": False, "delivery": root, "errors": errors} try: current = load_json(current_path) stored = load_json(os.path.join(root, "build_inputs.json")) for label, obj in (("current", current), ("stored", stored)): if obj.get("schema_version") != 1 or fingerprint(obj) != obj.get("fingerprint_sha256"): errors.append(label + " fingerprint invalid") if current.get("fingerprint_sha256") != stored.get("fingerprint_sha256"): errors.append("build-defining inputs differ") except Exception as exc: errors.append("fingerprint check failed: " + str(exc)) try: sums = checksum_entries(os.path.join(root, "SHA256SUMS")) for name in REQUIRED: if name not in sums: errors.append("artifact absent from SHA256SUMS: " + name) elif digest(os.path.join(root, name)) != sums[name]: errors.append("checksum mismatch: " + name) for name, expected in sums.items(): path = os.path.realpath(os.path.join(root, name)) if os.path.commonpath((os.path.realpath(root), path)) != os.path.realpath(root) or not os.path.isfile(path) or digest(path) != expected: errors.append("invalid checksum entry: " + name) except Exception as exc: errors.append("checksum check failed: " + str(exc)) try: summary = load_json(os.path.join(root, "summary.json")) gates = summary.get("build_gates", summary.get("gates", {})) for gate in ("G0", "G1", "G2", "G3"): item = gates.get(gate, {}) if item.get("status") != "PASS": errors.append("build gate not PASS: " + gate) except Exception as exc: errors.append("build gate check failed: " + str(exc)) try: with tarfile.open(os.path.join(root, "rootfs.tar.bz2"), "r:bz2") as archive: init = None for member in archive: name = member.name clean = posixpath.normpath(name) if name.startswith("/") or clean == ".." or clean.startswith("../"): errors.append("unsafe rootfs archive path: " + name) break if member.issym() or member.islnk(): target = member.linkname resolved_link = posixpath.normpath(posixpath.join(posixpath.dirname(clean), target)) if not target.startswith("/") and (resolved_link == ".." or resolved_link.startswith("../")): errors.append("unsafe rootfs archive link: " + name) break if clean.lstrip("./") == "sbin/init": init = member if init is None or not (init.isfile() or init.issym() or init.islnk()): errors.append("rootfs archive lacks sbin/init") except Exception as exc: errors.append("rootfs archive check failed: " + str(exc)) return {"reusable": not errors, "delivery": root, "errors": errors} def self_test(): root = tempfile.mkdtemp(prefix="emmc-cache-test-") expected_parent = os.path.realpath(tempfile.gettempdir()) try: profile = {"schema_version": 1, "files": {}, "values": {"version": "2018.1"}} profile["fingerprint_sha256"] = fingerprint(profile) current = os.path.join(root, "current.json") delivery = os.path.join(root, "delivery") os.mkdir(delivery) with open(current, "w", encoding="utf-8") as stream: json.dump(profile, stream) with open(os.path.join(delivery, "build_inputs.json"), "w", encoding="utf-8") as stream: json.dump(profile, stream) for name in ("BOOT.BIN", "image.ub", "MANIFEST.md"): with open(os.path.join(delivery, name), "wb") as stream: stream.write(name.encode("ascii")) with tarfile.open(os.path.join(delivery, "rootfs.tar.bz2"), "w:bz2") as archive: import io info = tarfile.TarInfo("./sbin/init") info.size = 4 archive.addfile(info, io.BytesIO(b"init")) with open(os.path.join(delivery, "summary.json"), "w", encoding="utf-8") as stream: json.dump({"build_gates": {gate: {"status": "PASS"} for gate in ("G0", "G1", "G2", "G3")}}, stream) with open(os.path.join(delivery, "SHA256SUMS"), "w", encoding="utf-8") as stream: for name in REQUIRED: stream.write("{} {}\n".format(digest(os.path.join(delivery, name)), name)) if not check(current, delivery)["reusable"]: raise AssertionError("valid delivery rejected") with open(os.path.join(delivery, "image.ub"), "ab") as stream: stream.write(b"corrupt") if check(current, delivery)["reusable"]: raise AssertionError("corrupt delivery accepted") print(json.dumps({"self_test": "PASS"})) return 0 finally: resolved = os.path.realpath(root) if os.path.dirname(resolved) == expected_parent and os.path.basename(resolved).startswith("emmc-cache-test-"): shutil.rmtree(resolved) def main(): parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="action", required=True) p = sub.add_parser("check") p.add_argument("--current", required=True) p.add_argument("--delivery", required=True) sub.add_parser("self-test") args = parser.parse_args() if args.action == "self-test": return self_test() result = check(args.current, args.delivery) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result["reusable"] else 2 if __name__ == "__main__": try: sys.exit(main()) except Exception as exc: print(json.dumps({"error": str(exc)}), file=sys.stderr) sys.exit(2)