268 lines
11 KiB
Python
268 lines
11 KiB
Python
#!/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())
|