photoanalyzer/photo_pipeline/path_policy.py

"""Library boundary and exclusion policy — the single source used everywhere.

One function decides what is inside the library, what is a supported photo, and
what is excluded (`_IGNORE/` and generated thumbnail/cache directories). Symlinks
that resolve outside the library root are rejected rather than silently followed,
so discovery can never escape the configured boundary.

Extracted from photo_analyzer.discover_photos / SUPPORTED_EXTENSIONS
(see donor_ledger.yaml: pa-discovery).
"""

from __future__ import annotations

import os
from pathlib import Path
from typing import Iterable, Iterator

SUPPORTED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".heic", ".heif", ".tiff", ".tif"}
EXCLUDED_DIR_NAMES = {"_IGNORE", ".@__thumb"}


class PathPolicyError(ValueError):
    """A path violates the library boundary (e.g. a symlink escaping the root)."""


def is_supported(path: os.PathLike | str) -> bool:
    return Path(path).suffix.lower() in SUPPORTED_EXTENSIONS


def is_excluded(path: os.PathLike | str) -> bool:
    return any(part in EXCLUDED_DIR_NAMES for part in Path(path).parts)


def normalize_root(root: os.PathLike | str) -> Path:
    return Path(root).expanduser().resolve()


def resolve_within(root: Path, path: os.PathLike | str) -> Path:
    """Resolve ``path`` (following symlinks) and confirm it stays within ``root``.

    Raises ``PathPolicyError`` if the resolved path escapes the root.
    """
    root = Path(root).resolve()
    resolved = Path(path).resolve()
    if resolved != root and root not in resolved.parents:
        raise PathPolicyError(f"path escapes library root {root}: {path} -> {resolved}")
    return resolved


def resolve_in_roots(roots: Iterable[os.PathLike | str], path: os.PathLike | str) -> Path:
    """The resolved path, proven to be inside one of ``roots`` and not excluded.

    Callers must use the **returned** path for whatever they do next: validating one
    name and then opening another is the symlink race this exists to close (US07-02).
    The message names no path — it reaches API responses.

    With no roots configured there is no boundary to check; that is a property of the
    configuration, not permission granted to this call.
    """
    if is_excluded(path):
        raise PathPolicyError("path is inside an excluded (_IGNORE/) tree")
    roots = list(roots)
    if not roots:
        return Path(path)
    for root in roots:
        try:
            return resolve_within(Path(root), path)
        except PathPolicyError:
            continue
    raise PathPolicyError("path is outside the configured library roots")


def in_container() -> bool:
    """Whether this process is running inside a container image build of the app."""
    return Path("/.dockerenv").exists()


def _under_mount(path: Path) -> bool:
    """Whether ``path`` or one of its parents below ``/`` is a mounted filesystem."""
    current = path.resolve()
    while current != current.parent:
        if os.path.ismount(current):
            return True
        current = current.parent
    return False


def roots_refusal(roots: Iterable[os.PathLike | str], *, require_mount: bool = False) -> str | None:
    """Why the configured library roots cannot be worked with, or ``None`` (US08-03).

    The roots name the directories this installation renames folders in, rewrites
    EXIF in, and archives from. Container path policy is the same problem as host
    path policy with one new failure mode: the configured roots must name the
    *container-side* mount paths. A host path configured inside a container is
    either absent or an ordinary directory of the image, so the library looks empty
    and the first write lands in the container's throwaway layer instead of in the
    library. That is worth refusing at startup, while an operator is still watching,
    rather than at the first write.

    ``require_mount`` is the container-only half of that: inside a container a real
    library arrives through a bind mount, so a root that is not on (or under) a
    mount point is not the library the deployment meant.

    Writability is deliberately *not* checked: a bind mount's ownership is
    virtualised by Docker Desktop and Colima (it arrives as ``root:root``), so
    ``os.access`` there is evidence about the virtio layer rather than about the
    library. A wrong UID/GID surfaces as a refused rename with the real errno, which
    is at least true; a refusal here would be false on two supported platforms.
    """
    for root in roots:
        path = Path(root)
        if not path.exists():
            return (
                f"library root {path} does not exist: PHOTO_PIPELINE_LIBRARY_ROOTS must "
                "name paths that exist here, and in a container that means the mount path"
            )
        if not path.is_dir():
            return f"library root {path} is not a directory"
        if not os.access(path, os.R_OK | os.X_OK):
            return f"library root {path} is not readable by this process"
        if require_mount and not _under_mount(path):
            return (
                f"library root {path} is not on a mounted filesystem in this container: "
                "the library was not bind-mounted there, so PHOTO_PIPELINE_LIBRARY_ROOTS "
                "names a directory of the image rather than the library"
            )
    return None


def iter_supported_files(root: os.PathLike | str) -> Iterator[Path]:
    """Yield supported, non-excluded files under ``root`` in deterministic order.

    Excluded directories are never descended into. A symlinked file whose target
    escapes the root raises ``PathPolicyError`` (unsafe paths fail rather than
    leaking outside the library).
    """
    root = normalize_root(root)
    found: list[Path] = []
    for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
        dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDED_DIR_NAMES)
        for name in filenames:
            candidate = Path(dirpath) / name
            if not is_supported(candidate) or is_excluded(candidate):
                continue
            if candidate.is_symlink():
                resolve_within(root, candidate)  # raises on escape
            found.append(candidate)
    return iter(sorted(found))


def discover(roots: Iterable[os.PathLike | str]) -> list[Path]:
    """Deterministic, de-duplicated discovery across one or more roots."""
    seen: set[Path] = set()
    for root in roots:
        seen.update(iter_supported_files(root))
    return sorted(seen)