"""Provenance and atomic publication for the DITAv2 Rust shared library. The loader must never silently pair a stale native library with current Python bindings or Rust source. A verified sidecar manifest binds the library to the crate source, Cargo.lock, compiler, target, release profile, and FFI schema. Builds happen in a staging target directory and are published only after the candidate library and manifest verify together. The previous verified pair is kept as a rollback copy. """ from __future__ import annotations import hashlib import json import os from pathlib import Path import platform import shutil import subprocess import tempfile from dataclasses import dataclass from typing import Any, Mapping FFI_SCHEMA_VERSION = 1 PROFILE = "release" MANIFEST_SUFFIX = ".manifest.json" class ArtifactProvenanceError(RuntimeError): """A native artifact is missing, stale, malformed, or unverifiable.""" @dataclass(frozen=True) class ArtifactVerification: """Successful verification result suitable for readiness telemetry.""" library_path: Path manifest_path: Path library_sha256: str source_tree_sha256: str cargo_lock_sha256: str rustc_fingerprint: str target_triple: str ffi_schema_version: int def library_name() -> str: if os.name == "nt": return "dita_v2_kernel.dll" if platform.system() == "Darwin": return "libdita_v2_kernel.dylib" return "libdita_v2_kernel.so" def manifest_path(library_path: Path) -> Path: return Path(f"{library_path}{MANIFEST_SUFFIX}") def rollback_directory(target_dir: Path) -> Path: return target_dir / "rollback" def _sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _sha256_file(path: Path) -> str: try: return _sha256_bytes(path.read_bytes()) except OSError as exc: raise ArtifactProvenanceError(f"cannot read artifact input {path}: {exc}") from exc def source_tree_sha256(crate_dir: Path) -> str: """Hash all source/build-input files under the Rust crate deterministically.""" if not crate_dir.is_dir(): raise ArtifactProvenanceError(f"Rust crate directory missing: {crate_dir}") paths = sorted( path for path in crate_dir.rglob("*") if path.is_file() and "target" not in path.parts and "__pycache__" not in path.parts and not path.name.endswith(".pyc") ) if not paths: raise ArtifactProvenanceError(f"Rust crate has no source files: {crate_dir}") digest = hashlib.sha256() for path in paths: relative = path.relative_to(crate_dir).as_posix().encode("utf-8") digest.update(relative) digest.update(b"\0") digest.update(path.read_bytes()) digest.update(b"\0") return digest.hexdigest() def _rustc_metadata() -> tuple[str, str]: """Return the exact rustc output fingerprint and target triple.""" try: result = subprocess.run( ["rustc", "-vV"], check=True, capture_output=True, text=True, timeout=10, ) except (OSError, subprocess.SubprocessError) as exc: raise ArtifactProvenanceError(f"rustc metadata unavailable: {exc}") from exc output = result.stdout.strip() target = "" for line in output.splitlines(): if line.startswith("host:"): target = line.split(":", 1)[1].strip() break if not output or not target: raise ArtifactProvenanceError("rustc -vV returned incomplete metadata") return _sha256_bytes(output.encode("utf-8")), target def _canonical_json(value: Mapping[str, Any]) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") def _manifest_fingerprint(manifest: Mapping[str, Any]) -> str: payload = dict(manifest) payload.pop("manifest_sha256", None) payload.pop("library_sha256", None) return _sha256_bytes(_canonical_json(payload)) def build_manifest(library_path: Path, crate_dir: Path) -> dict[str, Any]: """Build a manifest for an already-built library candidate.""" if not library_path.is_file(): raise ArtifactProvenanceError(f"native library missing: {library_path}") rustc_fingerprint, target = _rustc_metadata() cargo_lock = crate_dir / "Cargo.lock" cargo_toml = crate_dir / "Cargo.toml" if not cargo_lock.is_file() or not cargo_toml.is_file(): raise ArtifactProvenanceError("Cargo.toml and Cargo.lock are required") manifest: dict[str, Any] = { "component": "dita-v2-kernel", "crate_version": "0.1.0", "source_tree_sha256": source_tree_sha256(crate_dir), "cargo_lock_sha256": _sha256_file(cargo_lock), "cargo_manifest_sha256": _sha256_file(cargo_toml), "rustc_fingerprint": rustc_fingerprint, "target_triple": target, "profile": PROFILE, "features": [], "ffi_schema_version": FFI_SCHEMA_VERSION, "library_sha256": _sha256_file(library_path), } manifest["manifest_sha256"] = _manifest_fingerprint(manifest) return manifest def write_manifest(library_path: Path, crate_dir: Path) -> Path: """Atomically write and return the manifest for *library_path*.""" destination = manifest_path(library_path) destination.parent.mkdir(parents=True, exist_ok=True) payload = json.dumps(build_manifest(library_path, crate_dir), indent=2, sort_keys=True) + "\n" fd, temporary = tempfile.mkstemp(prefix=f".{destination.name}.", dir=str(destination.parent)) try: with os.fdopen(fd, "w", encoding="utf-8") as handle: handle.write(payload) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, destination) finally: try: os.unlink(temporary) except FileNotFoundError: pass return destination def verify_artifact(library_path: Path, crate_dir: Path) -> ArtifactVerification: """Verify a library and sidecar against current source and toolchain.""" library_path = Path(library_path) sidecar = manifest_path(library_path) if not library_path.is_file(): raise ArtifactProvenanceError(f"native library missing: {library_path}") if not sidecar.is_file(): raise ArtifactProvenanceError(f"native artifact manifest missing: {sidecar}") try: manifest = json.loads(sidecar.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ArtifactProvenanceError(f"native artifact manifest unreadable: {sidecar}: {exc}") from exc required = ( "component", "crate_version", "source_tree_sha256", "cargo_lock_sha256", "cargo_manifest_sha256", "rustc_fingerprint", "target_triple", "profile", "features", "ffi_schema_version", "library_sha256", "manifest_sha256", ) missing = sorted(set(required).difference(manifest)) if missing: raise ArtifactProvenanceError(f"native artifact manifest missing fields: {missing}") expected = build_manifest(library_path, crate_dir) # Compare source and toolchain inputs before derived fields so the failure # names the root cause rather than the manifest hash that depends on it. for field in required[:-1]: if manifest.get(field) != expected.get(field): raise ArtifactProvenanceError( f"native artifact provenance mismatch field={field} " f"recorded={manifest.get(field)!r} expected={expected.get(field)!r}" ) if manifest.get("manifest_sha256") != _manifest_fingerprint(manifest): raise ArtifactProvenanceError("native artifact manifest self-fingerprint mismatch") return ArtifactVerification( library_path=library_path, manifest_path=sidecar, library_sha256=str(expected["library_sha256"]), source_tree_sha256=str(expected["source_tree_sha256"]), cargo_lock_sha256=str(expected["cargo_lock_sha256"]), rustc_fingerprint=str(expected["rustc_fingerprint"]), target_triple=str(expected["target_triple"]), ffi_schema_version=int(expected["ffi_schema_version"]), ) def _copy_pair(library: Path, sidecar: Path, destination: Path) -> None: destination.mkdir(parents=True, exist_ok=True) shutil.copy2(library, destination / library.name) shutil.copy2(sidecar, destination / sidecar.name) def build_verified_artifact(crate_dir: Path, target_dir: Path) -> ArtifactVerification: """Build, verify, atomically publish, and retain rollback state.""" crate_dir = Path(crate_dir) target_dir = Path(target_dir) target_dir.mkdir(parents=True, exist_ok=True) final_release = target_dir / "release" final_library = final_release / library_name() final_sidecar = manifest_path(final_library) rollback = rollback_directory(target_dir) staging = target_dir.parent / f".{target_dir.name}.staging-{os.getpid()}" if staging.exists(): shutil.rmtree(staging) try: env = {**os.environ, "CARGO_TARGET_DIR": str(staging)} subprocess.run( ["cargo", "build", "--release", "--manifest-path", str(crate_dir / "Cargo.toml")], cwd=crate_dir.parents[3], check=True, env=env, timeout=300, ) candidate = staging / "release" / library_name() candidate_sidecar = write_manifest(candidate, crate_dir) verify_artifact(candidate, crate_dir) if final_library.is_file() and final_sidecar.is_file(): _copy_pair(final_library, final_sidecar, rollback) final_release.mkdir(parents=True, exist_ok=True) os.replace(candidate, final_library) os.replace(candidate_sidecar, final_sidecar) return verify_artifact(final_library, crate_dir) except Exception: raise finally: if staging.exists(): shutil.rmtree(staging, ignore_errors=True) def rollback_artifact(target_dir: Path, crate_dir: Path) -> ArtifactVerification: """Restore the last verified pair and verify it before returning.""" target_dir = Path(target_dir) rollback = rollback_directory(target_dir) restored_library = rollback / library_name() restored_sidecar = manifest_path(restored_library) if not restored_library.is_file() or not restored_sidecar.is_file(): raise ArtifactProvenanceError(f"no verified rollback pair in {rollback}") destination = target_dir / "release" destination.mkdir(parents=True, exist_ok=True) shutil.copy2(restored_library, destination / restored_library.name) shutil.copy2(restored_sidecar, destination / manifest_path(restored_library).name) return verify_artifact(destination / library_name(), crate_dir) __all__ = [ "ArtifactProvenanceError", "ArtifactVerification", "FFI_SCHEMA_VERSION", "build_manifest", "build_verified_artifact", "library_name", "manifest_path", "rollback_artifact", "source_tree_sha256", "verify_artifact", "write_manifest", ]