#!/usr/bin/env python3 """Capture build-defining inputs and validate reuse of a QSPI delivery.""" from __future__ import print_function import argparse import hashlib import json import os import shutil import sys import tempfile REQUIRED_ARTIFACTS = ( "BOOT.BIN", "image.ub.qspi", "flash_layout.json", "summary.json", "build_inputs.json", "MANIFEST.md", "SHA256SUMS", ) def sha256_file(path): digest = hashlib.sha256() with open(path, "rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest().upper() def parse_assignments(values, option): parsed = {} for item in values: if "=" not in item: raise ValueError("{} requires NAME=VALUE: {}".format(option, item)) name, value = item.split("=", 1) name = name.strip() if not name or name in parsed: raise ValueError("invalid or duplicate {} name: {}".format(option, name)) parsed[name] = value return parsed def canonical_fingerprint(payload): base = dict(payload) base.pop("fingerprint_sha256", None) encoded = json.dumps(base, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(encoded.encode("utf-8")).hexdigest().upper() def capture(args): profile = os.path.abspath(args.board_profile) if not os.path.isfile(profile): raise ValueError("board profile not found: {}".format(profile)) inputs = parse_assignments(args.input, "--input") values = parse_assignments(args.value, "--value") files = {} for name in sorted(inputs): path = os.path.abspath(inputs[name]) if not os.path.isfile(path): raise ValueError("input file not found for {}: {}".format(name, path)) files[name] = { "path_hint": os.path.basename(path), "bytes": os.path.getsize(path), "sha256": sha256_file(path), } payload = { "schema_version": 1, "board_profile": { "name": os.path.splitext(os.path.basename(profile))[0], "sha256": sha256_file(profile), }, "files": files, "values": dict(sorted(values.items())), } payload["fingerprint_sha256"] = canonical_fingerprint(payload) output = os.path.abspath(args.output) parent = os.path.dirname(output) if parent and not os.path.isdir(parent): os.makedirs(parent) with open(output, "w") as stream: json.dump(payload, stream, indent=2, sort_keys=True) stream.write("\n") print(json.dumps({"captured": output, "fingerprint": payload["fingerprint_sha256"]})) return 0 def read_json(path): with open(path, "r") as stream: return json.load(stream) def parse_checksums(path): entries = {} with open(path, "r") as stream: for number, 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: raise ValueError("invalid SHA256SUMS line {}".format(number)) name = parts[1].lstrip("* ") pieces = name.replace("\\", "/").split("/") if name == "SHA256SUMS" or os.path.isabs(name) or ".." in pieces: raise ValueError("unsafe checksum entry: {}".format(name)) entries[name] = parts[0].upper() return entries def validate_layout(delivery, errors): try: layout = read_json(os.path.join(delivery, "flash_layout.json")) total = int(layout["flash"]["total_bytes"]) erase = int(layout["flash"]["erase_bytes"]) regions = [] for part in layout["partitions"]: offset = int(str(part["offset"]), 0) size = int(str(part["size"]), 0) if offset % erase or size % erase: errors.append("partition not erase-aligned: {}".format(part["name"])) if size <= 0 or offset < 0 or offset + size > total: errors.append("partition out of flash bounds: {}".format(part["name"])) regions.append((offset, offset + size, part["name"])) regions.sort() for left, right in zip(regions, regions[1:]): if left[1] > right[0]: errors.append("overlapping partitions: {} and {}".format(left[2], right[2])) image = layout["image"] actual = os.path.getsize(os.path.join(delivery, image["file"])) declared = int(image["bytes"]) if actual != declared: errors.append("image byte count differs: actual {} declared {}".format(actual, declared)) image_offset = int(str(image["flash_offset"]), 0) erase_length = int(str(image["erase_length"]), 0) if image_offset % erase or erase_length % erase or erase_length < actual: errors.append("image erase range is invalid or unaligned") if not [r for r in regions if r[0] <= image_offset and image_offset + erase_length <= r[1]]: errors.append("image erase range is not contained in a partition") except Exception as exc: errors.append("flash layout validation failed: {}".format(exc)) def check(args): delivery = os.path.abspath(args.delivery) errors = [] for name in REQUIRED_ARTIFACTS: if not os.path.isfile(os.path.join(delivery, name)): errors.append("missing artifact: {}".format(name)) if errors: print(json.dumps({"reusable": False, "errors": errors}, indent=2)) return 2 try: current = read_json(args.current) stored = read_json(os.path.join(delivery, "build_inputs.json")) if canonical_fingerprint(current) != current.get("fingerprint_sha256"): errors.append("current input fingerprint is internally inconsistent") if canonical_fingerprint(stored) != stored.get("fingerprint_sha256"): errors.append("stored input fingerprint is internally inconsistent") if current.get("fingerprint_sha256") != stored.get("fingerprint_sha256"): errors.append("build-defining inputs differ") except Exception as exc: errors.append("input fingerprint validation failed: {}".format(exc)) try: sums = parse_checksums(os.path.join(delivery, "SHA256SUMS")) for name in REQUIRED_ARTIFACTS: if name == "SHA256SUMS": continue expected = sums.get(name) if not expected: errors.append("artifact absent from SHA256SUMS: {}".format(name)) elif sha256_file(os.path.join(delivery, name)) != expected: errors.append("checksum mismatch: {}".format(name)) except Exception as exc: errors.append("checksum validation failed: {}".format(exc)) validate_layout(delivery, errors) try: summary = read_json(os.path.join(delivery, "summary.json")) for gate in ("G0", "G1", "G2", "G3"): if summary.get("gates", {}).get(gate, {}).get("status") != "PASS": errors.append("prior build gate is not PASS: {}".format(gate)) except Exception as exc: errors.append("summary validation failed: {}".format(exc)) print(json.dumps({"reusable": not errors, "delivery": delivery, "errors": errors}, indent=2, sort_keys=True)) return 0 if not errors else 2 def self_test(_args): root = tempfile.mkdtemp(prefix="qspi-cache-test-") try: profile = os.path.join(root, "board.md") source = os.path.join(root, "system.hdf") current = os.path.join(root, "current.json") delivery = os.path.join(root, "delivery") os.makedirs(delivery) with open(profile, "w") as stream: stream.write("board\n") with open(source, "wb") as stream: stream.write(b"hdf") capture(argparse.Namespace(board_profile=profile, input=["HDF=" + source], value=["PETALINUX_VERSION=2018.1"], output=current)) payload = read_json(current) if payload["fingerprint_sha256"] != canonical_fingerprint(payload): raise RuntimeError("fingerprint self-test failed") shutil.copyfile(current, os.path.join(delivery, "build_inputs.json")) with open(os.path.join(delivery, "BOOT.BIN"), "wb") as stream: stream.write(b"boot") with open(os.path.join(delivery, "image.ub.qspi"), "wb") as stream: stream.write(b"image") with open(os.path.join(delivery, "MANIFEST.md"), "w") as stream: stream.write("# test\n") with open(os.path.join(delivery, "flash_layout.json"), "w") as stream: json.dump({ "flash": {"total_bytes": 262144, "erase_bytes": 65536}, "partitions": [ {"name": "boot", "offset": "0x0", "size": "0x10000"}, {"name": "kernel", "offset": "0x10000", "size": "0x30000"}, ], "image": {"file": "image.ub.qspi", "bytes": 5, "flash_offset": "0x10000", "erase_length": "0x10000"}, }, stream) with open(os.path.join(delivery, "summary.json"), "w") as stream: json.dump({"gates": dict((g, {"status": "PASS"}) for g in ("G0", "G1", "G2", "G3"))}, stream) checksum_names = [name for name in REQUIRED_ARTIFACTS if name != "SHA256SUMS"] with open(os.path.join(delivery, "SHA256SUMS"), "w") as stream: for name in checksum_names: stream.write("{} {}\n".format(sha256_file(os.path.join(delivery, name)), name)) if check(argparse.Namespace(current=current, delivery=delivery)) != 0: raise RuntimeError("valid delivery self-test failed") with open(os.path.join(delivery, "image.ub.qspi"), "ab") as stream: stream.write(b"corrupt") if check(argparse.Namespace(current=current, delivery=delivery)) == 0: raise RuntimeError("corrupt delivery self-test failed") print(json.dumps({"self_test": "PASS"})) return 0 finally: shutil.rmtree(root, ignore_errors=True) def main(): parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="command") capture_parser = sub.add_parser("capture", help="capture build-defining inputs") capture_parser.add_argument("--board-profile", required=True) capture_parser.add_argument("--input", action="append", default=[]) capture_parser.add_argument("--value", action="append", default=[]) capture_parser.add_argument("--output", required=True) capture_parser.set_defaults(func=capture) check_parser = sub.add_parser("check", help="check whether a delivery is reusable") check_parser.add_argument("--current", required=True) check_parser.add_argument("--delivery", required=True) check_parser.set_defaults(func=check) test_parser = sub.add_parser("self-test", help="run a non-destructive internal test") test_parser.set_defaults(func=self_test) args = parser.parse_args() if not hasattr(args, "func"): parser.print_help() return 2 try: return args.func(args) except Exception as exc: print(json.dumps({"error": str(exc)}), file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())