add petalinux qspi transition workflow
This commit is contained in:
bowei
@@ -0,0 +1,267 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user