Add shared PetaLinux eMMC and HJX9009 TF production skill
This commit is contained in:
bowei
@@ -0,0 +1,185 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user