id_configurator/id_configurator/context.py

"""Execution context shared by all modules.

The context is responsible for enforcing that modules never write outside of
the destination directory, and gives read access to the (optional) source
filesystem used to detect the existing configuration.
"""

from __future__ import annotations

from pathlib import Path

from id_configurator.config import Configuration


class Context:
    def __init__(
        self, config: Configuration, dst: Path, src: Path | None = None
    ) -> None:
        self.config = config
        self.dst = dst.resolve()
        self.src = src.resolve() if src is not None else None

    def dst_path(self, relative: str) -> Path:
        """Resolve a path relative to the destination directory, refusing to escape it."""
        path = (self.dst / relative.lstrip("/")).resolve()
        if path != self.dst and self.dst not in path.parents:
            raise ValueError(
                f"refusing to write outside of the destination directory: {relative}"
            )
        return path

    def src_path(self, relative: str) -> Path | None:
        """Resolve a path relative to the source directory, or None if no source was given."""
        if self.src is None:
            return None
        return (self.src / relative.lstrip("/")).resolve()

    def write_file(self, relative: str, content: bytes, mode: int = 0o644) -> Path:
        path = self.dst_path(relative)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(content)
        path.chmod(mode)
        return path

    def read_src_file(self, relative: str) -> bytes | None:
        path = self.src_path(relative)
        if path is None or not path.is_file():
            return None
        return path.read_bytes()

    def iter_src_dir(self, relative: str) -> list[Path]:
        """List the files present in a directory relative to the source filesystem, if any."""
        path = self.src_path(relative)
        if path is None or not path.is_dir():
            return []
        return sorted(entry for entry in path.iterdir() if entry.is_file())

    def glob_src(self, pattern: str) -> list[Path]:
        """Glob a pattern relative to the source filesystem, or [] if no source was given."""
        if self.src is None:
            return []
        return sorted(self.src.glob(pattern))