"""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 collections.abc import Sequence from pathlib import Path from id_configurator.config import Configuration from id_configurator.users import ( DEFAULT_UID_MIN, PasswdEntry, parse_login_defs_uid_min, parse_passwd, ) class Context: def __init__( self, config: Configuration, dst: Path, src: Path | None = None, users: Sequence[str] | None = None, auto_users: bool = False, ) -> None: self.config = config self.dst = dst.resolve() self.src = src.resolve() if src is not None else None self._passwd_entries = self._load_passwd_entries() self._uid_min = self._load_uid_min() self.users = self._resolve_users(users or [], auto_users) def _load_passwd_entries(self) -> dict[str, PasswdEntry]: content = self.read_src_file("etc/passwd") if content is None: return {} return {entry.name: entry for entry in parse_passwd(content.decode("utf-8"))} def _load_uid_min(self) -> int: content = self.read_src_file("etc/login.defs") if content is None: return DEFAULT_UID_MIN uid_min = parse_login_defs_uid_min(content.decode("utf-8")) return DEFAULT_UID_MIN if uid_min is None else uid_min def _resolve_users(self, users: Sequence[str], auto_users: bool) -> list[str]: resolved: list[str] = [] for user in users: if user not in resolved: resolved.append(user) if auto_users: for name, entry in self._passwd_entries.items(): if name not in resolved and entry.is_regular_user(self._uid_min): resolved.append(name) return resolved def home_dir(self, user: str) -> str: """Relative path (no leading `/`) to `user`'s home directory, as used by modules that write per-user configuration files. Taken from ``'s `/etc/passwd` entry for `user` when there is one (the case for a user resolved via `--auto-users`); otherwise falls back to the usual `/home/` convention (`/root` for root). """ entry = self._passwd_entries.get(user) if entry is not None and entry.home: return entry.home.lstrip("/") return "root" if user == "root" else f"home/{user}" 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))