Files
ThothII/harness/tht/adapters/evidence/filesystem.py
T

149 lines
5.6 KiB
Python

"""Contained, race-safe filesystem Evidence source."""
import hashlib
import os
import stat
from datetime import UTC, datetime
from pathlib import Path, PurePosixPath
from urllib.parse import unquote, urlsplit
from tht.evidence.contracts import (
AcquiredDocument,
EvidenceSourceError,
EvidenceSourceErrorCategory,
SourceObject,
)
class FilesystemEvidenceSource:
def __init__(
self,
root: Path | str,
*,
patterns: tuple[str, ...] | list[str] = ("**/*.md",),
max_bytes: int = 10 * 1024 * 1024,
) -> None:
if max_bytes < 1:
raise ValueError("max_bytes must be positive")
if not patterns or any(
not pattern or Path(pattern).is_absolute() or ".." in Path(pattern).parts
for pattern in patterns
):
raise ValueError("at least one non-empty discovery pattern is required")
try:
self.root = Path(root).expanduser().resolve(strict=True)
self._root_fd = os.open(
self.root,
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
)
except OSError as error:
raise ValueError("filesystem evidence root is unavailable") from error
self.patterns = tuple(patterns)
self.max_bytes = max_bytes
def __del__(self):
root_fd = getattr(self, "_root_fd", None)
if root_fd is not None:
try:
os.close(root_fd)
except OSError:
pass
@staticmethod
def _safe_error(operation: str, *, transient: bool = False, **details):
return EvidenceSourceError(
"filesystem source operation failed",
category=(
EvidenceSourceErrorCategory.TRANSIENT
if transient
else EvidenceSourceErrorCategory.PERMANENT
),
details={"operation": operation, **details},
)
def _open_read(self, relative: PurePosixPath) -> tuple[bytes, os.stat_result]:
parts = relative.parts
if not parts or any(part in {"", ".", ".."} for part in parts):
raise self._safe_error("path_validation")
directory_fd = os.dup(self._root_fd)
file_fd = None
try:
for component in parts[:-1]:
next_fd = os.open(
component,
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
dir_fd=directory_fd,
)
os.close(directory_fd)
directory_fd = next_fd
file_fd = os.open(
parts[-1],
os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC,
dir_fd=directory_fd,
)
file_stat = os.fstat(file_fd)
if not stat.S_ISREG(file_stat.st_mode):
raise self._safe_error("path_validation")
if file_stat.st_size > self.max_bytes:
raise self._safe_error("read", limit_bytes=self.max_bytes)
content = bytearray()
while len(content) <= self.max_bytes:
chunk = os.read(file_fd, min(64 * 1024, self.max_bytes + 1 - len(content)))
if not chunk:
break
content.extend(chunk)
if len(content) > self.max_bytes:
raise self._safe_error("read", limit_bytes=self.max_bytes)
return bytes(content), file_stat
except EvidenceSourceError:
raise
except OSError as error:
raise self._safe_error("open") from error
finally:
if file_fd is not None:
os.close(file_fd)
os.close(directory_fd)
def _item(
self, relative: PurePosixPath, content: bytes, file_stat: os.stat_result
) -> SourceObject:
relative_text = relative.as_posix()
return SourceObject(
source_id=f"filesystem:{hashlib.sha256(relative_text.encode()).hexdigest()}",
uri=(self.root / relative_text).as_uri(),
fingerprint=f"sha256:{hashlib.sha256(content).hexdigest()}",
modified_at=datetime.fromtimestamp(file_stat.st_mtime, tz=UTC),
metadata={"relative_path": relative_text},
)
def discover(self):
candidates = {
path.relative_to(self.root).as_posix()
for pattern in self.patterns
for path in self.root.glob(pattern)
}
for relative_text in sorted(candidates):
relative = PurePosixPath(relative_text)
content, file_stat = self._open_read(relative)
yield self._item(relative, content, file_stat)
def acquire(self, item: SourceObject) -> AcquiredDocument:
parsed = urlsplit(item.uri)
if parsed.scheme != "file" or parsed.netloc or parsed.query or parsed.fragment:
raise self._safe_error("acquire")
try:
relative = Path(unquote(parsed.path)).relative_to(self.root)
except ValueError as error:
raise self._safe_error("acquire") from error
pure_relative = PurePosixPath(relative.as_posix())
content, file_stat = self._open_read(pure_relative)
expected = self._item(pure_relative, content, file_stat)
if item.source_id != expected.source_id or item.fingerprint != expected.fingerprint:
raise self._safe_error("acquire")
return AcquiredDocument(
source=expected,
content=content,
media_type="text/markdown" if relative.suffix.lower() == ".md" else None,
acquired_at=datetime.now(UTC),
)