20-roundtrip/changed_only_import.py
#!/usr/bin/env python3
"""Sicherer Changed-only-Roundtrip-Import für vollständige Repository-ZIPs."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import shutil
import tempfile
import zipfile
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any
class ImportContractError(RuntimeError):
pass
@dataclass
class PreparedImport:
archive: Path
repository: str
manifest: dict[str, Any]
deleted_files: list[str]
target: Path
@dataclass
class ImportResult:
archive: str
artifact_type: str
artifact_name: str
version: str
status: str
target: str | None = None
archived_previous: str | None = None
message: str = ""
def _safe_members(zf: zipfile.ZipFile) -> list[zipfile.ZipInfo]:
members: list[zipfile.ZipInfo] = []
for info in zf.infolist():
name = info.filename.replace("\\", "/")
path = PurePosixPath(name)
if path.is_absolute() or ".." in path.parts:
raise ImportContractError(f"Unsicherer ZIP-Pfad: {info.filename}")
if info.is_dir():
members.append(info)
continue
mode = (info.external_attr >> 16) & 0o170000
if mode == 0o120000:
raise ImportContractError(f"Symbolischer Link im ZIP nicht erlaubt: {info.filename}")
members.append(info)
return members
def _find_manifest(extracted: Path) -> tuple[Path, dict[str, Any], Path]:
candidates = [p for p in extracted.rglob("manifest.json") if len(p.relative_to(extracted).parts) <= 2]
if len(candidates) != 1:
raise ImportContractError(f"Genau ein Manifest im Paket- oder Repository-Root erwartet, gefunden: {len(candidates)}")
manifest_path = candidates[0]
try:
data = json.loads(manifest_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ImportContractError(f"Manifest nicht lesbar: {exc}") from exc
return manifest_path, data, manifest_path.parent
def inspect_archive(archive: Path) -> tuple[dict[str, Any], Path, tempfile.TemporaryDirectory[str]]:
if not archive.is_file() or not zipfile.is_zipfile(archive):
raise ImportContractError(f"Kein gültiges ZIP: {archive}")
temp = tempfile.TemporaryDirectory(prefix="ns-roundtrip-")
root = Path(temp.name)
try:
with zipfile.ZipFile(archive) as zf:
members = _safe_members(zf)
zf.extractall(root, members)
for info in members:
mode = (info.external_attr >> 16) & 0o777
if mode:
os.chmod(root / PurePosixPath(info.filename), mode)
_, manifest, payload = _find_manifest(root)
return manifest, payload, temp
except Exception:
temp.cleanup()
raise
def validate_repository_manifest(manifest: dict[str, Any], expected_repository: str) -> list[str]:
artifact_type = str(manifest.get("artifactType", ""))
artifact_name = str(manifest.get("artifactName", ""))
package_class = str(manifest.get("packageClass", ""))
install_mode = str(manifest.get("installMode", ""))
export_mode = str(manifest.get("exportMode", ""))
if package_class != "installable":
raise ImportContractError(f"Repository {expected_repository}: packageClass muss installable sein")
if install_mode not in {"overlay", "replace"}:
raise ImportContractError(f"Repository {expected_repository}: installMode muss overlay oder replace sein")
if export_mode not in {"full", "changed-only"}:
raise ImportContractError(f"Repository {expected_repository}: exportMode muss full oder changed-only sein")
if expected_repository not in {artifact_type, artifact_name}:
raise ImportContractError(
f"Repository-Identität stimmt nicht: erwartet {expected_repository}, erhalten {artifact_type}/{artifact_name}"
)
raw_deleted = manifest.get("deletedFiles", [])
if not isinstance(raw_deleted, list) or any(not isinstance(item, str) or not item for item in raw_deleted):
raise ImportContractError(f"Repository {expected_repository}: deletedFiles muss eine Stringliste sein")
if len(set(raw_deleted)) != len(raw_deleted):
raise ImportContractError(f"Repository {expected_repository}: deletedFiles enthält Duplikate")
normalized_deleted: list[str] = []
for item in raw_deleted:
path = PurePosixPath(item)
if (
path.is_absolute()
or ".." in path.parts
or item.endswith("/")
or "\\" in item
or any(character in item for character in "*?[]")
):
raise ImportContractError(f"Repository {expected_repository}: unzulässiger deletedFiles-Pfad: {item}")
# Ältere Paketgeneratoren haben den Repository-Namen mitgeliefert.
# Der Import arbeitet jedoch bereits im Repository-Root. Beide
# Schreibweisen bleiben lesbar; die Ausführung verwendet immer den
# kanonisch relativen Pfad.
parts = path.parts
if parts and parts[0] == expected_repository:
if len(parts) == 1:
raise ImportContractError(
f"Repository {expected_repository}: deletedFiles darf nicht den Repository-Root referenzieren: {item}"
)
item = str(PurePosixPath(*parts[1:]))
normalized_deleted.append(item)
if len(set(normalized_deleted)) != len(normalized_deleted):
raise ImportContractError(
f"Repository {expected_repository}: deletedFiles enthält nach Pfadnormalisierung Duplikate"
)
return normalized_deleted
def validate_gap_manifest(manifest: dict[str, Any]) -> None:
if manifest.get("artifactType") != "semantic-gap":
raise ImportContractError("Separates GAP-Paket muss artifactType semantic-gap besitzen")
if manifest.get("packageClass") != "installable":
raise ImportContractError("Separates GAP-Paket muss packageClass installable besitzen")
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 _find_roundtrip_package_manifest(extracted: Path) -> tuple[Path, dict[str, Any], Path]:
candidates = [
path for path in extracted.rglob("roundtrip-package-manifest.json")
if len(path.relative_to(extracted).parts) <= 2
]
if len(candidates) != 1:
raise ImportContractError(
"Gesamtpaket: genau ein roundtrip-package-manifest.json im Paket-Root erwartet, "
f"gefunden: {len(candidates)}"
)
path = candidates[0]
try:
manifest = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ImportContractError(f"Gesamtpaket: Manifest nicht lesbar: {exc}") from exc
return path, manifest, path.parent
def _inspect_roundtrip_package(archive: Path, repositories: dict[str, str], gap_target: Path) -> tuple[list[PreparedImport], Path, tempfile.TemporaryDirectory[str]]:
"""Prüft einen vollständigen Container vor dem ersten Schreibvorgang."""
if not archive.is_file() or not zipfile.is_zipfile(archive):
raise ImportContractError(f"Gesamtpaket ist kein gültiges ZIP: {archive}")
package_temp = tempfile.TemporaryDirectory(prefix="ns-roundtrip-package-")
root = Path(package_temp.name)
try:
with zipfile.ZipFile(archive) as handle:
members = _safe_members(handle)
handle.extractall(root, members)
_, manifest, payload = _find_roundtrip_package_manifest(root)
if manifest.get("artifactType") != "roundtrip-import-package" or str(manifest.get("packageVersion")) != "1":
raise ImportContractError("Gesamtpaket muss artifactType roundtrip-import-package und packageVersion 1 besitzen")
order = manifest.get("repositoryOrder")
entries = manifest.get("repositories")
gap = manifest.get("semanticGap")
if not isinstance(order, list) or not order or any(not isinstance(item, str) for item in order) or len(set(order)) != len(order):
raise ImportContractError("Gesamtpaket: repositoryOrder muss eine nichtleere, duplikatfreie Stringliste sein")
if not isinstance(entries, dict) or set(entries) != set(order):
raise ImportContractError("Gesamtpaket: repositories muss genau die Einträge von repositoryOrder enthalten")
if not isinstance(gap, dict):
raise ImportContractError("Gesamtpaket: semanticGap muss ein Archiv und eine Prüfsumme beschreiben")
listed = {"roundtrip-package-manifest.json"}
prepared: list[PreparedImport] = []
for repository in order:
if repository not in repositories:
raise ImportContractError(f"Gesamtpaket: nicht freigegebenes Repository in repositoryOrder: {repository}")
entry = entries[repository]
if not isinstance(entry, dict):
raise ImportContractError(f"Gesamtpaket: Repositoryeintrag ungültig: {repository}")
archive_name, checksum = entry.get("archive"), entry.get("sha256")
if not isinstance(archive_name, str) or PurePosixPath(archive_name).name != archive_name or not archive_name.endswith(".zip"):
raise ImportContractError(f"Gesamtpaket: ungültiger Archivname für {repository}")
if not isinstance(checksum, str) or len(checksum) != 64:
raise ImportContractError(f"Gesamtpaket: ungültige SHA-256-Prüfsumme für {repository}")
inner = payload / archive_name
if not inner.is_file() or _sha256(inner) != checksum:
raise ImportContractError(f"Gesamtpaket: Prüfsumme oder Archiv für {repository} stimmt nicht")
listed.add(archive_name)
inner_manifest, inner_payload, inner_temp = inspect_archive(inner)
try:
deleted_files = validate_repository_manifest(inner_manifest, repository)
_validate_deletions(inner_payload, Path(repositories[repository]).expanduser(), deleted_files)
finally:
inner_temp.cleanup()
prepared.append(PreparedImport(inner, repository, inner_manifest, deleted_files, Path(repositories[repository]).expanduser()))
gap_archive, gap_checksum = gap.get("archive"), gap.get("sha256")
if not isinstance(gap_archive, str) or PurePosixPath(gap_archive).name != gap_archive or not gap_archive.endswith(".zip"):
raise ImportContractError("Gesamtpaket: ungültiger Archivname für semanticGap")
if not isinstance(gap_checksum, str) or len(gap_checksum) != 64:
raise ImportContractError("Gesamtpaket: ungültige SHA-256-Prüfsumme für semanticGap")
inner_gap = payload / gap_archive
if not inner_gap.is_file() or _sha256(inner_gap) != gap_checksum:
raise ImportContractError("Gesamtpaket: Prüfsumme oder semanticGap-Archiv stimmt nicht")
listed.add(gap_archive)
gap_manifest, _, gap_temp = inspect_archive(inner_gap)
try:
validate_gap_manifest(gap_manifest)
finally:
gap_temp.cleanup()
actual_files = {item.name for item in payload.iterdir() if item.is_file()}
if actual_files != listed or any(item.is_dir() for item in payload.iterdir()):
raise ImportContractError("Gesamtpaket: nur Manifest und exakt die deklarierten Repository- und GAP-ZIPs erlaubt")
return prepared, inner_gap, package_temp
except Exception:
package_temp.cleanup()
raise
def _copy_overlay(source: Path, target: Path) -> None:
target.mkdir(parents=True, exist_ok=True)
for item in source.iterdir():
destination = target / item.name
if item.is_dir():
shutil.copytree(item, destination, dirs_exist_ok=True)
else:
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(item, destination)
def _validate_deletions(payload: Path, target: Path, deleted_files: list[str]) -> None:
for relative in deleted_files:
if (payload / relative).exists():
raise ImportContractError(f"Datei wird geliefert und zugleich gelöscht: {relative}")
candidate = target / relative
if candidate.exists() and candidate.is_dir():
raise ImportContractError(f"deletedFiles darf kein Verzeichnis referenzieren: {relative}")
def _apply_deletions(staging: Path, deleted_files: list[str]) -> None:
for relative in deleted_files:
candidate = staging / relative
if candidate.exists():
candidate.unlink()
parent = candidate.parent
while parent != staging and parent.is_dir() and not any(parent.iterdir()):
parent.rmdir()
parent = parent.parent
def _archive_previous(target: Path, archive_root: Path, repository: str, timestamp: str) -> Path | None:
if not target.exists():
return None
target = target.resolve()
archive_root = archive_root.expanduser().resolve()
destination = archive_root / repository / f"{timestamp}-before-import"
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copytree(target, destination)
return destination
def _validate_archive_root(archive_root: Path, repositories: dict[str, str]) -> Path:
"""Archive must be local state outside every importable repository."""
resolved = archive_root.expanduser().resolve()
for repository, configured_target in repositories.items():
target = Path(configured_target).expanduser().resolve()
try:
resolved.relative_to(target)
except ValueError:
continue
raise ImportContractError(
f"archiveRoot darf nicht innerhalb des importierbaren Repositorys {repository} liegen: "
f"{resolved} unter {target}. Einen externen lokalen Zustandsordner verwenden."
)
return resolved
def import_repository(archive: Path, repository: str, target: Path, archive_root: Path, timestamp: str) -> ImportResult:
manifest, payload, temp = inspect_archive(archive)
try:
deleted_files = validate_repository_manifest(manifest, repository)
_validate_deletions(payload, target, deleted_files)
previous = _archive_previous(target, archive_root, repository, timestamp)
staging = target.parent / f".{target.name}.import-{os.getpid()}"
if staging.exists():
shutil.rmtree(staging)
if target.exists():
shutil.copytree(target, staging)
else:
staging.mkdir(parents=True)
if manifest.get("installMode") == "replace":
shutil.rmtree(staging)
shutil.copytree(payload, staging)
else:
_copy_overlay(payload, staging)
_apply_deletions(staging, deleted_files)
backup = target.parent / f".{target.name}.previous-{os.getpid()}"
if backup.exists():
shutil.rmtree(backup)
if target.exists():
target.rename(backup)
try:
staging.rename(target)
except Exception:
if backup.exists() and not target.exists():
backup.rename(target)
raise
shutil.rmtree(backup, ignore_errors=True)
return ImportResult(
archive=archive.name,
artifact_type=str(manifest.get("artifactType", "unknown")),
artifact_name=str(manifest.get("artifactName", "unknown")),
version=str(manifest.get("version", "unknown")),
status="IMPORTED",
target=str(target),
archived_previous=str(previous) if previous else None,
message=f"Repository-Identität, Installierbarkeit und {len(deleted_files)} explizite Löschungen geprüft; atomar importiert.",
)
finally:
temp.cleanup()
def _is_roundtrip_package(archive: Path) -> bool:
if not zipfile.is_zipfile(archive):
return False
with zipfile.ZipFile(archive) as handle:
return any(PurePosixPath(info.filename).name == "roundtrip-package-manifest.json" for info in handle.infolist())
def _write_report(report_path: Path, report: dict[str, Any]) -> None:
report_path.parent.mkdir(parents=True, exist_ok=True)
report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def _total_package_report(
timestamp: str,
input_dir: Path,
prepared: list[PreparedImport],
results: list[ImportResult],
repositories: dict[str, str],
dry_run: bool,
status: str,
error: str | None = None,
) -> dict[str, Any]:
report: dict[str, Any] = {
"schema": "netzwerksolution.changed-only-roundtrip-import-report.v1",
"startedAt": timestamp,
"mode": "dry-run" if dry_run else "import",
"inputDirectory": str(input_dir),
"inputForm": "roundtrip-import-package",
"deliveredRepositoryCount": len(prepared),
"configuredRepositoryCount": len(repositories),
"unchangedRepositories": sorted(set(repositories) - {item.repository for item in prepared}),
"semanticGapDelivered": True,
"results": [asdict(result) for result in results],
"status": status,
}
if error:
report["error"] = error
return report
def _process_roundtrip_package(
archive: Path,
repositories: dict[str, str],
archive_root: Path,
gap_target: Path,
timestamp: str,
input_dir: Path,
report_path: Path,
dry_run: bool,
) -> dict[str, Any]:
prepared, gap_archive, package_temp = _inspect_roundtrip_package(archive, repositories, gap_target)
results: list[ImportResult] = []
try:
if dry_run:
for item in prepared:
results.append(ImportResult(item.archive.name, str(item.manifest.get("artifactType", "")), str(item.manifest.get("artifactName", "")), str(item.manifest.get("version", "unknown")), "VALIDATED", str(item.target), message="Gesamtpaket vollständig vorgeprüft; kein Import im Dry-Run."))
results.append(ImportResult(gap_archive.name, "semantic-gap", "semantic-gap", "unknown", "VALIDATED", str(gap_target), message="Gesamtpaket vollständig vorgeprüft; GAP wird nach erfolgreicher Repository-Reihenfolge übernommen."))
report = _total_package_report(timestamp, input_dir, prepared, results, repositories, dry_run, "PASSED")
_write_report(report_path, report)
return report
for index, item in enumerate(prepared):
try:
results.append(import_repository(item.archive, item.repository, item.target, archive_root, timestamp))
except Exception as exc:
for remaining in prepared[index + 1:]:
results.append(ImportResult(remaining.archive.name, str(remaining.manifest.get("artifactType", "")), str(remaining.manifest.get("artifactName", "")), str(remaining.manifest.get("version", "unknown")), "NOT_EXECUTED", str(remaining.target), message="Wegen eines vorherigen Importfehlers nicht eingelesen."))
results.append(ImportResult(gap_archive.name, "semantic-gap", "semantic-gap", "unknown", "NOT_EXECUTED", str(gap_target), message="Wegen eines vorherigen Importfehlers nicht übernommen."))
_write_report(report_path, _total_package_report(timestamp, input_dir, prepared, results, repositories, dry_run, "FAILED", str(exc)))
raise
gap_target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(gap_archive, gap_target)
results.append(ImportResult(gap_archive.name, "semantic-gap", "semantic-gap", "unknown", "IMPORTED", str(gap_target), message="Nach vollständigem Repository-Import übernommen."))
report = _total_package_report(timestamp, input_dir, prepared, results, repositories, dry_run, "PASSED")
_write_report(report_path, report)
return report
finally:
package_temp.cleanup()
def process(config_path: Path, input_dir: Path, report_path: Path, dry_run: bool = False) -> dict[str, Any]:
config = json.loads(config_path.read_text(encoding="utf-8"))
repositories: dict[str, str] = config.get("repositories", {})
archive_root = _validate_archive_root(Path(config["archiveRoot"]), repositories)
gap_target = Path(config["semanticGapTarget"]).expanduser()
timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
results: list[ImportResult] = []
candidates = sorted(input_dir.glob("*.zip"))
package_candidates = [archive for archive in candidates if _is_roundtrip_package(archive)]
if package_candidates:
if len(package_candidates) != 1 or len(candidates) != 1:
raise ImportContractError("Ein Gesamtpaket muss allein in seinem Eingabeverzeichnis liegen")
return _process_roundtrip_package(
package_candidates[0], repositories, archive_root, gap_target,
timestamp, input_dir, report_path, dry_run,
)
seen_repositories: set[str] = set()
gap_seen = False
for archive in candidates:
manifest, _, temp = inspect_archive(archive)
try:
artifact_type = str(manifest.get("artifactType", ""))
artifact_name = str(manifest.get("artifactName", ""))
finally:
temp.cleanup()
if artifact_type == "semantic-gap":
if gap_seen:
raise ImportContractError("Mehr als ein semantic-gap-Paket im Lauf")
validate_gap_manifest(manifest)
gap_seen = True
if dry_run:
results.append(ImportResult(archive.name, artifact_type, artifact_name, str(manifest.get("version", "unknown")), "VALIDATED", str(gap_target), message="Separates GAP-Paket geprüft."))
else:
gap_target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(archive, gap_target)
results.append(ImportResult(archive.name, artifact_type, artifact_name, str(manifest.get("version", "unknown")), "IMPORTED", str(gap_target), message="Separates GAP-Paket aktualisiert."))
continue
repository = next((name for name in repositories if name in {artifact_type, artifact_name}), None)
if repository is None:
raise ImportContractError(f"Nicht freigegebenes oder unbekanntes Repository-Paket: {archive.name} ({artifact_type}/{artifact_name})")
if repository in seen_repositories:
raise ImportContractError(f"Repository mehrfach geliefert: {repository}")
seen_repositories.add(repository)
target = Path(repositories[repository]).expanduser()
if dry_run:
dry_manifest, payload, dry_temp = inspect_archive(archive)
try:
deleted_files = validate_repository_manifest(dry_manifest, repository)
_validate_deletions(payload, target, deleted_files)
finally:
dry_temp.cleanup()
results.append(ImportResult(archive.name, artifact_type, artifact_name, str(manifest.get("version", "unknown")), "VALIDATED", str(target), message=f"Changed-only-Lieferung und {len(deleted_files)} explizite Löschungen geprüft; kein Import im Dry-Run."))
else:
results.append(import_repository(archive, repository, target, archive_root, timestamp))
report = {
"schema": "netzwerksolution.changed-only-roundtrip-import-report.v1",
"startedAt": timestamp,
"mode": "dry-run" if dry_run else "import",
"inputDirectory": str(input_dir),
"deliveredRepositoryCount": len(seen_repositories),
"configuredRepositoryCount": len(repositories),
"unchangedRepositories": sorted(set(repositories) - seen_repositories),
"semanticGapDelivered": gap_seen,
"results": [asdict(r) for r in results],
"status": "PASSED",
}
_write_report(report_path, report)
return report
def main() -> int:
parser = argparse.ArgumentParser(description="Importiert ausschließlich gelieferte Roundtrip-Repository-ZIPs und ein separates semantic-gap.zip.")
parser.add_argument("--config", required=True, type=Path)
parser.add_argument("--input-dir", required=True, type=Path)
parser.add_argument("--report", required=True, type=Path)
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args()
try:
process(args.config, args.input_dir, args.report, args.dry_run)
except (OSError, json.JSONDecodeError, ImportContractError) as exc:
try:
existing = json.loads(args.report.read_text(encoding="utf-8")) if args.report.exists() else {}
except (OSError, json.JSONDecodeError):
existing = {}
if existing.get("status") != "FAILED" or existing.get("error") != str(exc):
_write_report(args.report, {"schema": "netzwerksolution.changed-only-roundtrip-import-report.v1", "status": "FAILED", "error": str(exc)})
print(f"FEHLER: {exc}")
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())