Spaces:
Sleeping
Sleeping
| """Access the retained source edition, rather than a mutable mirror URL.""" | |
| from __future__ import annotations | |
| import hashlib | |
| import re | |
| import tarfile | |
| from pathlib import Path | |
| from typing import Any | |
| DOCUMENT_ID = re.compile(r"(?:[A-Za-z0-9][A-Za-z0-9._#@-]{0,127}|@[a-f0-9]{16})") | |
| _SHA256 = re.compile(r"[a-f0-9]{64}") | |
| _ARCHIVE_LAYOUT = re.compile( | |
| r"(?:^|/)(?P<relative>data/tar/year=\d{4}/english/english\.tar|" | |
| r"recovered_official/year=\d{4}/official_recovered\.tar|" | |
| r"legislation/constitution-\d{4}-\d{2}-\d{2}\.tar)$" | |
| ) | |
| def _archive(record: dict[str, Any], root: str | Path) -> Path: | |
| trusted_root = Path(root).resolve() | |
| declared = str(record.get("source_archive") or "") | |
| path = Path(declared).resolve() | |
| if not path.is_relative_to(trusted_root): | |
| # A deployed snapshot can move without rewriting its immutable corpus. | |
| # Rebase only the known dataset layout, then enforce the trusted root | |
| # again after resolving symlinks. The PDF hash is still checked on read. | |
| layout = _ARCHIVE_LAYOUT.search(declared) | |
| if layout: | |
| path = (trusted_root / layout["relative"]).resolve() | |
| if not path.is_relative_to(trusted_root) or path.suffix != ".tar": | |
| raise FileNotFoundError("Source archive is outside the configured source root") | |
| if not path.is_file(): | |
| raise FileNotFoundError("Retained source archive is unavailable") | |
| if not _SHA256.fullmatch(str(record.get("pdf_sha256") or "")): | |
| raise FileNotFoundError("Source PDF has no verifiable hash") | |
| if not str(record.get("source_member") or "").lower().endswith(".pdf"): | |
| raise FileNotFoundError("Source PDF member is unavailable") | |
| return path | |
| def source_pdf_path(record: dict[str, Any], root: str | Path) -> str | None: | |
| """Advertise only retained sources within the configured archive root.""" | |
| parent = str(record.get("parent_id") or record.get("document_id") or "") | |
| if not DOCUMENT_ID.fullmatch(parent): | |
| return None | |
| try: | |
| _archive(record, root) | |
| except (FileNotFoundError, OSError, ValueError): | |
| return None | |
| return f"/document/{parent}/pdf" | |
| def read_source_pdf( | |
| record: dict[str, Any], root: str | Path, *, max_bytes: int = 64 * 1024 * 1024 | |
| ) -> bytes: | |
| """Read one regular PDF member and verify its exact extraction-source hash. | |
| The archive is opened separately per request; TarFile is not thread-safe. | |
| No member is extracted to the filesystem, and untrusted archive paths cannot | |
| escape the configured source root. Reject oversized or changed source bytes. | |
| """ | |
| path = _archive(record, root) | |
| try: | |
| with tarfile.open(path, "r:") as archive: | |
| member = archive.getmember(str(record["source_member"])) | |
| if not member.isfile() or member.size > max_bytes: | |
| raise ValueError("Source PDF is not a bounded regular file") | |
| handle = archive.extractfile(member) | |
| if handle is None: | |
| raise FileNotFoundError("Source PDF member is unavailable") | |
| with handle: | |
| data = handle.read(max_bytes + 1) | |
| except KeyError as exc: | |
| raise FileNotFoundError("Source PDF member is unavailable") from exc | |
| if len(data) > max_bytes or not data.lstrip().startswith(b"%PDF-"): | |
| raise ValueError("Retained source is not a bounded PDF") | |
| if hashlib.sha256(data).hexdigest() != record["pdf_sha256"]: | |
| raise ValueError("Retained PDF does not match the extraction-source hash") | |
| return data | |