| """Fail-closed integrity guard. |
| |
| A signed file manifest ships inside the built wheel. On import the package |
| recomputes the hash of every shipped file and compares it to the manifest. Any |
| edit, overwrite, added file, or removed file trips the guard: the package marks |
| itself bricked and refuses to run. |
| |
| Honest limit: this is tamper-evident, not tamper-proof. A user who edits the |
| guard itself, or deletes the manifest, disables the check, because the code runs |
| on their machine. No client-side mechanism can prevent that. What this does |
| guarantee is that any accidental or casual modification of a shipped file stops |
| the product working rather than running altered code silently. |
| |
| The manifest is generated at build time (see hatch_build.py) and is absent from |
| the source tree, so an editable development checkout is not enforced. |
| """ |
|
|
| |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| from collections.abc import Iterator |
| from pathlib import Path |
|
|
| MANIFEST_NAME = "_integrity_manifest.json" |
| MANIFEST_ALGORITHM = "sha256" |
| _SKIP_SUFFIXES = (".pyc", ".pyo") |
| _SKIP_DIRS = frozenset({"__pycache__"}) |
|
|
| _bricked_reason: str | None = None |
|
|
|
|
| class IntegrityError(RuntimeError): |
| """Raised when a shipped file has been altered, added, or removed.""" |
|
|
|
|
| def _iter_files(root: Path) -> Iterator[Path]: |
| for path in sorted(root.rglob("*")): |
| if not path.is_file(): |
| continue |
| if any(part in _SKIP_DIRS for part in path.relative_to(root).parts): |
| continue |
| if path.suffix in _SKIP_SUFFIXES: |
| continue |
| if path.name == MANIFEST_NAME: |
| continue |
| yield path |
|
|
|
|
| def _hash_file(path: Path) -> str: |
| digest = hashlib.new(MANIFEST_ALGORITHM) |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(65536), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def build_manifest(root: Path) -> dict[str, object]: |
| """Compute the manifest for a package directory.""" |
| files = { |
| str(path.relative_to(root).as_posix()): _hash_file(path) |
| for path in _iter_files(root) |
| } |
| return {"version": 1, "algorithm": MANIFEST_ALGORITHM, "files": files} |
|
|
|
|
| def verify_tree(root: Path) -> None: |
| """ |
| Verify a package directory against its manifest. |
| |
| No manifest present means an unbuilt or editable tree, which is not |
| enforced. A present manifest is enforced strictly. |
| """ |
| manifest_path = root / MANIFEST_NAME |
| if not manifest_path.exists(): |
| return |
|
|
| try: |
| manifest = json.loads(manifest_path.read_text(encoding="utf-8")) |
| expected = dict(manifest["files"]) |
| except (OSError, ValueError, KeyError, TypeError) as exc: |
| raise IntegrityError("Integrity manifest is unreadable or malformed") from exc |
|
|
| actual = { |
| str(path.relative_to(root).as_posix()): _hash_file(path) |
| for path in _iter_files(root) |
| } |
|
|
| added = sorted(set(actual) - set(expected)) |
| removed = sorted(set(expected) - set(actual)) |
| changed = sorted( |
| name for name in expected.keys() & actual.keys() if expected[name] != actual[name] |
| ) |
|
|
| if added or removed or changed: |
| parts = [] |
| if changed: |
| parts.append(f"modified: {', '.join(changed)}") |
| if removed: |
| parts.append(f"missing: {', '.join(removed)}") |
| if added: |
| parts.append(f"unexpected: {', '.join(added)}") |
| raise IntegrityError( |
| "Package integrity check failed (" + "; ".join(parts) + "). " |
| "A shipped file was altered. Reinstall a clean copy: " |
| "pip install --force-reinstall affix-huggingface" |
| ) |
|
|
|
|
| def enforce() -> None: |
| """Run the guard for this installed package and remember the outcome.""" |
| global _bricked_reason |
| try: |
| verify_tree(Path(__file__).resolve().parent) |
| except IntegrityError as exc: |
| _bricked_reason = str(exc) |
| raise |
|
|
|
|
| def ensure_intact() -> None: |
| """Refuse to proceed once the guard has tripped.""" |
| if _bricked_reason is not None: |
| raise IntegrityError(_bricked_reason) |
|
|
