186 lines
7.7 KiB
Python
186 lines
7.7 KiB
Python
#!/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)
|