export_archive.py #1

#!/usr/bin/env python3
"""Export an operator-identified deployment archive without changing its bytes.

Windows members are recorded as excluded, never extracted or executed. Only
regular files/directories inside the single sdp/ tree are accepted.
"""
import argparse
import hashlib
import io
import json
from pathlib import Path, PurePosixPath
import tarfile


def export(archive, destination, confirmed=False):
    if destination.exists():
        raise ValueError("Baseline destination must be new")
    data = archive.read_bytes()
    digest = hashlib.sha256(data).hexdigest()
    manifest, excluded, names, contents = [], [], set(), []
    with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as source:
        for member in source:
            path = PurePosixPath(member.name)
            if path.is_absolute() or ".." in path.parts or not path.parts or path.parts[0] != "sdp":
                raise ValueError("Unsafe archive path: " + member.name)
            if not (member.isfile() or member.isdir()):
                raise ValueError("Archive links/devices are unsupported: " + member.name)
            if member.isdir():
                continue
            relative = PurePosixPath(*path.parts[1:]).as_posix()
            if relative in names or relative in (".", ".export-manifest.json", ".baseline-info.json"):
                raise ValueError("Duplicate or reserved archive path: " + member.name)
            names.add(relative)
            if relative.startswith("Server/Windows/"):
                excluded.append(relative)
                continue
            contents.append((relative, source.extractfile(member).read(), member.mode))
    if not contents:
        raise ValueError("Archive contains no testable files")
    destination.mkdir(parents=True)
    for relative, content, mode in contents:
        target = destination / relative
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(content)
        target.chmod(0o755 if mode & 0o111 else 0o644)
        manifest.append({"path": relative, "source_kind": "deployment_archive", "action": "",
                         "archive_sha256": digest, "sha256": hashlib.sha256(content).hexdigest(), "symlink": None})
    manifest.sort(key=lambda row: row["path"])
    (destination / ".export-manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
    (destination / ".baseline-info.json").write_text(json.dumps({
        "source_kind": "deployment_archive", "archive": str(archive.resolve()), "archive_sha256": digest,
        "production_baseline_confirmed": confirmed,
        "production_equivalence": "Operator identified this exact archive; individual host drift is not measured" if confirmed else "Unconfirmed archive",
        "excluded_windows_files": sorted(excluded), "files": len(manifest)}, indent=2) + "\n")
    return manifest


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("archive", type=Path)
    parser.add_argument("destination", type=Path)
    parser.add_argument("--operator-confirmed", action="store_true", help="Use only after the operator identifies this exact deployed archive")
    args = parser.parse_args()
    try:
        rows = export(args.archive, args.destination, args.operator_confirmed)
    except (OSError, ValueError, tarfile.TarError) as error:
        parser.exit(1, str(error) + "\n")
    print("Exported %d deployment archive files" % len(rows))


if __name__ == "__main__":
    main()