"""Bounded country extraction for markets sharing a larger Geofabrik PBF.""" from __future__ import annotations import hashlib import os import stat import subprocess import tempfile from dataclasses import dataclass from pathlib import Path from ..errors import PbfDownloadRejected from ..markets import GeoBounds from .geofabrik import AcquiredPbf _CHUNK_SIZE = 1024 * 1024 _MAX_EXTRACT_BYTES = 4 * 1024 * 1024 * 1024 def _regular_file(path: Path, field: str) -> os.stat_result: try: info = path.lstat() except OSError as exc: raise PbfDownloadRejected(f"{field} cannot be inspected") from exc if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode): raise PbfDownloadRejected(f"{field} must be a regular file") return info def _ensure_directory(path: Path) -> None: path.mkdir(mode=0o700, parents=True, exist_ok=True) try: info = path.lstat() except OSError as exc: raise PbfDownloadRejected("extraction cache cannot be inspected") from exc if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): raise PbfDownloadRejected("extraction cache must be a regular directory") def _hash_file(path: Path) -> tuple[int, str]: before = _regular_file(path, "extracted PBF") if not 1 <= before.st_size <= _MAX_EXTRACT_BYTES: raise PbfDownloadRejected("extracted PBF size is outside the approved bound") digest = hashlib.sha256() total = 0 try: with path.open("rb") as handle: while chunk := handle.read(_CHUNK_SIZE): total += len(chunk) if total > _MAX_EXTRACT_BYTES: raise PbfDownloadRejected("extracted PBF exceeds the approved bound") digest.update(chunk) except OSError as exc: raise PbfDownloadRejected("extracted PBF cannot be read") from exc after = _regular_file(path, "extracted PBF") if ( total != before.st_size or before.st_dev != after.st_dev or before.st_ino != after.st_ino or before.st_mtime_ns != after.st_mtime_ns or before.st_ctime_ns != after.st_ctime_ns ): raise PbfDownloadRejected("extracted PBF changed during validation") return total, digest.hexdigest() def _coordinate(value: int) -> str: return f"{value / 10_000_000:.7f}".rstrip("0").rstrip(".") @dataclass(frozen=True, slots=True) class ExtractedPbf: """Validated evidence for one country-bounded PBF derived from a source PBF.""" local_path: Path content_sha256: str size_bytes: int source_sha256: str bounds: GeoBounds cache_hit: bool def as_mapping(self) -> dict[str, object]: return { "bounds": self.bounds.as_mapping(), "cache_hit": self.cache_hit, "content_sha256": self.content_sha256, "local_path": str(self.local_path), "size_bytes": self.size_bytes, "source_sha256": self.source_sha256, } def extract_country_pbf( acquisition: AcquiredPbf, bounds: GeoBounds, cache_dir: Path, *, osmium_binary: Path = Path("/usr/bin/osmium"), timeout_seconds: float = 7200.0, ) -> ExtractedPbf: """Extract one fixed bounding box through osmium-tool without shell execution.""" if not isinstance(acquisition, AcquiredPbf): raise PbfDownloadRejected("acquisition must be AcquiredPbf") if not isinstance(bounds, GeoBounds): raise PbfDownloadRejected("bounds must be GeoBounds") if ( isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, (int, float)) or not 30 <= float(timeout_seconds) <= 14_400 ): raise PbfDownloadRejected("extraction timeout is outside the approved range") if not osmium_binary.is_absolute() or osmium_binary.name != "osmium": raise PbfDownloadRejected("osmium binary path is not approved") try: binary_info = osmium_binary.lstat() except OSError as exc: raise PbfDownloadRejected("osmium binary is unavailable") from exc if ( stat.S_ISLNK(binary_info.st_mode) or not stat.S_ISREG(binary_info.st_mode) or not os.access(osmium_binary, os.X_OK) ): raise PbfDownloadRejected("osmium binary must be an executable regular file") _regular_file(acquisition.local_path, "source PBF") root = cache_dir.expanduser() _ensure_directory(root) cache_key = hashlib.sha256( ( f"{acquisition.content_sha256}:" f"{bounds.min_longitude_e7}:{bounds.min_latitude_e7}:" f"{bounds.max_longitude_e7}:{bounds.max_latitude_e7}" ).encode() ).hexdigest() target = root / f"{cache_key}.osm.pbf" if target.exists(): size, digest = _hash_file(target) return ExtractedPbf( target, digest, size, acquisition.content_sha256, bounds, True, ) temporary_path: Path | None = None try: with tempfile.NamedTemporaryFile( dir=root, prefix=".extract-", suffix=".osm.pbf", delete=False, ) as temporary: temporary_path = Path(temporary.name) os.chmod(temporary_path, 0o600) bbox = ",".join( ( _coordinate(bounds.min_longitude_e7), _coordinate(bounds.min_latitude_e7), _coordinate(bounds.max_longitude_e7), _coordinate(bounds.max_latitude_e7), ) ) try: result = subprocess.run( [ str(osmium_binary), "extract", f"--bbox={bbox}", "--strategy=complete_ways", "--overwrite", f"--output={temporary_path}", str(acquisition.local_path), ], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, timeout=float(timeout_seconds), check=False, ) except (OSError, subprocess.SubprocessError) as exc: raise PbfDownloadRejected("country PBF extraction failed") from exc if result.returncode != 0: raise PbfDownloadRejected("country PBF extraction returned a nonzero status") size, digest = _hash_file(temporary_path) try: os.link(temporary_path, target, follow_symlinks=False) except FileExistsError: existing_size, existing_digest = _hash_file(target) if existing_size != size or existing_digest != digest: raise PbfDownloadRejected("country extraction cache collision detected") from None return ExtractedPbf( target, digest, size, acquisition.content_sha256, bounds, False, ) finally: if temporary_path is not None: temporary_path.unlink(missing_ok=True) __all__ = ["ExtractedPbf", "extract_country_pbf"]