ss-tools/scripts/dr_evidence_io.py

# #region Tooling.DrEvidenceCheck.Containment [C:5] [TYPE Module] [SEMANTICS dr,evidence,containment]
# @BRIEF Resolve typed evidence refs and open exact regular inodes through pinned directory descriptors.
import errno
import hashlib
import os
from pathlib import Path
import stat

from dr_evidence_scope import ScopeContractRefused

# Nonregular leaves must reach fstat without waiting for a FIFO writer/device peer. The same
# descriptor is still rejected below unless it is a regular file; no pathname type check replaces it.
_OPEN_FLAGS = os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_CLOEXEC", 0)


# #region Tooling.DrEvidenceCheck.Containment.Resolve [C:3] [TYPE Function] [SEMANTICS dr,recovery]
# @INVARIANT Handle refs require the explicit handle: prefix and 64 lowercase hexadecimal digits.
def _resolve(parse_draft, ref: str) -> tuple[tuple[str, ...], str]:
    """Map one opaque ref to (relative path components, digest); no filesystem access."""
    if ref.startswith("draft:"):
        run_id, digest = parse_draft(ref)
        if run_id in {".", ".."}:
            raise ValueError("DRAFT_OWNER_INVALID")
        return ("drafts", run_id, digest), digest
    if not ref.startswith("handle:"):
        raise ValueError("REF_SCHEME_INVALID")
    digest = ref[len("handle:"):]
    if len(digest) != 64 or any(character not in "0123456789abcdef" for character in digest):
        raise ValueError("HANDLE_REF_INVALID")
    return ("drafts", "handles", digest), digest
# #endregion Tooling.DrEvidenceCheck.Containment.Resolve


# #region Tooling.DrEvidenceCheck.Containment.OpenRoot [C:5] [TYPE Function] [SEMANTICS dr,evidence,containment]
# @RATIONALE Opening each absolute component relative to its pinned parent rejects ancestor aliases and replacement races.
# @REJECTED One absolute O_NOFOLLOW open protects only the final component; resolve/lstat admission can select another root.
def _open_root(root_real: Path) -> int:
    """Pin every absolute directory component without following a symlink at any depth."""
    current = None
    try:
        root = Path(root_real)
        if not root.is_absolute() or ".." in root.parts:
            raise ValueError("absolute root required")
        flags = _OPEN_FLAGS | os.O_DIRECTORY | os.O_NOFOLLOW
        current = os.open(root.anchor, flags)
        for part in root.parts[1:]:
            child = os.open(part, flags, dir_fd=current)
            os.close(current)
            current = child
        result = current
        current = None  # Transfer only the final owned directory descriptor to the caller.
        return result
    except (OSError, TypeError, ValueError):
        raise ScopeContractRefused("EVIDENCE_STORAGE_ROOT_INVALID") from None
    finally:
        if current is not None:
            os.close(current)
# #endregion Tooling.DrEvidenceCheck.Containment.OpenRoot


# #region Tooling.DrEvidenceCheck.Containment.ClassifyComponent [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _classify_component(part: str, parent_fd: int) -> str | None:
    """Name the reason a component is not a followable directory/file, or None when it is plain."""
    try:
        info = os.stat(part, dir_fd=parent_fd, follow_symlinks=False)
    except OSError:
        return "EVIDENCE_BYTES_MISSING"
    if stat.S_ISLNK(info.st_mode):
        return "EVIDENCE_PATH_SYMLINK"
    return None
# #endregion Tooling.DrEvidenceCheck.Containment.ClassifyComponent


# #region Tooling.DrEvidenceCheck.Containment.OpenFailure [C:3] [TYPE Function] [SEMANTICS dr,recovery]
# @RATIONALE A replacement symlink can yield ELOOP or ENOTDIR; nofollow reclassification preserves the typed symlink refusal.
def _open_failure(part, parent_fd, error):
    """Classify a classify/open race without following a replacement symlink."""
    reason = _classify_component(part, parent_fd)
    if reason == "EVIDENCE_PATH_SYMLINK" or error.errno == errno.ELOOP:
        return "EVIDENCE_PATH_SYMLINK"
    return "EVIDENCE_BYTES_MISSING"
# #endregion Tooling.DrEvidenceCheck.Containment.OpenFailure


# #region Tooling.DrEvidenceCheck.Containment.OpenContained [C:5] [TYPE Function] [SEMANTICS dr,evidence,containment]
# @INVARIANT Every component opens relative to its pinned parent with O_NOFOLLOW; a parent swap cannot redirect the leaf.
# @POST Returns the fstat-validated regular inode; all intermediate descriptors close on return or failure.
def _open_contained(root_fd: int, relative_parts: tuple[str, ...]):
    """Open the exact regular leaf through pinned parent descriptors; close all helper fds."""
    opened: list[int] = []
    current = root_fd
    try:
        for index, part in enumerate(relative_parts):
            leaf = index == len(relative_parts) - 1
            reason = _classify_component(part, current)
            if reason is not None:
                return None, reason
            flags = _OPEN_FLAGS | getattr(os, "O_NOFOLLOW", 0)
            if not leaf:
                flags |= getattr(os, "O_DIRECTORY", 0)
            try:
                descriptor = os.open(part, flags, dir_fd=current)
            except OSError as error:
                return None, _open_failure(part, current, error)
            opened.append(descriptor)
            current = descriptor
        file_info = os.fstat(current)
        if not stat.S_ISREG(file_info.st_mode):
            return None, "EVIDENCE_BYTES_MISSING"
        if file_info.st_size <= 0:
            return None, "EVIDENCE_BYTES_EMPTY"
        return opened.pop(), None
    finally:
        for descriptor in opened:
            os.close(descriptor)
# #endregion Tooling.DrEvidenceCheck.Containment.OpenContained


# #region Tooling.DrEvidenceCheck.Containment.Sha256Descriptor [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _sha256_descriptor(descriptor: int) -> str:
    hasher = hashlib.sha256()
    with os.fdopen(descriptor, "rb") as stream:
        while chunk := stream.read(1024 * 1024):
            hasher.update(chunk)
    return hasher.hexdigest()
# #endregion Tooling.DrEvidenceCheck.Containment.Sha256Descriptor


# #region Tooling.DrEvidenceCheck.Containment.ArchiveNames [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _archive_names(path: str | None) -> set[str] | None:
    if not path:
        return None
    names = set()
    for line in Path(path).read_text(encoding="utf-8", errors="replace").splitlines():
        name = line.strip()
        if not name or name.endswith("/"):
            continue
        names.add(name[2:] if name.startswith("./") else name)
    return names
# #endregion Tooling.DrEvidenceCheck.Containment.ArchiveNames
# #endregion Tooling.DrEvidenceCheck.Containment