60-packaging/roundtrip_delivery.py
#!/usr/bin/env python3
"""Prüft und bündelt eindeutige Einzel-ZIPs für einen Roundtrip-Rücklauf."""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import shutil
import sys
import tempfile
import zipfile
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from deterministic_zip import FIXED_TIMESTAMP, build_deterministic_zip
class DeliveryError(RuntimeError):
pass
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _named(values: list[str], label: str) -> list[tuple[str, str]]:
result: list[tuple[str, str]] = []
for value in values:
if "=" not in value:
raise DeliveryError(f"{label}: REPOSITORY=PFAD erwartet: {value}")
repository, path = value.split("=", 1)
if not re.fullmatch(r"[A-Za-z0-9._-]+", repository) or not path:
raise DeliveryError(f"{label}: ungültiger Eintrag: {value}")
result.append((repository, path))
return result
def _safe_members(archive: zipfile.ZipFile, label: str) -> list[str]:
names: list[str] = []
for info in archive.infolist():
path = PurePosixPath(info.filename)
if not info.filename or "\\" in info.filename or path.is_absolute() or ".." in path.parts:
raise DeliveryError(f"{label}: unsicherer ZIP-Pfad: {info.filename}")
if info.filename in names:
raise DeliveryError(f"{label}: doppelter ZIP-Pfad: {info.filename}")
names.append(info.filename)
if archive.testzip() is not None:
raise DeliveryError(f"{label}: ZIP ist beschädigt")
return names
def _manifest_and_paths(path: Path, label: str) -> tuple[dict, set[str]]:
if not path.is_file() or not zipfile.is_zipfile(path):
raise DeliveryError(f"{label}: kein gültiges ZIP: {path}")
with zipfile.ZipFile(path) as archive:
names = _safe_members(archive, label)
candidates = [name for name in names if PurePosixPath(name).name == "manifest.json" and len(PurePosixPath(name).parts) <= 2]
if len(candidates) != 1:
raise DeliveryError(f"{label}: genau ein Root-manifest.json erwartet")
try:
manifest = json.loads(archive.read(candidates[0]).decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise DeliveryError(f"{label}: Manifest ist ungültig: {exc}") from exc
if not isinstance(manifest, dict):
raise DeliveryError(f"{label}: Manifest muss ein Objekt sein")
return manifest, set(names)
def _member_exists(member: str, paths: set[str]) -> bool:
return member in paths or any(path.endswith("/" + member) for path in paths)
def _delivery_id(value: str | None) -> str:
result = value or datetime.now(timezone.utc).strftime("RT-%Y%m%dT%H%M%SZ")
if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", result):
raise DeliveryError("delivery-id enthält unzulässige Zeichen")
return result
def create_delivery(
*, output_dir: Path, repository_specs: list[str], gap_path: Path,
expected_specs: list[str], marker_specs: list[str], delivery_id: str | None = None,
) -> dict:
delivery = _delivery_id(delivery_id)
if output_dir.exists():
raise DeliveryError(f"Ausgabeziel existiert bereits und wird nicht überschrieben: {output_dir}")
repositories = _named(repository_specs, "repository")
expected = _named(expected_specs, "expected")
markers = _named(marker_specs, "marker")
order = [name for name, _ in repositories]
if not order or len(set(order)) != len(order):
raise DeliveryError("repository: mindestens ein eindeutiges Repository erforderlich")
expected_by = {name: [] for name in order}
marker_by = {name: [] for name in order}
for name, member in expected:
if name not in expected_by:
raise DeliveryError(f"expected referenziert kein geliefertes Repository: {name}")
expected_by[name].append(member)
for name, member in markers:
if name not in marker_by:
raise DeliveryError(f"marker referenziert kein geliefertes Repository: {name}")
marker_by[name].append(member)
if any(not expected_by[name] or not marker_by[name] for name in order):
raise DeliveryError("Jedes Repository benötigt mindestens expected und marker")
gap_manifest, _ = _manifest_and_paths(gap_path, "semanticGap")
if gap_manifest.get("artifactType") != "semantic-gap" or gap_manifest.get("packageClass") != "installable":
raise DeliveryError("semanticGap: installierbares semantic-gap erwartet")
records: list[dict] = []
for name, raw in repositories:
path = Path(raw).expanduser().resolve()
manifest, paths = _manifest_and_paths(path, name)
if name not in {manifest.get("artifactType"), manifest.get("artifactName")}:
raise DeliveryError(f"{name}: Repository-Identität im Manifest stimmt nicht")
if manifest.get("packageClass") != "installable":
raise DeliveryError(f"{name}: packageClass installable erwartet")
for member in expected_by[name] + marker_by[name]:
if not _member_exists(member, paths):
raise DeliveryError(f"{name}: erwarteter Pfad fehlt im ausgelieferten ZIP: {member}")
archive = f"{delivery}--{name}.zip"
records.append({"repository": name, "source": path, "archive": archive, "sha256": _sha256(path),
"expectedFiles": sorted(set(expected_by[name])), "markers": sorted(set(marker_by[name]))})
gap_name = f"{delivery}--semantic-gap-000.zip"
stage_parent = output_dir.parent
stage_parent.mkdir(parents=True, exist_ok=True)
stage = Path(tempfile.mkdtemp(prefix=f".{output_dir.name}.staging-", dir=stage_parent))
try:
for record in records:
destination = stage / record["archive"]
shutil.copyfile(record["source"], destination)
if _sha256(destination) != record["sha256"]:
raise DeliveryError(f"{record['repository']}: kopiertes ZIP stimmt nicht mit der Prüfung überein")
shutil.copyfile(gap_path, stage / gap_name)
gap_hash = _sha256(stage / gap_name)
package_manifest = {
"artifactType": "roundtrip-import-package", "packageVersion": "1",
"repositoryOrder": order,
"repositories": {record["repository"]: {"archive": record["archive"], "sha256": record["sha256"]} for record in records},
"semanticGap": {"archive": gap_name, "sha256": gap_hash},
}
package_entries = {"roundtrip-package-manifest.json": (json.dumps(package_manifest, ensure_ascii=False, indent=2) + "\n").encode("utf-8")}
package_entries.update({record["archive"]: (stage / record["archive"]).read_bytes() for record in records})
package_entries[gap_name] = (stage / gap_name).read_bytes()
aggregate = stage / f"{delivery}--roundtrip-import-package.zip"
aggregate.write_bytes(build_deterministic_zip(package_entries))
delivery_manifest = {
"schemaVersion": "1.0", "artifactType": "roundtrip-delivery", "deliveryId": delivery,
"repositoryOrder": order, "repositories": [{key: value for key, value in record.items() if key != "source"} for record in records],
"semanticGap": {"archive": gap_name, "sha256": gap_hash},
"aggregatePackage": {"archive": aggregate.name, "sha256": _sha256(aggregate)},
}
(stage / "delivery-manifest.json").write_text(json.dumps(delivery_manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
if output_dir.exists():
raise DeliveryError("Ausgabeziel wurde während der Erzeugung angelegt")
stage.replace(output_dir)
except Exception:
shutil.rmtree(stage, ignore_errors=True)
raise
return delivery_manifest
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--output-dir", required=True, type=Path)
parser.add_argument("--repository", action="append", default=[], metavar="NAME=ZIP")
parser.add_argument("--gap", required=True, type=Path)
parser.add_argument("--expected", action="append", default=[], metavar="NAME=ZIP-PFAD")
parser.add_argument("--marker", action="append", default=[], metavar="NAME=ZIP-PFAD")
parser.add_argument("--delivery-id")
args = parser.parse_args()
try:
manifest = create_delivery(output_dir=args.output_dir, repository_specs=args.repository, gap_path=args.gap,
expected_specs=args.expected, marker_specs=args.marker, delivery_id=args.delivery_id)
except (DeliveryError, OSError, zipfile.BadZipFile) as exc:
print(f"FEHLER: {exc}", file=sys.stderr)
return 2
print(json.dumps(manifest, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())