diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f619fd..51bf8b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,58 @@ All notable changes to this project are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.6.0] - 2026-06-27 + +### Added +- Current-directory target resolution from the `.ephemdir` marker. `keep`, + `rm`, `explain` and `extend` can now operate on the ephemdir directory you are + standing in (or any subdirectory of it) without an explicit name or path; the + target is always the managed root. They act only when the nearest `.ephemdir` + marker matches an active tracked entry by marker id and inode, and otherwise + fail closed rather than guessing. +- `ephemdir extend` accepts a bare lifetime when run inside a tracked directory: + `ephemdir extend 30m` and `ephemdir extend --forever` extend the current + directory, while `ephemdir extend 30m` keeps working as before. +- `ephemdir path` now prefers the current ephemdir directory when run from + inside one, preserving the most-recently-created fallback when run from + outside. A present-but-invalid marker makes `path` fail closed instead of + falling back. + +### Changed +- `install-service` now has a runtime-trust policy, selectable with + `--runtime-policy strict|balanced` (or the `EPHEMDIR_SERVICE_RUNTIME_POLICY` + environment variable). `balanced` is the default on macOS and `strict` the + default on Linux. `balanced` allows a group-writable directory ancestor of + the service runtime only as a narrow Homebrew/usr-local carve-out — macOS, a + path under `/opt/homebrew` or `/usr/local`, owned by root or you, not + world-writable, and whose owning group is a local administrator group + (`admin`) — emitting a warning. This lets a stock Homebrew interpreter host + the scheduled sweep, while a group-writable directory owned by an ordinary + shared group (which could contain another local user) is still rejected. + World-writable components, foreign-owned components, symlinked package + subdirectories, and group/world-writable executable files or + interpreter-startup hooks remain hard failures under both policies. + +### Fixed +- Directories were never removed after a reboot. The ownership identity check + compared the stored device number (`st_dev`) as well as the inode, but a + device number is not stable across reboots (macOS reassigns it for an APFS + volume at every boot, and it is not guaranteed stable on Linux). After a + restart every tracked directory therefore looked like a replacement and was + left untouched, defeating restart and expiry cleanup. The identity check now + compares the inode number only; the random ownership marker remains the + primary proof, and a directory replaced at the same path still gets a new + inode and is still detected. Within-operation safety checks (fd-relative + delete, mount-boundary detection) are unchanged and continue to use the full + device+inode pair. +- The scheduled sweep service could not be installed from a default Homebrew + Python on macOS, because the previous, unconditional rejection of any + group-writable runtime component refused `/opt/homebrew/Cellar`. Without an + installed service, periodic and post-reboot cleanup never ran automatically. + The new default `balanced` policy installs successfully on a stock Homebrew + setup while keeping the strict behaviour available. (Together with the inode + fix above, this is what restores automatic post-reboot cleanup on macOS.) + ## [0.5.0] - 2026-06-18 ### Added diff --git a/README.md b/README.md index 0f2b4f2..c9920ba 100644 --- a/README.md +++ b/README.md @@ -232,25 +232,46 @@ ephemdir uninstall-service `install-service` validates the persistent runtime before writing anything: every component of the interpreter and package paths must be owned by you -(or root) and must not be writable by other users. This deliberately rejects -**any** group/world-writable component — including sticky directories like -`/tmp` — so a virtualenv created under `/tmp` cannot host the scheduled -service even though it is fine for one-off interactive use. Install the venv -under your home directory instead. It also checks the interpreter-startup -hooks Python runs before ephemdir is imported (`.pth` files in site-packages, -`sitecustomize`, `pyvenv.cfg`, and the `tomli` package on Python 3.10). - -On macOS, `install-service` may reject a Homebrew or otherwise shared Python -runtime if any interpreter/package component is group/world-writable. That is -expected: launchd will run the interpreter later, so ephemdir refuses a runtime -another local user could modify after installation. For a dedicated service -runtime, use a private uv-managed virtual environment under your home directory: +(or root). It always rejects a **world-writable** component (including sticky +directories like `/tmp`, so a virtualenv under `/tmp` cannot host the scheduled +service), a **foreign-owned** component, a symlinked package subdirectory, and +any group/world-writable **executable** file or interpreter-startup hook +(`.pth` files in site-packages, `sitecustomize`, `pyvenv.cfg`, and the `tomli` +package on Python 3.10). + +How strictly it treats a merely **group-writable directory ancestor** is set by +the runtime policy: + +```bash +ephemdir install-service --interval 600 # default +ephemdir install-service --interval 600 --runtime-policy strict +ephemdir install-service --interval 600 --runtime-policy balanced +``` + +* **`balanced`** (the default on macOS) allows a group-writable directory + ancestor **only** as a narrow Homebrew/usr-local carve-out, printing a + warning. All of these must hold: macOS; the resolved path is under + `/opt/homebrew` or `/usr/local`; the directory is owned by root or you; it is + not world-writable; and its **owning group is a local administrator group** + (`admin`). This is exactly what a stock Homebrew interpreter needs — its + `/opt/homebrew/Cellar` ancestor is mode `0775`, group `admin` (the machine's + administrators, i.e. the owner on a personal Mac) — which `strict` rejects, + silently preventing the scheduled sweep from ever being installed. A + group-writable directory whose owning group is an ordinary shared group (which + could contain another unprivileged user) is **not** covered and is rejected. +* **`strict`** (the default off macOS) rejects **any** group-writable component. + Use it on a genuinely shared multi-user host. + +The default is also overridable with `EPHEMDIR_SERVICE_RUNTIME_POLICY=strict|balanced`. +If you would rather keep `strict` everywhere, install into a private uv-managed +virtual environment under your home directory, whose components are owned only +by you: ```bash uv python install 3.12 uv venv ~/.venvs/ephemdir-safe --python 3.12 uv pip install --python ~/.venvs/ephemdir-safe/bin/python ephemdir -~/.venvs/ephemdir-safe/bin/python -I -m ephemdir install-service +~/.venvs/ephemdir-safe/bin/python -I -m ephemdir install-service --runtime-policy strict ``` > **Trust boundary for `install-service`.** The scheduled job runs your Python @@ -264,12 +285,23 @@ uv pip install --python ~/.venvs/ephemdir-safe/bin/python ephemdir > do; on a shared multi-user host, ensure the environment is owned by you and > not group/world-writable before scheduling sweeps. -Prefer to wire it up yourself? The equivalents are: - -* **Linux (cron):** `*/10 * * * * ephemdir sweep` -* **macOS (launchd):** a `LaunchAgent` running `ephemdir sweep`; template in +Prefer to wire it up yourself? These are **manual alternatives, not equivalents** — +they are less hardened than `install-service` unless you reproduce all of its +properties: run the validated interpreter as `python -I -m ephemdir sweep` (not +a bare `ephemdir` from `PATH`), with working directory `/`, a fixed trusted +`PATH`, and pinned `EPHEMDIR_DATA_DIR` / `EPHEMDIR_CONFIG_DIR`. `install-service` +additionally validates the runtime and verifies the isolated import before +writing anything; a hand-written job does none of that. + +* **Linux (cron):** `*/10 * * * * /path/to/validated/python -I -m ephemdir sweep` +* **macOS (launchd):** a `LaunchAgent` running the same + `/path/to/validated/python -I -m ephemdir sweep`; template in [`packaging/`](packaging/). +Set `EPHEMDIR_DATA_DIR`/`EPHEMDIR_CONFIG_DIR` in the job's environment so the +scheduled sweep cannot drift to a different registry. When in doubt, prefer +`install-service`. + You can also keep a foreground watcher running: ```bash @@ -324,6 +356,28 @@ unique prefix (`bra` or `brave-otter` for `brave-otter-a81f42c9d047315b`). Add `-v` for more output or `-q` to stay quiet. +### Managing the current directory + +If you are inside a directory ephemdir created (or any subdirectory of it), the +`path`, `explain`, `extend`, `keep` and `rm` commands work with no name: + +```bash +cd "$(ephemdir new --lifetime 2h)" +ephemdir explain # describe the current ephemdir directory +ephemdir extend 30m # extend the current directory (or --forever) +ephemdir keep # keep the current directory; stop tracking it +ephemdir rm # delete the current directory's managed root +``` + +ephemdir looks for the nearest `.ephemdir` marker at or above your working +directory and applies the command only when that marker matches an active +tracked entry. From a subdirectory the target is always the managed **root**, +not the subdirectory. If the marker is missing, altered or does not match an +active entry, the command refuses to guess and exits with an error rather than +falling back to another directory. Outside any tracked directory these commands +report that there is no target; only `ephemdir path` keeps its old fallback to +the most recently created directory. + ### Listing with time left `ephemdir list` shows each directory's status at a glance: diff --git a/SECURITY.md b/SECURITY.md index c586709..35e130b 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -2,9 +2,9 @@ ## Supported Versions -The supported release lines are `0.4.x` and `0.5.x`. Supported runtimes are -Python 3.10+ on Linux and macOS. Windows is not supported until a handle-bound -recursive deletion backend is available. +The supported release line is `0.6.x` (with `0.5.x` still receiving security +fixes). Supported runtimes are Python 3.10+ on Linux and macOS. Windows is not +supported until a handle-bound recursive deletion backend is available. ## Reporting a Vulnerability @@ -49,7 +49,9 @@ unattended, later, as your user. Before writing any unit/plist it verifies that the interpreter, the entire `ephemdir` package tree (rejecting symlinked package subdirectories) and the interpreter-startup hooks (`.pth` files, `sitecustomize`, `pyvenv.cfg`, and `tomli` on Python 3.10) are owned by you or -root and not writable by other users. +root. A **world-writable** component, a **foreign-owned** component, a +symlinked package subdirectory, and any group/world-writable **executable** +file or startup hook are always rejected, under every policy. It also pins the verified effective `EPHEMDIR_DATA_DIR` and `EPHEMDIR_CONFIG_DIR` into the installed launchd/systemd definition so the scheduled sweep does not drift to a different registry after logout/login or @@ -67,8 +69,43 @@ confirm the environment's ownership and permissions before scheduling sweeps. One-off interactive `tempdir()`, `ephemdir sweep` and the rest of the CLI do not rely on this and are unaffected. -On macOS, a Homebrew or shared Python runtime can be rejected for scheduled -service use when any runtime component is group/world-writable. That rejection -is intentional; launchd runs the interpreter later, after the current shell is -gone. A safe pattern is to install the service from a private uv-managed venv -under the user's home directory. +### Runtime-trust policy: `strict` vs `balanced` + +The one place the policy is configurable is how a **group-writable directory +ancestor** of the runtime is treated. This is governed by +`--runtime-policy strict|balanced` (or `EPHEMDIR_SERVICE_RUNTIME_POLICY`). + +* **`strict`** rejects any group-writable component. It is the default on every + platform except macOS and is the correct choice on a genuinely shared + multi-user host. +* **`balanced`** (the default on macOS) allows a group-writable directory + ancestor only as a narrow, property-checked Homebrew/usr-local carve-out, with + a warning. **All** of the following must hold, or the component is rejected: + the platform is macOS; the resolved path is under `/opt/homebrew` or + `/usr/local`; the directory is owned by root or the installing user; it is not + world-writable; and its **owning group is a local administrator group** + (`admin`, gid 80). This reflects the single-user model: the `admin` group on a + personal Mac is the owner, not an attacker. It exists because a stock Homebrew + interpreter lives under `/opt/homebrew/Cellar` (mode `0775`, group `admin`); + under `strict` that ancestor is refused, which silently prevents the scheduled + sweep from ever being installed — and therefore prevents reboot/expiry cleanup + from running automatically. + +The owning-group check is the load-bearing restriction: POSIX write permission +on a directory lets any member of its group replace entries inside it, so a +group-writable directory on the import path whose group could contain a +*different* unprivileged user would be a code-execution vector for the scheduled +service. Restricting the carve-out to the local `admin` group (plus the prefix +allowlist) keeps the relaxation within the threat model. A group-writable +directory owned by an ordinary shared group is rejected even under `balanced`. + +`balanced` relaxes **only** group-writable *directory ancestors* that pass that +carve-out. World-writable components, foreign-owned components, symlinked +package subdirectories, and group/world-writable *executable* files or startup +hooks remain hard failures under both policies. If you prefer `strict` +everywhere, install the service from a private uv-managed venv under your home +directory, whose components are owned only by you. + +This threat model deliberately excludes root, the local administrator, and +other members of the owner's own `admin` group on a personal machine; it +defends against a different unprivileged local user, not against the owner. diff --git a/pyproject.toml b/pyproject.toml index b8f2069..b122a5a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "ephemdir" -version = "0.5.0" +version = "0.6.0" description = "Create self-cleaning ephemeral directories that vanish after a lifetime or on restart." readme = "README.md" requires-python = ">=3.10" diff --git a/src/ephemdir/__init__.py b/src/ephemdir/__init__.py index dae1f4d..a6852f8 100644 --- a/src/ephemdir/__init__.py +++ b/src/ephemdir/__init__.py @@ -45,7 +45,7 @@ tempdir, ) -__version__ = "0.5.0" +__version__ = "0.6.0" __author__ = "vindfjur" __all__ = [ diff --git a/src/ephemdir/_service.py b/src/ephemdir/_service.py index c9e9163..ff46adb 100644 --- a/src/ephemdir/_service.py +++ b/src/ephemdir/_service.py @@ -29,13 +29,21 @@ import sys import uuid from collections.abc import Iterator +from dataclasses import dataclass, field +from enum import Enum from pathlib import Path from ._platform import user_config_dir, user_data_dir from ._security import open_private_directory from ._trusted_exec import resolve_executable_in_dirs, trusted_system_dirs -__all__ = ["ServiceError", "install_service", "uninstall_service", "sweep_command"] +__all__ = [ + "ServiceError", + "RuntimePolicy", + "install_service", + "uninstall_service", + "sweep_command", +] # Scheduler/runtime subprocesses use fixed argv, trusted resolution and no shell. @@ -56,6 +64,79 @@ class ServiceError(RuntimeError): _SCHEDULER_TIMEOUT_SECONDS = 30 _SERVICE_FILE_MODE = 0o600 +# Environment override for the runtime-trust policy applied to install-service. +_RUNTIME_POLICY_ENV = "EPHEMDIR_SERVICE_RUNTIME_POLICY" + + +class RuntimePolicy(str, Enum): + """How strictly ``install-service`` validates the interpreter/package runtime. + + Both policies hard-fail on the real takeover vectors in ephemdir's + single-user POSIX threat model: a world-writable component, a component + owned by another user, a symlinked package subdirectory, and any + group/world-writable *executable* file or interpreter-startup hook. They + differ on one point only. + + ``strict`` additionally rejects any group-writable *directory* ancestor. + This is correct on a genuinely shared multi-user host, but on a normal + single-user macOS box it rejects every default Homebrew interpreter, whose + ``/opt/homebrew/Cellar`` ancestor is group-writable by ``admin`` (i.e. by + the owner). That refusal silently prevents the scheduled sweep from ever + being installed, so reboot/expiry cleanup never runs automatically. + + ``balanced`` allows a group-writable directory ancestor only as a narrow + macOS carve-out — when :func:`_is_trusted_homebrew_group_writable_dir` + passes: macOS, a directory owned by root or you, not world-writable, whose + owning group is a local administrator group (``admin``) and whose resolved + path is under ``/opt/homebrew`` or ``/usr/local``. It emits a warning. A + group-writable directory owned by an ordinary shared group (which could + contain another unprivileged user) is rejected, and every other vector + stays a hard failure. + """ + + STRICT = "strict" + BALANCED = "balanced" + + +@dataclass +class _RuntimeCheck: + """Carries the active policy and de-duplicated warnings through validation.""" + + policy: RuntimePolicy + warnings: dict[str, None] = field(default_factory=dict) + + def warn(self, message: str) -> None: + self.warnings.setdefault(message, None) + + +def _resolve_runtime_policy(explicit: str | RuntimePolicy | None) -> RuntimePolicy: + """Resolve the runtime policy from explicit arg, env, then platform default. + + Precedence: an explicit ``--runtime-policy`` value wins, then + ``EPHEMDIR_SERVICE_RUNTIME_POLICY``, then the platform default (macOS is + ``balanced`` so a stock Homebrew install works; every other platform is + ``strict``). + """ + if explicit is not None: + return _coerce_runtime_policy(explicit, source="--runtime-policy") + env_value = os.environ.get(_RUNTIME_POLICY_ENV) + if env_value: + return _coerce_runtime_policy(env_value, source=_RUNTIME_POLICY_ENV) + if sys.platform == "darwin": + return RuntimePolicy.BALANCED + return RuntimePolicy.STRICT + + +def _coerce_runtime_policy(value: str | RuntimePolicy, *, source: str) -> RuntimePolicy: + if isinstance(value, RuntimePolicy): + return value + try: + return RuntimePolicy(value.strip().lower()) + except ValueError: + raise ServiceError( + f"invalid {source} value {value!r}: expected 'strict' or 'balanced'" + ) from None + def _trusted_scheduler_dirs() -> tuple[Path, ...]: """Return fixed or kernel-derived directories trusted for schedulers.""" @@ -270,8 +351,77 @@ def _write_service_file(path: Path, content: str) -> None: os.close(dir_fd) -def _writable_by_other_users(mode: int) -> bool: - return bool(mode & (stat.S_IWGRP | stat.S_IWOTH)) + +# The balanced policy tolerates a group-writable *directory* ancestor only as a +# narrow macOS carve-out: the resolved path lives under one of these package- +# manager prefixes AND its owning group is a local administrator group (admin). +# /opt/homebrew/Cellar is mode 0775, group admin — writable only by machine +# administrators, who on a personal Mac are the owner. A group-writable +# directory whose owning group might contain a *different* unprivileged user (a +# shared "project" group) is deliberately NOT covered: that user is in scope and +# could replace a directory on the import path, gaining code execution as the +# installing user when the scheduled sweep runs. +_GROUP_WRITABLE_PREFIX_ALLOWLIST: tuple[str, ...] = ("/opt/homebrew", "/usr/local") +_ADMIN_GROUP_NAMES: tuple[str, ...] = ("admin",) +_ADMIN_GROUP_FALLBACK_GIDS: frozenset[int] = frozenset({80}) # macOS `admin` + + +def _trusted_admin_gids() -> frozenset[int]: + """Return gids of local administrator groups the carve-out may trust.""" + gids = set(_ADMIN_GROUP_FALLBACK_GIDS) + try: + import grp + except ImportError: # pragma: no cover - non-POSIX + return frozenset(gids) + for name in _ADMIN_GROUP_NAMES: + try: + gids.add(grp.getgrnam(name).gr_gid) + except KeyError: # pragma: no cover - admin group absent + continue + return frozenset(gids) + + +def _within_allowlisted_prefix(path: Path) -> bool: + """Whether ``path`` resolves to within an allowlisted package-manager prefix.""" + try: + real = path.resolve(strict=True) + except OSError: # pragma: no cover - component vanished mid-check + return False + for prefix in _GROUP_WRITABLE_PREFIX_ALLOWLIST: + prefix_path = Path(prefix) + if real == prefix_path or prefix_path in real.parents: + return True + return False + + +def _is_trusted_homebrew_group_writable_dir(current: Path, info: os.stat_result) -> bool: + """Whether ``balanced`` may tolerate this group-writable directory. + + Only the macOS Homebrew / usr-local carve-out qualifies: macOS, a real + directory, owning group is a local admin group, and the resolved path lives + under an allowlisted prefix. Any other group-writable directory — including + one whose owning group could contain another unprivileged local user — is + rejected, so the relaxation never exceeds the documented threat model. + """ + if sys.platform != "darwin": + return False + if not stat.S_ISDIR(info.st_mode): + return False + if info.st_gid not in _trusted_admin_gids(): + return False + return _within_allowlisted_prefix(current) + + +def _balanced_policy_hint(check: _RuntimeCheck, current: Path, info: os.stat_result) -> str: + """Mention balanced under strict only when the dir actually qualifies for it.""" + if check.policy is RuntimePolicy.STRICT and _is_trusted_homebrew_group_writable_dir( + current, info + ): + return ( + " This is a Homebrew/usr-local prefix owned by a local admin group; " + "--runtime-policy balanced (the macOS default) allows it." + ) + return "" def _safe_uv_runtime_hint() -> str: @@ -287,31 +437,55 @@ def _safe_uv_runtime_hint() -> str: ) -def _check_runtime_component(current: Path, info: os.stat_result, description: str) -> None: +def _check_runtime_component( + current: Path, info: os.stat_result, description: str, check: _RuntimeCheck +) -> None: """Reject one runtime path component another local user could replace. - Group/world-writability is not the only takeover vector: the owner of a - component can always rewrite it regardless of its mode, so every - component must belong to root or the installing user. A `0755` venv owned - by a different local user must never become a scheduled service runtime. + Ownership is the primary gate: the owner of a component can always rewrite + it regardless of mode, so every component must belong to root or the + installing user. World-writable is always fatal. A group-writable + *directory* ancestor is fatal under ``strict`` and is fatal under + ``balanced`` too, except for the narrow macOS Homebrew/usr-local admin + carve-out — when :func:`_is_trusted_homebrew_group_writable_dir` passes — + where it is allowed with a warning. A group-writable *file* — an interpreter + or importable module — stays fatal under both policies, because that is + executable code. """ if stat.S_ISLNK(info.st_mode): return # The resolved chain re-checks what the link points to. - if _writable_by_other_users(info.st_mode): + if hasattr(os, "geteuid") and info.st_uid not in (0, os.geteuid()): raise ServiceError( f"refusing to install service: {description} {current} is " - "group/world-writable (writable by other users); shared directories " + "owned by another user" + ) + if info.st_mode & stat.S_IWOTH: + raise ServiceError( + f"refusing to install service: {description} {current} is " + "world-writable (writable by any local user); shared directories " "like /tmp cannot host a scheduled service runtime, sticky bit or not." f"{_safe_uv_runtime_hint()}" ) - if hasattr(os, "geteuid") and info.st_uid not in (0, os.geteuid()): + if info.st_mode & stat.S_IWGRP: + if check.policy is RuntimePolicy.BALANCED and _is_trusted_homebrew_group_writable_dir( + current, info + ): + check.warn( + f"service runtime uses group-writable Homebrew/usr-local path " + f"component {current} (owning group is a local admin group); " + "allowed by --runtime-policy balanced. Use --runtime-policy " + "strict to fail closed." + ) + return raise ServiceError( f"refusing to install service: {description} {current} is " - "owned by another user" + "group-writable (writable by other users in its group)." + f"{_balanced_policy_hint(check, current, info)}" + f"{_safe_uv_runtime_hint()}" ) -def _validate_runtime_chain(absolute: Path, label: str) -> os.stat_result: +def _validate_runtime_chain(absolute: Path, label: str, check: _RuntimeCheck) -> os.stat_result: """Validate every ancestor of ``absolute`` (lexical and resolved). Each component must resist replacement by other local users. Returns the @@ -326,7 +500,7 @@ def _validate_runtime_chain(absolute: Path, label: str) -> os.stat_result: info = os.lstat(current) except OSError as error: raise ServiceError(f"cannot verify {label} path {current}: {error}") from error - _check_runtime_component(current, info, f"{label} path component") + _check_runtime_component(current, info, f"{label} path component", check) try: resolved = absolute.resolve(strict=True) @@ -343,7 +517,7 @@ def _validate_runtime_chain(absolute: Path, label: str) -> os.stat_result: raise ServiceError( f"cannot verify resolved {label} path {current}: {error}" ) from error - _check_runtime_component(current, info, f"resolved {label} path component") + _check_runtime_component(current, info, f"resolved {label} path component", check) try: return os.stat(absolute) @@ -351,7 +525,7 @@ def _validate_runtime_chain(absolute: Path, label: str) -> os.stat_result: raise ServiceError(f"cannot stat {label} path {absolute}: {error}") from error -def _validate_runtime_path(path: Path, label: str) -> None: +def _validate_runtime_path(path: Path, label: str, check: _RuntimeCheck) -> None: """Reject service runtime paths another local user could swap or edit. A scheduled job persists beyond the current shell, so every existing path @@ -361,14 +535,14 @@ def _validate_runtime_path(path: Path, label: str) -> None: remain supported. """ absolute = Path(os.path.abspath(path)) - final_info = _validate_runtime_chain(absolute, label) + final_info = _validate_runtime_chain(absolute, label, check) if not stat.S_ISREG(final_info.st_mode): raise ServiceError(f"refusing to install service: {label} is not a regular file") if hasattr(os, "geteuid") and final_info.st_uid not in (0, os.geteuid()): raise ServiceError(f"refusing to install service: {label} is owned by another user") -def _validate_runtime_dir(path: Path, label: str) -> None: +def _validate_runtime_dir(path: Path, label: str, check: _RuntimeCheck) -> None: """Reject a runtime *directory* another local user could swap or populate. Unlike :func:`_validate_runtime_path`, the final component must be a real @@ -379,17 +553,15 @@ def _validate_runtime_dir(path: Path, label: str) -> None: load. Validating the directory itself closes that gap. """ absolute = Path(os.path.abspath(path)) - final_info = _validate_runtime_chain(absolute, label) + final_info = _validate_runtime_chain(absolute, label, check) if not stat.S_ISDIR(final_info.st_mode): raise ServiceError(f"refusing to install service: {label} is not a directory") - if _writable_by_other_users(final_info.st_mode): - raise ServiceError( - f"refusing to install service: {label} {absolute} is " - "group/world-writable (writable by other users)." - f"{_safe_uv_runtime_hint()}" - ) - if hasattr(os, "geteuid") and final_info.st_uid not in (0, os.geteuid()): - raise ServiceError(f"refusing to install service: {label} is owned by another user") + # Re-apply the per-component policy to the resolved directory itself (the + # chain validated it through lstat; this covers the symlink-followed + # target). Balanced relaxes a group-writable directory here only when it + # passes the macOS Homebrew/usr-local admin carve-out + # (_is_trusted_homebrew_group_writable_dir); otherwise it is rejected. + _check_runtime_component(absolute, final_info, label, check) # Files the interpreter can load and execute as the service user. A `.py` @@ -398,7 +570,7 @@ def _validate_runtime_dir(path: Path, label: str) -> None: _EXECUTABLE_MODULE_SUFFIXES = (".py", ".pyc", ".pyo", ".so", ".pyd", ".dylib") -def _validate_package_tree(package_dir: Path) -> None: +def _validate_package_tree(package_dir: Path, check: _RuntimeCheck) -> None: """Verify every importable module under the package resists local tampering. ``python -I -m ephemdir sweep`` imports far more than one entry point: @@ -427,7 +599,7 @@ def _walk_error(error: OSError) -> None: # foreign `__pycache__` -- has nothing to validate indirectly today, # but its owner can later drop an unchecked `.pyc` the interpreter would # load. - _validate_runtime_dir(root_path, "ephemdir package directory") + _validate_runtime_dir(root_path, "ephemdir package directory", check) for name in list(dirs): sub = root_path / name # os.walk does not descend into a symlinked subdirectory, so a @@ -443,11 +615,11 @@ def _walk_error(error: OSError) -> None: # Validate each subdirectory *now*, before os.walk tries to descend. # An inaccessible or foreign-owned subdir is caught here (or by the # onerror handler above) rather than slipping through unvalidated. - _validate_runtime_dir(sub, "ephemdir package directory") + _validate_runtime_dir(sub, "ephemdir package directory", check) for name in files: file_path = root_path / name if file_path.suffix in _EXECUTABLE_MODULE_SUFFIXES: - _validate_runtime_path(file_path, "ephemdir module") + _validate_runtime_path(file_path, "ephemdir module", check) validated = True if not validated: raise ServiceError( @@ -510,10 +682,10 @@ def _iter_startup_files() -> Iterator[tuple[Path, str]]: yield pyvenv_cfg, "pyvenv.cfg" -def _validate_startup_environment() -> None: +def _validate_startup_environment(check: _RuntimeCheck) -> None: """Verify interpreter-startup hooks resist tampering by other local users.""" for path, label in _iter_startup_files(): - _validate_runtime_path(path, label) + _validate_runtime_path(path, label, check) # On Python 3.10 the sweep imports `tomli` for TOML config; the whole # package runs as the service user, so validate every module in it. @@ -523,13 +695,13 @@ def _validate_startup_environment() -> None: except (ImportError, AttributeError, ValueError): # pragma: no cover - defensive tomli_spec = None if tomli_spec is not None and tomli_spec.origin: - _validate_package_tree(Path(tomli_spec.origin).resolve().parent) + _validate_package_tree(Path(tomli_spec.origin).resolve().parent, check) -def _validate_service_runtime() -> None: - _validate_runtime_path(Path(sys.executable), "Python interpreter") - _validate_package_tree(Path(__file__).resolve().parent) - _validate_startup_environment() +def _validate_service_runtime(check: _RuntimeCheck) -> None: + _validate_runtime_path(Path(sys.executable), "Python interpreter", check) + _validate_package_tree(Path(__file__).resolve().parent, check) + _validate_startup_environment(check) def _reject_elevated_user_install() -> None: @@ -885,9 +1057,19 @@ def _uninstall_windows() -> str: # --- Public dispatch ------------------------------------------------------- -def install_service(interval: int = 600) -> str: +def install_service( + interval: int = 600, + *, + runtime_policy: str | RuntimePolicy | None = None, +) -> str: """Install the periodic sweep service for the current platform. + ``runtime_policy`` selects how strictly the interpreter/package runtime is + validated (see :class:`RuntimePolicy`); ``None`` resolves it from the + ``EPHEMDIR_SERVICE_RUNTIME_POLICY`` env var, then the platform default + (``balanced`` on macOS, ``strict`` elsewhere). Any relaxation taken under + ``balanced`` is logged as a warning. + Raises :class:`ServiceError` when the platform scheduler reports failure, so a broken installation is never reported as success. """ @@ -899,8 +1081,11 @@ def install_service(interval: int = 600) -> str: "Windows is unsupported because Python does not expose the " "handle-bound recursive deletion primitives ephemdir requires" ) - _validate_service_runtime() + check = _RuntimeCheck(policy=_resolve_runtime_policy(runtime_policy)) + _validate_service_runtime(check) _verify_isolated_import() + for warning in check.warnings: + logger.warning("%s", warning) if sys.platform == "darwin": return _install_launchd(interval) return _install_systemd(interval) diff --git a/src/ephemdir/cli.py b/src/ephemdir/cli.py index 5a95df5..c643907 100644 --- a/src/ephemdir/cli.py +++ b/src/ephemdir/cli.py @@ -37,11 +37,14 @@ from ._service import ServiceError, install_service, uninstall_service from .core import ( _UNSET, + _current_target_path, + _CurrentTargetNotFound, _path_state, dir_status, explain, extend, keep, + parse_lifetime, plan_sweep, prune, recover, @@ -293,7 +296,8 @@ def _cmd_sweep(args: argparse.Namespace) -> int: def _cmd_explain(args: argparse.Namespace) -> int: try: - decision = explain(args.target) + target = args.target if args.target is not None else _current_target_path() + decision = explain(target) except LookupError as error: logger.error("%s", error) return 1 @@ -389,10 +393,17 @@ def _cmd_list(args: argparse.Namespace) -> int: def _cmd_path(args: argparse.Namespace) -> int: try: - if args.target is None: - path = _latest_tracked() - else: + if args.target is not None: path = resolve(args.target) + else: + # Inside a tracked ephemdir directory, print its root. Outside one, + # preserve the old fallback to the most recently created directory. + # A present-but-invalid marker (_CurrentTargetMismatch) fails closed + # and is NOT silently replaced by the latest fallback. + try: + path = _current_target_path() + except _CurrentTargetNotFound: + path = _latest_tracked() except LookupError as error: logger.error("%s", error) return 1 @@ -416,7 +427,8 @@ def _latest_tracked() -> Path: def _cmd_keep(args: argparse.Namespace) -> int: try: - path = keep(args.target) + target = args.target if args.target is not None else _current_target_path() + path = keep(target) except LookupError as error: logger.error("%s", error) return 1 @@ -425,28 +437,60 @@ def _cmd_keep(args: argparse.Namespace) -> int: return 0 +def _looks_like_lifetime(text: str) -> bool: + """Whether ``text`` parses as a lifetime (used to disambiguate `extend`).""" + try: + parse_lifetime(text) + except (ValueError, TypeError): + return False + return True + + def _cmd_extend(args: argparse.Namespace) -> int: - if args.lifetime is None and not args.forever: + # Grammar (target optional when inside a tracked ephemdir directory): + # extend extend --forever + # extend extend --forever + arg1, arg2, forever = args.arg1, args.arg2, args.forever + if arg2 is not None: + if forever: + logger.error("--forever cannot be combined with a lifetime") + return 2 + target, lifetime_str = arg1, arg2 + elif arg1 is not None: + if forever: + if _looks_like_lifetime(arg1): + # `extend --forever 2h` (a lone lifetime + --forever) is a + # combination error, not a directory named "2h". + logger.error("--forever cannot be combined with a lifetime") + return 2 + target, lifetime_str = arg1, None # extend --forever + elif _looks_like_lifetime(arg1): + target, lifetime_str = None, arg1 # extend (current) + else: + logger.error("specify a lifetime (e.g. 2h) or --forever") + return 2 + elif forever: + target, lifetime_str = None, None # extend --forever (current) + else: logger.error("specify a lifetime (e.g. 2h) or --forever") return 2 - if args.lifetime is not None and args.forever: - logger.error("--forever cannot be combined with a lifetime") - return 2 try: - path = extend(args.target, None if args.forever else args.lifetime) + resolved = target if target is not None else _current_target_path() + path = extend(resolved, None if forever else lifetime_str) except (LookupError, ValueError) as error: logger.error("%s", error) return 1 - if args.forever: + if forever: logger.warning("extended %s -- no time limit (restart policy still applies)", path) else: - logger.warning("extended %s by %s from now", path, args.lifetime) + logger.warning("extended %s by %s from now", path, lifetime_str) return 0 def _cmd_rm(args: argparse.Namespace) -> int: try: - path = remove(args.target) + target = args.target if args.target is not None else _current_target_path() + path = remove(target) except (LookupError, OSError) as error: logger.error("%s", error) return 1 @@ -504,7 +548,9 @@ def _detect_shell() -> str: def _cmd_install_service(args: argparse.Namespace) -> int: try: - message = install_service(interval=args.interval) + message = install_service( + interval=args.interval, runtime_policy=args.runtime_policy + ) except (ServiceError, ValueError) as error: logger.error("%s", error) return 1 @@ -572,25 +618,31 @@ def build_parser() -> argparse.ArgumentParser: path_cmd = sub.add_parser( "path", help="print the path of a tracked directory (by name, prefix or path)") path_cmd.add_argument("target", nargs="?", default=None, - help="directory name, unique prefix or path " - "(default: most recently created)") + help="directory name, unique prefix or path (default: the " + "current ephemdir directory, else the most recently created)") path_cmd.set_defaults(func=_cmd_path) keep_cmd = sub.add_parser( "keep", help="stop tracking a directory so it is never auto-removed") - keep_cmd.add_argument("target", help="directory name, unique prefix or path") + keep_cmd.add_argument("target", nargs="?", default=None, + help="directory name, unique prefix or path " + "(default: the current ephemdir directory)") keep_cmd.set_defaults(func=_cmd_keep) extend_cmd = sub.add_parser("extend", help="give a directory a fresh lifetime from now") - extend_cmd.add_argument("target", help="directory name, unique prefix or path") - extend_cmd.add_argument("lifetime", nargs="?", default=None, - help='new time to live, e.g. "2h" or "1d"') + extend_cmd.add_argument("arg1", nargs="?", default=None, + help="directory name/prefix/path, or a lifetime like " + '"2h" to extend the current ephemdir directory') + extend_cmd.add_argument("arg2", nargs="?", default=None, + help='new time to live when a target is given, e.g. "2h" or "1d"') extend_cmd.add_argument("--forever", action="store_true", help="remove the time limit (restart policy still applies)") extend_cmd.set_defaults(func=_cmd_extend) rm = sub.add_parser("rm", help="remove a tracked directory now") - rm.add_argument("target", help="directory name, unique prefix or path") + rm.add_argument("target", nargs="?", default=None, + help="directory name, unique prefix or path " + "(default: the current ephemdir directory)") rm.set_defaults(func=_cmd_rm) sweep_cmd = sub.add_parser("sweep", help="remove directories that are due for cleanup") @@ -601,7 +653,9 @@ def build_parser() -> argparse.ArgumentParser: sweep_cmd.set_defaults(func=_cmd_sweep) explain_cmd = sub.add_parser("explain", help="explain cleanup state for a directory") - explain_cmd.add_argument("target", help="directory name, unique prefix or path") + explain_cmd.add_argument("target", nargs="?", default=None, + help="directory name, unique prefix or path " + "(default: the current ephemdir directory)") explain_cmd.set_defaults(func=_cmd_explain) doctor_cmd = sub.add_parser("doctor", help="diagnose ephemdir safety prerequisites") @@ -665,6 +719,14 @@ def build_parser() -> argparse.ArgumentParser: help="install a scheduled sweep service for this platform") install.add_argument("--interval", type=int, default=600, help="seconds between sweeps (default: 600)") + install.add_argument( + "--runtime-policy", + choices=["strict", "balanced"], + default=None, + help="runtime-trust policy for the service interpreter/package " + "(default: balanced on macOS, strict elsewhere; " + "env EPHEMDIR_SERVICE_RUNTIME_POLICY also applies)", + ) install.set_defaults(func=_cmd_install_service) uninstall = sub.add_parser("uninstall-service", help="remove the scheduled sweep service") diff --git a/src/ephemdir/core.py b/src/ephemdir/core.py index 2a52881..40fd2bb 100644 --- a/src/ephemdir/core.py +++ b/src/ephemdir/core.py @@ -315,28 +315,41 @@ def _fsync_directory(path: Path) -> None: def _inode_matches(path: Path, entry: Entry) -> bool | None: - """Compare ``path`` against the entry's stored inode. + """Compare ``path`` against the entry's stored inode *number*. + + Only ``st_ino`` is compared, never ``st_dev``. A device number is not + stable across reboots — macOS reassigns it for an APFS volume at every boot, + and it is not guaranteed stable on Linux either — so comparing it would mark + every tracked directory ``foreign`` after a restart and silently defeat + restart/expiry cleanup (the directory is never deleted). The on-disk inode + number is stable across reboots, and the random ownership marker is the + primary proof of ownership; the inode is a secondary cross-check that still + distinguishes a replacement created at the same path (which gets a new + inode number). Returns ``True``/``False`` for a definite answer and ``None`` when the entry carries no inode information to compare against. """ - dev, ino = entry.get("dev"), entry.get("ino") - if not (isinstance(dev, int) and isinstance(ino, int)): + ino = entry.get("ino") + if not isinstance(ino, int): return None try: path_stat = os.stat(path, follow_symlinks=False) except OSError: return False - return (path_stat.st_dev, path_stat.st_ino) == (dev, ino) + return path_stat.st_ino == ino def _fd_inode_matches(fd: int, entry: Entry) -> bool | None: - """Compare an already-open directory fd against the entry's stored inode.""" - dev, ino = entry.get("dev"), entry.get("ino") - if not (isinstance(dev, int) and isinstance(ino, int)): + """Compare an already-open directory fd against the entry's stored inode number. + + Like :func:`_inode_matches`, only ``st_ino`` is compared so the check is + stable across reboots. + """ + ino = entry.get("ino") + if not isinstance(ino, int): return None - fd_stat = os.fstat(fd) - return (fd_stat.st_dev, fd_stat.st_ino) == (dev, ino) + return os.fstat(fd).st_ino == ino def _is_real_directory(path: Path) -> bool: @@ -409,7 +422,7 @@ def _staging_ownership(original: Path, staging: Path, entry: Entry) -> str: marker. The inode recorded at claim time is a necessary cross-check but never sufficient on its own: filesystems such as ext4 and tmpfs reuse an inode number the instant the original tree is removed, so a newcomer - created at the same private path can inherit the recorded ``(dev, ino)``. + created at the same private path can inherit the recorded inode number. When the marker is recorded but gone, the result is therefore ambiguous (``"unverified"``) and recovery parks it rather than deleting a possible replacement. @@ -1440,6 +1453,86 @@ def resolve(target: str | os.PathLike[str], *, registry: Registry | None = None) return path +# --- Current-directory target resolution ----------------------------------- +# +# Commands like `keep`, `rm`, `explain` and `extend` can act on the ephemdir +# directory the user is standing in, with no name argument. We find the nearest +# `.ephemdir` marker at or above the current directory and act on it only when +# it matches an active registry entry by marker id and inode -- never guessing. + +_CURRENT_NOTFOUND_MESSAGE = ( + "no target provided and current directory is not inside a tracked ephemdir " + "directory; pass a name/path or cd into one" +) +_CURRENT_MISMATCH_MESSAGE = ( + "current directory contains an .ephemdir marker, but it does not match an " + "active tracked ephemdir entry; refusing to guess" +) + + +class _CurrentTargetNotFound(LookupError): + """No ``.ephemdir`` marker was found at or above the start directory.""" + + +class _CurrentTargetMismatch(LookupError): + """A marker was found but does not match an active tracked entry.""" + + +def _marker_present(directory: Path) -> bool: + """Whether ``directory`` holds an ``.ephemdir`` entry (even a broken link).""" + return os.path.lexists(directory / _MARKER_NAME) + + +def _resolve_current_target( + *, registry: Registry, start: Path | None = None +) -> tuple[Path, Entry]: + """Resolve the active tracked ephemdir root containing ``start`` (or cwd). + + Walks upward from ``start`` and stops at the **nearest** directory holding + an ``.ephemdir`` marker. That directory is returned only when it is an + active registry entry whose marker id and inode match (``_ownership`` is + ``"ours"``) and whose runtime is compatible. A present-but-unmatched marker + raises :class:`_CurrentTargetMismatch` — it never falls through to a marker + higher up, and never guesses. No marker anywhere raises + :class:`_CurrentTargetNotFound`. The resolver performs no registry + mutations. + """ + reg = registry or Registry() + try: + base = Path.cwd() if start is None else Path(start) + except OSError as error: + raise _CurrentTargetNotFound(_CURRENT_NOTFOUND_MESSAGE) from error + base = _canonical_private_dir_path(base) + state = { + key: entry + for key, entry in reg.load(read_only=True).items() + if entry.get("state", "active") == "active" + } + for directory in (base, *base.parents): + if not _marker_present(directory): + continue + entry = state.get(str(directory)) + if ( + entry is not None + and _ownership(directory, entry) == "ours" + and not _entry_compatibility_blockers(entry) + ): + return directory, dict(entry) + # A marker is here but it does not match an active, owned entry. Refuse + # to guess and refuse to walk past it to a marker further up. + raise _CurrentTargetMismatch(_CURRENT_MISMATCH_MESSAGE) + raise _CurrentTargetNotFound(_CURRENT_NOTFOUND_MESSAGE) + + +def _current_target_path( + *, registry: Registry | None = None, start: Path | None = None +) -> Path: + """Return the path of the active ephemdir directory containing the cwd.""" + reg = registry or Registry() + path, _ = _resolve_current_target(registry=reg, start=start) + return path + + def keep(target: str | os.PathLike[str], *, registry: Registry | None = None) -> Path: """Stop tracking a directory without deleting it; return its path. diff --git a/tests/test_cli.py b/tests/test_cli.py index 9042baf..64249bd 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -460,10 +460,10 @@ def test_shell_init_autodetects_known_and_default_shell(capsys, monkeypatch): def test_install_and_uninstall_service_cli(capsys, monkeypatch): monkeypatch.setattr( "ephemdir.cli.install_service", - lambda interval: f"installed every {interval}s", + lambda interval, runtime_policy: f"installed every {interval}s ({runtime_policy})", ) - assert main(["install-service", "--interval", "7"]) == 0 - assert "installed every 7s" in capsys.readouterr().err + assert main(["install-service", "--interval", "7", "--runtime-policy", "strict"]) == 0 + assert "installed every 7s (strict)" in capsys.readouterr().err monkeypatch.setattr("ephemdir.cli.uninstall_service", lambda: "removed") assert main(["uninstall-service"]) == 0 @@ -471,7 +471,7 @@ def test_install_and_uninstall_service_cli(capsys, monkeypatch): def test_service_cli_reports_errors(capsys, monkeypatch): - def install_failure(interval): + def install_failure(interval, runtime_policy): raise ValueError("bad interval") def uninstall_failure(): diff --git a/tests/test_core.py b/tests/test_core.py index ee99a05..7ee1db1 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -187,14 +187,22 @@ def test_core_inode_and_directory_helpers(tmp_path, monkeypatch): info = path.stat() assert core_module._inode_matches(path, {}) is None - assert core_module._inode_matches(path, {"dev": info.st_dev, "ino": info.st_ino}) is True - assert core_module._inode_matches(path, {"dev": info.st_dev, "ino": info.st_ino + 1}) is False - assert core_module._inode_matches(tmp_path / "missing", {"dev": 1, "ino": 2}) is False + assert core_module._inode_matches(path, {"ino": info.st_ino}) is True + assert core_module._inode_matches(path, {"ino": info.st_ino + 1}) is False + assert core_module._inode_matches(tmp_path / "missing", {"ino": 2}) is False + # The device number is deliberately ignored: it is not stable across reboots, + # so a stale st_dev with a matching st_ino must still count as a match. + assert core_module._inode_matches( + path, {"dev": info.st_dev + 9999, "ino": info.st_ino} + ) is True fd = os.open(path, os.O_RDONLY) try: assert core_module._fd_inode_matches(fd, {}) is None - assert core_module._fd_inode_matches(fd, {"dev": info.st_dev, "ino": info.st_ino}) is True + assert core_module._fd_inode_matches(fd, {"ino": info.st_ino}) is True + assert core_module._fd_inode_matches( + fd, {"dev": info.st_dev + 9999, "ino": info.st_ino} + ) is True finally: os.close(fd) diff --git a/tests/test_current_target.py b/tests/test_current_target.py new file mode 100644 index 0000000..0cd1c3c --- /dev/null +++ b/tests/test_current_target.py @@ -0,0 +1,250 @@ +"""Tests for current-directory target resolution (0.6.0 Tier 1).""" + +from __future__ import annotations + +import os +import time + +import pytest + +from ephemdir.cli import main +from ephemdir.core import ( + _MARKER_NAME, + _CurrentTargetMismatch, + _CurrentTargetNotFound, + _resolve_current_target, + registered, + tempdir, +) + +# --- Core resolver: detection ------------------------------------------------ + + +def test_current_target_detects_root_marker(tmp_path, registry): + d = tempdir(parent=tmp_path, registry=registry) + path, entry = _resolve_current_target(registry=registry, start=d.path) + assert path == d.path + assert entry["marker_id"] == registered(registry=registry)[str(d.path)]["marker_id"] + + +def test_current_target_detects_from_child_directory(tmp_path, registry): + d = tempdir(parent=tmp_path, registry=registry) + child = d.path / "a" / "b" + child.mkdir(parents=True) + path, _ = _resolve_current_target(registry=registry, start=child) + assert path == d.path # the managed root, not the child + + +def test_current_target_uses_nearest_nested_marker(tmp_path, registry): + outer = tempdir(parent=tmp_path, registry=registry) + inner = tempdir(parent=outer.path, registry=registry) + start = inner.path / "sub" + start.mkdir() + path, _ = _resolve_current_target(registry=registry, start=start) + assert path == inner.path # nearest marker wins over the outer one + + +# --- Core resolver: fail-closed rejections ----------------------------------- + + +def test_current_target_rejects_invalid_marker(tmp_path, registry): + d = tempdir(parent=tmp_path, registry=registry) + (d.path / _MARKER_NAME).write_text("not-32-hex-garbage", encoding="utf-8") + with pytest.raises(_CurrentTargetMismatch): + _resolve_current_target(registry=registry, start=d.path) + + +@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlink support required") +def test_current_target_rejects_marker_symlink(tmp_path, registry): + d = tempdir(parent=tmp_path, registry=registry) + marker = d.path / _MARKER_NAME + marker.unlink() + marker.symlink_to(tmp_path / "elsewhere") + with pytest.raises(_CurrentTargetMismatch): + _resolve_current_target(registry=registry, start=d.path) + + +def test_current_target_rejects_registry_marker_mismatch(tmp_path, registry): + d = tempdir(parent=tmp_path, registry=registry) + # A different but well-formed marker no longer matches the registry id. + (d.path / _MARKER_NAME).write_text("0" * 32, encoding="utf-8") + with pytest.raises(_CurrentTargetMismatch): + _resolve_current_target(registry=registry, start=d.path) + + +def test_current_target_rejects_replaced_directory(tmp_path, registry): + d = tempdir(parent=tmp_path, registry=registry) + import shutil + + shutil.rmtree(d.path) + d.path.mkdir() + (d.path / _MARKER_NAME).write_text("0" * 32, encoding="utf-8") # foreign marker + with pytest.raises(_CurrentTargetMismatch): + _resolve_current_target(registry=registry, start=d.path) + + +def test_current_target_broken_inner_marker_does_not_reach_outer(tmp_path, registry): + # Nearest-invalid-marker stops the search: a broken inner marker must fail + # closed, never fall through to a valid outer marker. + outer = tempdir(parent=tmp_path, registry=registry) + inner = tempdir(parent=outer.path, registry=registry) + (inner.path / _MARKER_NAME).write_text("garbage", encoding="utf-8") + start = inner.path / "sub" + start.mkdir() + with pytest.raises(_CurrentTargetMismatch): + _resolve_current_target(registry=registry, start=start) + + +def test_current_target_not_found_outside_any_ephemdir(tmp_path, registry): + plain = tmp_path / "plain" + plain.mkdir() + with pytest.raises(_CurrentTargetNotFound): + _resolve_current_target(registry=registry, start=plain) + + +def test_failed_resolution_does_not_mutate_registry(tmp_path, registry): + d = tempdir(parent=tmp_path, registry=registry) + (d.path / _MARKER_NAME).write_text("not-hex", encoding="utf-8") + before = registered(registry=registry) + with pytest.raises(_CurrentTargetMismatch): + _resolve_current_target(registry=registry, start=d.path) + assert registered(registry=registry) == before + + +# --- CLI: no-target commands act on the current directory -------------------- + + +def test_keep_without_target_uses_current_directory(tmp_path, monkeypatch, capsys): + d = tempdir(parent=tmp_path) + monkeypatch.chdir(d.path) + assert main(["keep"]) == 0 + assert d.path.is_dir() # kept on disk + assert str(d.path) not in registered() # no longer tracked + assert not (d.path / _MARKER_NAME).exists() # our marker removed + + +def test_rm_without_target_uses_current_directory_root(tmp_path, monkeypatch): + d = tempdir(parent=tmp_path) + child = d.path / "a" / "b" + child.mkdir(parents=True) + monkeypatch.chdir(child) + assert main(["rm"]) == 0 + assert not d.path.exists() # the managed root was removed, not just the child + assert str(d.path) not in registered() + + +def test_explain_without_target_uses_current_directory(tmp_path, monkeypatch, capsys): + d = tempdir(parent=tmp_path) + monkeypatch.chdir(d.path) + assert main(["explain"]) == 0 + assert str(d.path) in capsys.readouterr().out + + +def test_extend_without_target_uses_current_directory(tmp_path, monkeypatch): + d = tempdir(parent=tmp_path, lifetime="1h") + monkeypatch.chdir(d.path) + assert main(["extend", "30m"]) == 0 + entry = registered()[str(d.path)] + expires = entry["expires_at"] + assert expires is not None + assert abs(float(expires) - (time.time() + 30 * 60)) < 30 + + +def test_extend_without_target_forever(tmp_path, monkeypatch): + d = tempdir(parent=tmp_path, lifetime="1h") + monkeypatch.chdir(d.path) + assert main(["extend", "--forever"]) == 0 + assert registered()[str(d.path)]["expires_at"] is None + + +def test_extend_forever_with_lifetime_is_usage_error(tmp_path, monkeypatch, capsys): + d = tempdir(parent=tmp_path, lifetime="1h") + monkeypatch.chdir(d.path) + # `extend --forever 30m` must not silently treat 30m as a target name. + assert main(["extend", "--forever", "30m"]) == 2 + assert "cannot be combined with a lifetime" in capsys.readouterr().err + assert registered()[str(d.path)]["expires_at"] is not None # unchanged + + +def test_extend_with_explicit_target_still_works(tmp_path, monkeypatch): + d = tempdir(parent=tmp_path, lifetime="1h") + # Run from elsewhere; the explicit name must still resolve. + monkeypatch.chdir(tmp_path) + assert main(["extend", d.path.name, "45m"]) == 0 + entry = registered()[str(d.path)] + assert abs(float(entry["expires_at"]) - (time.time() + 45 * 60)) < 30 + + +# --- CLI: outside context, fail closed without side effects ------------------ + + +def test_extend_without_target_outside_context_fails(tmp_path, monkeypatch, capsys): + tempdir(parent=tmp_path, lifetime="1h") # exists but we are not inside it + outside = tmp_path / "outside" + outside.mkdir() + monkeypatch.chdir(outside) + before = registered() + assert main(["extend", "30m"]) == 1 + assert registered() == before # no mutation + assert "Traceback" not in capsys.readouterr().err + + +@pytest.mark.parametrize("argv", [["keep"], ["rm"], ["extend", "30m"]]) +def test_no_target_destructive_commands_do_not_fallback_to_latest( + tmp_path, monkeypatch, argv +): + survivor = tempdir(parent=tmp_path, lifetime="1h") + outside = tmp_path / "outside" + outside.mkdir() + monkeypatch.chdir(outside) + assert main(argv) == 1 + assert survivor.path.is_dir() # latest was never touched + assert str(survivor.path) in registered() + + +def test_no_target_error_has_no_traceback(tmp_path, monkeypatch, capsys): + outside = tmp_path / "outside" + outside.mkdir() + monkeypatch.chdir(outside) + assert main(["rm"]) == 1 + err = capsys.readouterr().err + assert "Traceback" not in err + assert "not inside a tracked ephemdir directory" in err + + +# --- CLI: path command priority --------------------------------------------- + + +def test_path_prefers_current_context_over_latest(tmp_path, monkeypatch, capsys): + first = tempdir(parent=tmp_path) + time.sleep(0.01) + latest = tempdir(parent=tmp_path) # most recently created + monkeypatch.chdir(first.path) # but cwd is inside the first + assert main(["path"]) == 0 + out = capsys.readouterr().out.strip() + assert out == str(first.path) + assert out != str(latest.path) + + +def test_path_outside_context_preserves_latest_fallback(tmp_path, monkeypatch, capsys): + tempdir(parent=tmp_path) + time.sleep(0.01) + latest = tempdir(parent=tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + monkeypatch.chdir(outside) + assert main(["path"]) == 0 + assert capsys.readouterr().out.strip() == str(latest.path) + + +def test_path_inside_invalid_marker_fails_instead_of_latest_fallback( + tmp_path, monkeypatch, capsys +): + tempdir(parent=tmp_path) # a valid 'latest' that must NOT be returned + broken = tempdir(parent=tmp_path) + (broken.path / _MARKER_NAME).write_text("garbage", encoding="utf-8") + monkeypatch.chdir(broken.path) + assert main(["path"]) == 1 # fail closed, no fallback + out = capsys.readouterr() + assert out.out.strip() == "" + assert "refusing to guess" in out.err diff --git a/tests/test_hardening.py b/tests/test_hardening.py index 39e7ccc..81c6b9b 100644 --- a/tests/test_hardening.py +++ b/tests/test_hardening.py @@ -1220,3 +1220,60 @@ def test_sweep_refuses_writable_registry_and_keeps_real_directory(tmp_path): assert d.path.exists() assert registry.path.exists() assert not list(tmp_path.glob("registry.json.corrupt-*")) + + +# --- Reboot stability: st_dev is not stable across reboots ------------------- + +def test_ownership_survives_device_number_change(tmp_path, registry, monkeypatch): + # macOS reassigns an APFS volume's st_dev at every boot. The identity check + # must compare st_ino only, or every tracked directory would look 'foreign' + # after a restart and never be cleaned up (the real 0.6.0 reboot bug). + d = tempdir(parent=tmp_path, registry=registry) + entry = registered(registry=registry)[str(d.path)] + real_stat = os.stat + + def reboot_stat(path, *args, **kwargs): + result = real_stat(path, *args, **kwargs) + if str(path) == str(d.path): + values = list(result) + values[2] = result.st_dev + 99999 # st_dev (index 2) changes; st_ino stays + return os.stat_result(values) + return result + + monkeypatch.setattr(core.os, "stat", reboot_stat) + assert core._ownership(d.path, entry) == "ours" + + +def test_ownership_still_detects_replacement_by_inode(tmp_path, registry, monkeypatch): + # Dropping st_dev must NOT weaken replacement detection: a different st_ino + # at the same path (a directory replaced after the entry was recorded) is + # still foreign and never auto-deleted. + d = tempdir(parent=tmp_path, registry=registry) + entry = registered(registry=registry)[str(d.path)] + real_stat = os.stat + + def replaced_stat(path, *args, **kwargs): + result = real_stat(path, *args, **kwargs) + if str(path) == str(d.path): + values = list(result) + values[1] = result.st_ino + 1 # st_ino (index 1) changes + return os.stat_result(values) + return result + + monkeypatch.setattr(core.os, "stat", replaced_stat) + assert core._ownership(d.path, entry) == "foreign" + + +def test_restarted_directory_with_stale_device_number_is_swept(tmp_path, registry, monkeypatch): + # End-to-end: a restart renumbers the device, so the entry's recorded dev no + # longer matches, but st_ino and the marker do. Restart cleanup must still + # delete the directory instead of leaving it tracked-but-foreign forever. + d = tempdir(parent=tmp_path, registry=registry) # remove_on_restart=True by default + with registry.transaction() as state: + e = state[str(d.path)] + e["dev"] = int(e["dev"]) + 99999 # as if written under a previous boot + e["boot_id"] = "old-boot-session" + monkeypatch.setattr(core, "boot_session_id", lambda: "new-boot-session") + assert sweep(registry=registry) == 1 + assert not d.path.exists() + assert str(d.path) not in registered(registry=registry) diff --git a/tests/test_service.py b/tests/test_service.py index 8dc5de1..e0eb982 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -12,12 +12,21 @@ from ephemdir._service import ( LAUNCHD_LABEL, SYSTEMD_UNIT, + RuntimePolicy, render_launchd_plist, render_systemd_units, sweep_command, ) +def _strict_check() -> _service._RuntimeCheck: + return _service._RuntimeCheck(policy=RuntimePolicy.STRICT) + + +def _balanced_check() -> _service._RuntimeCheck: + return _service._RuntimeCheck(policy=RuntimePolicy.BALANCED) + + def test_sweep_command_uses_current_interpreter_in_isolated_mode(): import sys @@ -408,10 +417,10 @@ def test_runtime_path_and_dir_type_checks(tmp_path, monkeypatch): monkeypatch.setattr(os, "stat", _synthetic_lstat(os.stat)) with pytest.raises(_service.ServiceError, match="not a regular file"): - _service._validate_runtime_path(directory, "runtime") + _service._validate_runtime_path(directory, "runtime", _strict_check()) with pytest.raises(_service.ServiceError, match="not a directory"): - _service._validate_runtime_dir(file_path, "runtime dir") + _service._validate_runtime_dir(file_path, "runtime dir", _strict_check()) @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlink support required") @@ -539,8 +548,8 @@ def test_runtime_path_rejects_other_user_writable_component(tmp_path): runtime.write_text("placeholder", encoding="utf-8") runtime.chmod(0o666) - with pytest.raises(_service.ServiceError, match="group/world-writable"): - _service._validate_runtime_path(runtime, "test runtime") + with pytest.raises(_service.ServiceError, match="world-writable"): + _service._validate_runtime_path(runtime, "test runtime", _strict_check()) def test_runtime_component_rejects_sticky_shared_temp_dir(): @@ -548,11 +557,14 @@ def test_runtime_component_rejects_sticky_shared_temp_dir(): (stat.S_IFDIR | 0o1777, 1, 2, 1, 0, 0, 0, 0, 0, 0) ) - with pytest.raises(_service.ServiceError, match="group/world-writable"): + # A world-writable directory is rejected even under balanced — sticky or not, + # /tmp can never host a scheduled service runtime. + with pytest.raises(_service.ServiceError, match="world-writable"): _service._check_runtime_component( Path("/tmp"), sticky_tmp, "test runtime path component", + _balanced_check(), ) @@ -567,10 +579,11 @@ def test_runtime_error_recommends_safe_uv_managed_venv(): Path("/tmp"), writable, "test runtime path component", + _strict_check(), ) message = str(exc_info.value) - assert "group/world-writable" in message + assert "world-writable" in message assert "uv python install 3.12" in message assert "uv venv ~/.venvs/ephemdir-safe --python 3.12" in message assert "uv pip install --python ~/.venvs/ephemdir-safe/bin/python ephemdir" in message @@ -614,7 +627,7 @@ def test_runtime_path_rejects_foreign_owned_0755_component(tmp_path, monkeypatch monkeypatch.setattr(os, "lstat", _synthetic_lstat(os.lstat, foreign=(venv,))) with pytest.raises(_service.ServiceError, match="component .* owned by another user"): - _service._validate_runtime_path(runtime, "test runtime") + _service._validate_runtime_path(runtime, "test runtime", _strict_check()) @pytest.mark.skipif(not hasattr(os, "geteuid"), reason="POSIX ownership semantics required") @@ -639,7 +652,7 @@ def foreign_final_stat(path, *args, **kwargs): monkeypatch.setattr(os, "stat", foreign_final_stat) with pytest.raises(_service.ServiceError, match="test runtime is owned by another user"): - _service._validate_runtime_path(runtime, "test runtime") + _service._validate_runtime_path(runtime, "test runtime", _strict_check()) @pytest.mark.skipif(not hasattr(os, "geteuid"), reason="POSIX ownership semantics required") @@ -652,14 +665,16 @@ def test_check_runtime_component_rejects_foreign_owner(): from pathlib import Path with pytest.raises(_service.ServiceError, match="owned by another user"): - _service._check_runtime_component(Path("/synthetic"), foreign, "test component") + _service._check_runtime_component( + Path("/synthetic"), foreign, "test component", _strict_check() + ) def test_install_service_validates_persistent_runtime(monkeypatch): monkeypatch.setattr(_service.sys, "platform", "linux") monkeypatch.setattr(_service, "_reject_elevated_user_install", lambda: None) - def reject(): + def reject(check): raise _service.ServiceError("unsafe persistent runtime") monkeypatch.setattr(_service, "_validate_service_runtime", reject) @@ -678,7 +693,7 @@ def test_install_service_rejects_root_user(monkeypatch): def test_install_service_dispatches_linux_after_validation(monkeypatch): monkeypatch.setattr(_service.sys, "platform", "linux") monkeypatch.setattr(_service, "_reject_elevated_user_install", lambda: None) - monkeypatch.setattr(_service, "_validate_service_runtime", lambda: None) + monkeypatch.setattr(_service, "_validate_service_runtime", lambda check: None) monkeypatch.setattr(_service, "_verify_isolated_import", lambda: None) monkeypatch.setattr(_service, "_install_systemd", lambda interval: f"linux:{interval}") @@ -688,7 +703,7 @@ def test_install_service_dispatches_linux_after_validation(monkeypatch): def test_install_service_dispatches_darwin_after_validation(monkeypatch): monkeypatch.setattr(_service.sys, "platform", "darwin") monkeypatch.setattr(_service, "_reject_elevated_user_install", lambda: None) - monkeypatch.setattr(_service, "_validate_service_runtime", lambda: None) + monkeypatch.setattr(_service, "_validate_service_runtime", lambda check: None) monkeypatch.setattr(_service, "_verify_isolated_import", lambda: None) monkeypatch.setattr(_service, "_install_launchd", lambda interval: f"darwin:{interval}") @@ -918,9 +933,11 @@ def test_startup_file_iteration_and_tomli_validation(tmp_path, monkeypatch): "find_spec", lambda name: type("Spec", (), {"origin": str(tomli)})() if name == "tomli" else None, ) - monkeypatch.setattr(_service, "_validate_package_tree", lambda path: checked.append(path)) + monkeypatch.setattr( + _service, "_validate_package_tree", lambda path, check: checked.append(path) + ) - _service._validate_startup_environment() + _service._validate_startup_environment(_strict_check()) assert checked == [tomli.parent] @@ -930,20 +947,20 @@ def test_validate_service_runtime_checks_interpreter_package_and_startup(monkeyp monkeypatch.setattr( _service, "_validate_runtime_path", - lambda path, label: calls.append(f"path:{label}"), + lambda path, label, check: calls.append(f"path:{label}"), ) monkeypatch.setattr( _service, "_validate_package_tree", - lambda path: calls.append("package"), + lambda path, check: calls.append("package"), ) monkeypatch.setattr( _service, "_validate_startup_environment", - lambda: calls.append("startup"), + lambda check: calls.append("startup"), ) - _service._validate_service_runtime() + _service._validate_service_runtime(_strict_check()) assert calls == ["path:Python interpreter", "package", "startup"] @@ -969,12 +986,14 @@ def test_validate_package_tree_checks_every_module(tmp_path, monkeypatch): pkg = _make_fake_package(tmp_path) checked: list[str] = [] monkeypatch.setattr( - _service, "_validate_runtime_path", lambda path, label: checked.append(Path(path).name) + _service, + "_validate_runtime_path", + lambda path, label, check: checked.append(Path(path).name), ) # The per-directory check is exercised separately; here we test file walking. - monkeypatch.setattr(_service, "_validate_runtime_dir", lambda path, label: None) + monkeypatch.setattr(_service, "_validate_runtime_dir", lambda path, label, check: None) - _service._validate_package_tree(pkg) + _service._validate_package_tree(pkg, _strict_check()) for required in ("__main__.py", "cli.py", "core.py", "_registry.py"): assert required in checked @@ -988,14 +1007,14 @@ def test_validate_package_tree_rejects_writable_module(tmp_path, monkeypatch): # the install even when __init__.py and the directories are locked down. pkg = _make_fake_package(tmp_path) - def fake_validate(path, label): + def fake_validate(path, label, check): if Path(path).name == "__main__.py": raise _service.ServiceError(f"{path} is writable by other users") monkeypatch.setattr(_service, "_validate_runtime_path", fake_validate) - monkeypatch.setattr(_service, "_validate_runtime_dir", lambda path, label: None) + monkeypatch.setattr(_service, "_validate_runtime_dir", lambda path, label, check: None) with pytest.raises(_service.ServiceError, match="writable by other users"): - _service._validate_package_tree(pkg) + _service._validate_package_tree(pkg, _strict_check()) def test_validate_package_tree_rejects_symlinked_subdir(tmp_path, monkeypatch): @@ -1011,10 +1030,10 @@ def test_validate_package_tree_rejects_symlinked_subdir(tmp_path, monkeypatch): shutil.rmtree(pkg / "__pycache__") (pkg / "__pycache__").symlink_to(outside, target_is_directory=True) - monkeypatch.setattr(_service, "_validate_runtime_path", lambda path, label: None) - monkeypatch.setattr(_service, "_validate_runtime_dir", lambda path, label: None) + monkeypatch.setattr(_service, "_validate_runtime_path", lambda path, label, check: None) + monkeypatch.setattr(_service, "_validate_runtime_dir", lambda path, label, check: None) with pytest.raises(_service.ServiceError, match="symlinked subdirectory"): - _service._validate_package_tree(pkg) + _service._validate_package_tree(pkg, _strict_check()) @pytest.mark.skipif(not hasattr(os, "geteuid"), reason="POSIX ownership semantics required") @@ -1033,12 +1052,12 @@ def test_validate_package_tree_rejects_foreign_owned_empty_subdir(tmp_path, monk monkeypatch.setattr(os, "lstat", _synthetic_lstat(real_lstat, foreign=(cache,))) monkeypatch.setattr(os, "stat", _synthetic_lstat(real_stat, foreign=(cache,))) with pytest.raises(_service.ServiceError, match="owned by another user"): - _service._validate_package_tree(pkg) + _service._validate_package_tree(pkg, _strict_check()) # Positive control: the same tree with no foreign directory validates. monkeypatch.setattr(os, "lstat", _synthetic_lstat(real_lstat)) monkeypatch.setattr(os, "stat", _synthetic_lstat(real_stat)) - _service._validate_package_tree(pkg) # must not raise + _service._validate_package_tree(pkg, _strict_check()) # must not raise def test_validate_package_tree_fails_closed_on_unreadable_subdir(tmp_path, monkeypatch): @@ -1055,7 +1074,7 @@ def unreadable_walk(path, *, onerror=None, **kwargs): monkeypatch.setattr(_service.os, "walk", unreadable_walk) with pytest.raises(_service.ServiceError, match="cannot inspect package directory"): - _service._validate_package_tree(pkg) + _service._validate_package_tree(pkg, _strict_check()) def test_validate_package_tree_requires_some_module(tmp_path, monkeypatch): @@ -1063,10 +1082,10 @@ def test_validate_package_tree_requires_some_module(tmp_path, monkeypatch): # success without having verified any actual code. empty = tmp_path / "empty" empty.mkdir() - monkeypatch.setattr(_service, "_validate_runtime_path", lambda path, label: None) - monkeypatch.setattr(_service, "_validate_runtime_dir", lambda path, label: None) + monkeypatch.setattr(_service, "_validate_runtime_path", lambda path, label, check: None) + monkeypatch.setattr(_service, "_validate_runtime_dir", lambda path, label, check: None) with pytest.raises(_service.ServiceError, match="no ephemdir module files"): - _service._validate_package_tree(empty) + _service._validate_package_tree(empty, _strict_check()) def test_iter_startup_files_includes_site_pth(tmp_path, monkeypatch): @@ -1090,10 +1109,320 @@ def test_validate_startup_environment_rejects_writable_pth(tmp_path, monkeypatch (site_dir / "evil.pth").write_text("import os\n", encoding="utf-8") monkeypatch.setattr(_service.site, "getsitepackages", lambda: [str(site_dir)]) - def fake_validate(path, label): + def fake_validate(path, label, check): if Path(path).name == "evil.pth": raise _service.ServiceError(f"{path} is writable by other users") monkeypatch.setattr(_service, "_validate_runtime_path", fake_validate) with pytest.raises(_service.ServiceError, match="writable by other users"): - _service._validate_startup_environment() + _service._validate_startup_environment(_strict_check()) + + +# --- Runtime policy: strict vs balanced (0.6.0) ---------------------------- + + +def _own_uid() -> int: + return os.geteuid() if hasattr(os, "geteuid") else 0 + + +def _dir_stat(mode_bits: int, uid: int) -> os.stat_result: + return os.stat_result((stat.S_IFDIR | mode_bits, 1, 2, 1, uid, 0, 0, 0, 0, 0)) + + +def _file_stat(mode_bits: int, uid: int) -> os.stat_result: + return os.stat_result((stat.S_IFREG | mode_bits, 1, 2, 1, uid, 0, 0, 0, 0, 0)) + + +def _sanitize_ancestors_lstat(real, target): + """lstat/stat shim: keep the target component real, lock down its ancestors. + + The component under test reports its true mode and owner; every other path + reports a root-owned, non-other-writable directory so the result does not + depend on where pytest places ``tmp_path`` (often a world-writable /tmp). + """ + target = str(target) + + def fake(path, *args, **kwargs): + result = real(path, *args, **kwargs) + if str(path) == target: + return result + values = list(result) + values[0] = result.st_mode & ~0o022 # clear group/world write + values[4] = 0 # root-owned + return os.stat_result(values) + + return fake + + +def test_runtime_policy_default_is_balanced_on_macos(monkeypatch): + monkeypatch.delenv(_service._RUNTIME_POLICY_ENV, raising=False) + monkeypatch.setattr(_service.sys, "platform", "darwin") + assert _service._resolve_runtime_policy(None) is RuntimePolicy.BALANCED + + +def test_runtime_policy_default_is_strict_off_macos(monkeypatch): + monkeypatch.delenv(_service._RUNTIME_POLICY_ENV, raising=False) + monkeypatch.setattr(_service.sys, "platform", "linux") + assert _service._resolve_runtime_policy(None) is RuntimePolicy.STRICT + + +def test_runtime_policy_env_overrides_platform_default(monkeypatch): + monkeypatch.setattr(_service.sys, "platform", "darwin") + monkeypatch.setenv(_service._RUNTIME_POLICY_ENV, "strict") + assert _service._resolve_runtime_policy(None) is RuntimePolicy.STRICT + + +def test_runtime_policy_explicit_overrides_env(monkeypatch): + monkeypatch.setenv(_service._RUNTIME_POLICY_ENV, "strict") + assert _service._resolve_runtime_policy("balanced") is RuntimePolicy.BALANCED + + +def test_runtime_policy_rejects_invalid_value(monkeypatch): + monkeypatch.setenv(_service._RUNTIME_POLICY_ENV, "paranoid") + with pytest.raises(_service.ServiceError, match="invalid"): + _service._resolve_runtime_policy(None) + + +def _dir_stat_g(mode_bits: int, uid: int, gid: int) -> os.stat_result: + return os.stat_result((stat.S_IFDIR | mode_bits, 1, 2, 1, uid, gid, 0, 0, 0, 0)) + + +def _enable_homebrew_carveout(monkeypatch, *, gids=frozenset({80}), within=True): + """Make the balanced macOS Homebrew/usr-local carve-out apply deterministically.""" + monkeypatch.setattr(_service.sys, "platform", "darwin") + monkeypatch.setattr(_service, "_trusted_admin_gids", lambda: gids) + monkeypatch.setattr(_service, "_within_allowlisted_prefix", lambda path: within) + + +def test_within_allowlisted_prefix(tmp_path, monkeypatch): + monkeypatch.setattr(_service, "_GROUP_WRITABLE_PREFIX_ALLOWLIST", (str(tmp_path.resolve()),)) + inside = tmp_path / "Cellar" / "python@3.12" + inside.mkdir(parents=True) + assert _service._within_allowlisted_prefix(inside) + assert not _service._within_allowlisted_prefix(Path("/")) + + +def test_trusted_admin_gids_includes_macos_admin_fallback(): + assert 80 in _service._trusted_admin_gids() + + +def test_balanced_allows_homebrew_admin_group_writable_dir(monkeypatch): + # The carve-out: macOS, /opt/homebrew/Cellar (0775, group admin), owned by you. + _enable_homebrew_carveout(monkeypatch) + check = _balanced_check() + _service._check_runtime_component( + Path("/opt/homebrew/Cellar"), + _dir_stat_g(0o775, _own_uid(), 80), + "Python interpreter path component", + check, + ) + assert any("group-writable" in w for w in check.warnings) + assert any("--runtime-policy strict" in w for w in check.warnings) + + +def test_balanced_allows_root_owned_homebrew_group_writable_dir(monkeypatch): + _enable_homebrew_carveout(monkeypatch) + check = _balanced_check() + _service._check_runtime_component( + Path("/opt/homebrew/Cellar"), _dir_stat_g(0o775, 0, 80), "interpreter path component", check + ) + assert check.warnings + + +def test_balanced_rejects_group_writable_dir_with_nonadmin_group(monkeypatch): + # H-01 regression: a shared "project" group may contain a *different* + # unprivileged user, who could replace a directory on the import path and + # gain code execution as the installing user. A group-writable dir whose + # owning group is not a local admin group must be rejected even under + # balanced, regardless of being owned by you and under an allowlisted prefix. + monkeypatch.setattr(_service.sys, "platform", "darwin") + monkeypatch.setattr(_service, "_within_allowlisted_prefix", lambda path: True) + info = _dir_stat_g(0o775, _own_uid(), 5000) # gid 5000 is not an admin group + with pytest.raises(_service.ServiceError, match="group-writable"): + _service._check_runtime_component( + Path("/Users/alice/shared-venv"), info, "interpreter path component", _balanced_check() + ) + + +def test_balanced_rejects_group_writable_dir_outside_allowlist(monkeypatch): + # Admin-owned group but not under /opt/homebrew or /usr/local: not the + # Homebrew carve-out, so balanced must still reject it. + _enable_homebrew_carveout(monkeypatch, within=False) + info = _dir_stat_g(0o775, _own_uid(), 80) + with pytest.raises(_service.ServiceError, match="group-writable"): + _service._check_runtime_component( + Path("/home/user/venv"), info, "interpreter path component", _balanced_check() + ) + + +def test_balanced_carveout_is_macos_only(monkeypatch): + # The relaxation is macOS-specific; on Linux a group-writable dir is rejected + # even under an explicit --runtime-policy balanced. + monkeypatch.setattr(_service.sys, "platform", "linux") + monkeypatch.setattr(_service, "_within_allowlisted_prefix", lambda path: True) + info = _dir_stat_g(0o775, _own_uid(), 80) + with pytest.raises(_service.ServiceError, match="group-writable"): + _service._check_runtime_component( + Path("/opt/homebrew/Cellar"), info, "interpreter path component", _balanced_check() + ) + + +def test_strict_rejects_homebrew_group_writable_component(monkeypatch): + _enable_homebrew_carveout(monkeypatch) + with pytest.raises(_service.ServiceError, match="group-writable"): + _service._check_runtime_component( + Path("/opt/homebrew/Cellar"), + _dir_stat_g(0o775, _own_uid(), 80), + "interpreter path component", + _strict_check(), + ) + + +def test_strict_group_writable_error_mentions_balanced_for_homebrew(monkeypatch): + _enable_homebrew_carveout(monkeypatch) + with pytest.raises(_service.ServiceError, match="--runtime-policy balanced"): + _service._check_runtime_component( + Path("/opt/homebrew/Cellar"), + _dir_stat_g(0o775, _own_uid(), 80), + "interpreter path component", + _strict_check(), + ) + + +def test_strict_group_writable_error_omits_balanced_for_nonhomebrew(monkeypatch): + # Don't dangle a balanced hint for a dir balanced would also reject. + monkeypatch.setattr(_service.sys, "platform", "darwin") + monkeypatch.setattr(_service, "_within_allowlisted_prefix", lambda path: True) + with pytest.raises(_service.ServiceError) as exc_info: + _service._check_runtime_component( + Path("/Users/alice/shared"), + _dir_stat_g(0o775, _own_uid(), 5000), + "interpreter path component", + _strict_check(), + ) + assert "--runtime-policy balanced" not in str(exc_info.value) + + +def test_balanced_rejects_world_writable_component(): + # World-writable (sticky /tmp) is fatal even under balanced. + with pytest.raises(_service.ServiceError, match="world-writable"): + _service._check_runtime_component( + Path("/tmp"), _dir_stat(0o1777, _own_uid()), "interpreter path component", + _balanced_check(), + ) + + +@pytest.mark.skipif(not hasattr(os, "geteuid"), reason="POSIX ownership semantics required") +def test_balanced_rejects_foreign_owned_component(): + foreign = _dir_stat(0o755, os.geteuid() + 1) + with pytest.raises(_service.ServiceError, match="owned by another user"): + _service._check_runtime_component( + Path("/opt/foreign"), foreign, "interpreter path component", _balanced_check() + ) + + +def test_balanced_rejects_group_writable_executable_file(): + # A group-writable *file* is executable code (module/.pth), so it stays a + # hard failure even under balanced; only directory ancestors are relaxed. + info = _file_stat(0o664, _own_uid()) + with pytest.raises(_service.ServiceError, match="group-writable"): + _service._check_runtime_component( + Path("/opt/homebrew/.../ephemdir/core.py"), + info, + "ephemdir module path component", + _balanced_check(), + ) + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX permission semantics required") +def test_install_service_macos_balanced_allows_homebrew_group_writable_dir(tmp_path, monkeypatch): + # End-to-end on a real filesystem: a genuinely group-writable directory in + # the carve-out (allowlisted prefix + admin owning group) passes under + # balanced (with a warning) and fails under strict. + runtime_dir = tmp_path / "Cellar" + runtime_dir.mkdir() + runtime_dir.chmod(0o775) # real group-writable bit, owned by the test user + real_gid = runtime_dir.stat().st_gid + + monkeypatch.setattr(_service.sys, "platform", "darwin") + monkeypatch.setattr( + _service, "_GROUP_WRITABLE_PREFIX_ALLOWLIST", (str(tmp_path.resolve()),) + ) + monkeypatch.setattr(_service, "_trusted_admin_gids", lambda: frozenset({real_gid})) + monkeypatch.setattr(os, "lstat", _sanitize_ancestors_lstat(os.lstat, runtime_dir)) + monkeypatch.setattr(os, "stat", _sanitize_ancestors_lstat(os.stat, runtime_dir)) + + balanced = _balanced_check() + _service._validate_runtime_dir(runtime_dir, "ephemdir package directory", balanced) + assert balanced.warnings # allowed, but warned + + with pytest.raises(_service.ServiceError, match="group-writable"): + _service._validate_runtime_dir(runtime_dir, "ephemdir package directory", _strict_check()) + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX permission semantics required") +def test_install_service_balanced_rejects_group_writable_dir_foreign_group(tmp_path, monkeypatch): + # The same allowlisted, group-writable directory is rejected when its owning + # group is NOT a trusted admin group (it might contain another local user). + runtime_dir = tmp_path / "Cellar" + runtime_dir.mkdir() + runtime_dir.chmod(0o775) + + monkeypatch.setattr(_service.sys, "platform", "darwin") + monkeypatch.setattr( + _service, "_GROUP_WRITABLE_PREFIX_ALLOWLIST", (str(tmp_path.resolve()),) + ) + monkeypatch.setattr(_service, "_trusted_admin_gids", lambda: frozenset()) # none trusted + monkeypatch.setattr(os, "lstat", _sanitize_ancestors_lstat(os.lstat, runtime_dir)) + monkeypatch.setattr(os, "stat", _sanitize_ancestors_lstat(os.stat, runtime_dir)) + + with pytest.raises(_service.ServiceError, match="group-writable"): + _service._validate_runtime_dir( + runtime_dir, "ephemdir package directory", _balanced_check() + ) + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX permission semantics required") +def test_install_service_balanced_rejects_world_writable_tmp_runtime(tmp_path, monkeypatch): + runtime_dir = tmp_path / "shared" + runtime_dir.mkdir() + runtime_dir.chmod(0o1777) # sticky, world-writable like /tmp + + monkeypatch.setattr(os, "lstat", _sanitize_ancestors_lstat(os.lstat, runtime_dir)) + monkeypatch.setattr(os, "stat", _sanitize_ancestors_lstat(os.stat, runtime_dir)) + + with pytest.raises(_service.ServiceError, match="world-writable"): + _service._validate_runtime_dir( + runtime_dir, "ephemdir package directory", _balanced_check() + ) + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX permission semantics required") +def test_install_service_balanced_rejects_group_writable_module_file(tmp_path, monkeypatch): + module = tmp_path / "core.py" + module.write_text("", encoding="utf-8") + module.chmod(0o664) # group-writable regular file == executable code + + monkeypatch.setattr(os, "lstat", _sanitize_ancestors_lstat(os.lstat, module)) + monkeypatch.setattr(os, "stat", _sanitize_ancestors_lstat(os.stat, module)) + + with pytest.raises(_service.ServiceError, match="group-writable"): + _service._validate_runtime_path(module, "ephemdir module", _balanced_check()) + + +def test_install_service_surfaces_balanced_warning(monkeypatch, caplog): + monkeypatch.setattr(_service.sys, "platform", "darwin") + monkeypatch.setattr(_service, "_reject_elevated_user_install", lambda: None) + + def validate(check): + check.warn("service runtime uses group-writable path component /opt/homebrew/Cellar") + + monkeypatch.setattr(_service, "_validate_service_runtime", validate) + monkeypatch.setattr(_service, "_verify_isolated_import", lambda: None) + monkeypatch.setattr(_service, "_install_launchd", lambda interval: "installed launchd") + + with caplog.at_level("WARNING", logger="ephemdir"): + result = _service.install_service(interval=600, runtime_policy="balanced") + + assert result == "installed launchd" + assert any("group-writable" in record.message for record in caplog.records)