Repository navigation
Add scoped FileHelper foundation for filesystem policy #4908
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
mnriem
wants to merge
5
commits into
github:main
Choose a base branch
from
mnriem:mnriem-filehelper-foundation
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
+1,346
−16
Open
Changes from all commits
Commits
Show all changes
5 commits
Select commit
Hold shift + click to select a range
12a51c4
feat: add scoped FileHelper foundation
mnriem 369fa47
fix: normalize Windows extended FileHelper targets
mnriem a4c8578
fix: scope parent traversal checks below trusted roots
mnriem ae29ace
fix: handle mixed-case Windows UNC namespaces
mnriem 2ad33b7
fix: reject existing leaves before exclusive file creation
mnriem File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,138 @@ | ||
| # Scoped FileHelper foundation | ||
|
|
||
| `specify_cli.file_helper.FileHelper` owns filesystem operations under one | ||
| explicitly established root. This is the first migration stage: existing CLI | ||
| commands retain their current behavior and do not yet use this helper. There is | ||
| no global CLI flag or process-global policy. | ||
|
|
||
| The user-directed persistence requirement belongs to the dependent | ||
| project-policy wiring stage: save the setting in `.specify/init-options.json`, | ||
| then load and validate it for future invocations and pass the resulting policy | ||
| explicitly to helper instances. Only `specify init` exposes and persists the | ||
| setting. Every other CLI command reads the current project configuration on | ||
| each invocation, so manual edits take effect without a cached policy. There | ||
| is no root/global flag and no per-command override flag. The low-level helper | ||
| does not read configuration. Strict persisted-policy loading and configuration- | ||
| read bootstrap semantics must be coordinated before migration. | ||
|
|
||
| ```python | ||
| from specify_cli.file_helper import FileHelper | ||
|
|
||
| files = FileHelper(project_root) # deny symlinks by default | ||
| files.mkdir("generated", exist_ok=True) | ||
| files.create_text("generated/config.txt", "initial\n") | ||
| files.write_text("generated/config.txt", "updated\n") | ||
| content = files.read_text("generated/config.txt") | ||
| files.delete("generated", recursive=True) | ||
| ``` | ||
|
|
||
| ## Boundary and policy | ||
|
|
||
| Each helper has a frozen root and `allow_symlinks` setting. Relative operation | ||
| paths are root-relative. Absolute paths must be beneath the supplied root or | ||
| its canonical equivalent. User operation paths containing `..` are rejected | ||
| in their root-relative suffix rather than normalized before inspection. A | ||
| caller-established root spelling may itself contain `..`; absolute operation | ||
| paths derived from that exact supplied root prefix are supported. Only that | ||
| trusted prefix is removed before checking operation components, without | ||
| lexically collapsing the root or ignoring traversal in the remaining suffix. | ||
| Link targets may contain `..`, but | ||
| are walked component by component and may not leave the root, even temporarily. | ||
| Windows rooted-but-not-absolute paths (`/outside/file`, `\outside\file`) and | ||
| drive-relative paths (`C:outside/file`, `C:`) are rejected for both operation | ||
| paths and link targets. They are not root-relative: joining them can reset the | ||
| root or drive. Fully absolute paths still require scoped root matching. | ||
| For Windows containment comparisons, extended-length drive and UNC spellings | ||
| (`\\?\D:\project\file`, `\\?\UNC\server\share\project\file`) are compared with | ||
| their ordinary equivalents on both the target and root sides. This does not | ||
| resolve links or collapse `..`: accessed hierarchy evidence is retained for | ||
| the component walk. The UNC namespace marker is matched case-insensitively, | ||
| while the remaining path's original spelling is preserved. External targets, | ||
| ambiguous rooted-relative paths, and | ||
| unsupported device namespaces are still rejected. POSIX filenames are not | ||
| reinterpreted as Windows path spellings. | ||
|
|
||
| The caller must establish a trusted, existing directory root. Its alias is | ||
| resolved once, including OS and worktree aliases; the root itself and OS | ||
| ancestors above it are not subjected to the traversal policy. Thus explicitly | ||
| establishing a linked directory as a root trusts that alias even in deny mode. | ||
| Links *below* that boundary are checked in both the original accessed path and | ||
| any followed target hierarchy. An unrelated project symlink does not prevent | ||
| an operation. A helper cannot cross to another root; callers establish separate | ||
| helpers for independent project/source/managed boundaries. | ||
|
|
||
| | Operation | Default deny | Explicit allow | | ||
| |---|---|---| | ||
| | Read file | Reject accessed parent or leaf links | Follow only existing contained regular-file targets | | ||
| | Create directory | Reject accessed links before creating parents | Follow contained directory links; reject dangling targets | | ||
| | Exclusively create file | Reject accessed links | Follow contained parents; reject an existing leaf without opening its target | | ||
| | Atomic write/upsert | Reject accessed links | Replace the resolved regular-file target and preserve every link | | ||
| | Delete leaf link | Reject without mutation | Unlink the link, including dangling, cyclic, or external-target links | | ||
| | Delete through linked parent | Reject without mutation | Delete the contained target entry, leaving the parent link intact | | ||
| | Recursive delete | Preflight all descendants and reject any link before mutation | Preflight all descendants; unlink links without traversing their targets | | ||
| | Create symlink | Reject | Require an existing contained file/directory target | | ||
|
|
||
| ## API and failures | ||
|
|
||
| - `mkdir(path, parents=False, exist_ok=False)` creates directories. | ||
| - `read_bytes(path)` / `read_text(path, encoding="utf-8")` read regular files. | ||
| - `create_bytes(path, content, mode=0o644)` / `create_text(...)` exclusively | ||
| create files with existing parents. A no-follow `lstat` preflight rejects | ||
| existing leaves, including dangling links, before opening the destination. | ||
| This avoids target creation on backends that follow dangling links despite | ||
| exclusive-open flags. `O_EXCL` and `O_NOFOLLOW` (where available) remain in use; | ||
| the static preflight does not eliminate concurrent path-change races. | ||
| - `write_bytes(path, content, mode=0o644)` / `write_text(...)` atomically | ||
| create or update files with existing parents. Text variants also accept | ||
| `encoding="utf-8"`. Replacement files use the supplied mode; metadata and | ||
| hard-link identity are not preserved. Atomic means same-directory | ||
| `os.replace`, not crash durability. | ||
| - `delete(path, recursive=False)` unlinks files/links or removes directories. | ||
| Nonrecursive directory deletion requires an empty directory. The established | ||
| root cannot be deleted. | ||
| - `symlink(target, path)` creates a link exclusively. Relative targets are | ||
| interpreted relative to the link parent, not the helper root. | ||
|
|
||
| `SymlinkDeniedError`, `PathEscapeError`, and `UnsupportedPathError` derive from | ||
| `FileHelperError` and describe policy failures. Missing paths (including | ||
| dangling read/update targets), permissions, existing exclusive-create | ||
| destinations, non-directory parents, and OS I/O failures retain native Python | ||
| exceptions. Nothing prints, warns-and-skips, or exits the CLI. Callers own | ||
| translation into their normal command error envelopes. | ||
|
|
||
| Only regular files, directories, and explicitly allowed symlinks are supported. | ||
| Below the established root, Windows directory junctions and other non-symlink | ||
| reparse entries are unsupported and rejected with `UnsupportedPathError` in | ||
| **both** policy modes. This includes contained junctions, leaf deletion, and | ||
| recursive deletion: preflight rejects them without traversing or removing | ||
| them. `allow_symlinks=True` does not authorize junction traversal or deletion. | ||
| The trusted root alias boundary may itself resolve through a junction just as | ||
| it may through a symlink; this does not authorize junctions below that root. | ||
| Read/update cycles and excessive link chains fail explicitly. A recursive | ||
| delete includes the entire selected tree; there is no exclusion filter in this | ||
| foundation. Future operations with exclusions must inspect only included | ||
| entries, not reject a whole project for excluded/unrelated links. | ||
|
|
||
| ## Safety limits and migration | ||
|
|
||
| Preflight detects static policy failures before recursive deletion or parent | ||
| creation. Atomic writes recheck the original path before replacement, and | ||
| deletion rechecks each accessed entry. Portable path checks are **not** a | ||
| race-free sandbox: another process can exchange parents or entries between | ||
| checks and syscalls, and an I/O failure during mutation can leave partial | ||
| results. Protect against concurrent untrusted mutation separately; descriptor- | ||
| relative traversal or platform-specific mechanisms are future work, not a | ||
| claim of this API. The helper does not constrain subprocesses or third-party | ||
| code. Exclusive creation uses `O_EXCL` and `O_NOFOLLOW` where available. | ||
|
|
||
| Existing safe-write mechanisms informed this helper, but existing shared | ||
| infrastructure warning/skip behavior and development-mode link creation are | ||
| unchanged. Migrate one owned operation boundary at a time, carrying original | ||
| paths and an explicit root/policy into the helper before any early resolution. | ||
| CLI option wiring, linked command directories, and development-mode | ||
| registration remain dependent stages. | ||
|
|
||
| Windows path classification and directory-mode reparse rejection have portable | ||
| regression tests. Native Windows path and junction tests also exist and are | ||
| skipped on other hosts; running the portable cases on macOS does not validate | ||
| Windows filesystem syscalls or junction behavior. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,259 @@ | ||
| """Scoped filesystem operations; see design/file-helper.md for the contract.""" | ||
|
|
||
| from dataclasses import dataclass, field | ||
| import errno | ||
| import os | ||
| from pathlib import Path, PurePath, PureWindowsPath | ||
| import stat | ||
| import tempfile | ||
| from typing import TypeVar | ||
|
|
||
|
|
||
| _PathType = TypeVar("_PathType", bound=PurePath) | ||
|
|
||
|
|
||
| def _strip_extended_length_prefix(path: _PathType) -> _PathType: | ||
| """Normalize Windows extended drive/UNC spellings for comparison only.""" | ||
| raw = str(path) | ||
| if raw.lower().startswith("\\\\?\\unc\\"): | ||
| return type(path)("\\\\" + raw[len("\\\\?\\UNC\\"):]) | ||
| if raw.startswith("\\\\?\\"): | ||
| return type(path)(raw[len("\\\\?\\"):]) | ||
| return path | ||
|
|
||
|
|
||
| class FileHelperError(ValueError): | ||
| """An operation violates the scoped filesystem contract.""" | ||
|
|
||
|
|
||
| class SymlinkDeniedError(FileHelperError): | ||
| """An accessed symlink requires explicit traversal permission.""" | ||
|
|
||
|
|
||
| class PathEscapeError(FileHelperError): | ||
| """An accessed path is outside the established root.""" | ||
|
|
||
|
|
||
| class UnsupportedPathError(FileHelperError): | ||
| """An accessed object cannot be used by the requested operation.""" | ||
|
|
||
|
|
||
| @dataclass(frozen=True) | ||
| class FileHelper: | ||
| """Own operations beneath one trusted root with an immutable link policy. | ||
|
|
||
| Relative operation paths are relative to ``root``, not the process cwd. | ||
| The root itself is a trusted alias boundary, resolved once on construction. | ||
| """ | ||
|
|
||
| root: Path | ||
| allow_symlinks: bool = False | ||
| _canonical_root: Path = field(init=False, repr=False) | ||
|
|
||
| def __post_init__(self) -> None: | ||
| root = Path(self.root).absolute() | ||
| canonical = root.resolve(strict=True) | ||
| if not canonical.is_dir(): | ||
| raise NotADirectoryError(root) | ||
| object.__setattr__(self, "root", root) | ||
| object.__setattr__(self, "_canonical_root", canonical) | ||
|
|
||
| def _parts(self, path: PurePath) -> tuple[str, ...]: | ||
| original = path | ||
| if isinstance(path, PureWindowsPath): | ||
| path = _strip_extended_length_prefix(path) | ||
| if not path.is_absolute(): | ||
| if original.anchor or path.anchor: | ||
| raise PathEscapeError(f"Rooted-relative, drive-relative, or device path is not permitted: {original}") | ||
| return path.parts | ||
| for root in (self.root, self._canonical_root): | ||
| if isinstance(root, PureWindowsPath): | ||
| root = _strip_extended_length_prefix(root) | ||
| try: | ||
| return path.relative_to(root).parts | ||
| except ValueError: | ||
| continue | ||
| raise PathEscapeError(f"Path is outside root {self.root}: {original}") | ||
|
|
||
| @staticmethod | ||
| def _entry_mode(path: Path) -> int: | ||
| entry = path.lstat() | ||
| if ( | ||
| getattr(entry, "st_file_attributes", 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT | ||
| and not stat.S_ISLNK(entry.st_mode) | ||
| ): | ||
| raise UnsupportedPathError( | ||
| f"Unsupported Windows reparse point (including directory junctions): {path}" | ||
| ) | ||
| return entry.st_mode | ||
|
|
||
| def _walk( | ||
| self, | ||
| parts: tuple[str, ...], | ||
| *, | ||
| allow_missing: bool = False, | ||
| follow_leaf: bool = True, | ||
| links: tuple[Path, ...] = (), | ||
| ) -> Path: | ||
| current = self._canonical_root | ||
| for index, part in enumerate(parts): | ||
| if not current.is_dir(): | ||
| if not (allow_missing and not current.exists()): | ||
| raise NotADirectoryError(current) | ||
| if part == "..": | ||
| if current == self._canonical_root: | ||
| raise PathEscapeError(f"Symlink target escapes root {self.root}") | ||
| current = current.parent | ||
| continue | ||
| candidate = current / part | ||
| try: | ||
| mode = self._entry_mode(candidate) | ||
| except FileNotFoundError: | ||
| if not allow_missing: | ||
| raise | ||
| current = candidate | ||
| continue | ||
| if stat.S_ISLNK(mode): | ||
| if not self.allow_symlinks: | ||
| raise SymlinkDeniedError( | ||
| f"Refusing symlink {candidate}; explicitly allow symlinks " | ||
| "on this FileHelper to use supported contained targets" | ||
| ) | ||
| if not follow_leaf and index == len(parts) - 1: | ||
| return candidate | ||
| if candidate in links or len(links) >= 40: | ||
| raise UnsupportedPathError(f"Symlink cycle or excessive chain: {candidate}") | ||
| target = Path(os.readlink(candidate)) | ||
| target_parts = self._parts(target) | ||
| if not target.is_absolute(): | ||
| target_parts = current.relative_to(self._canonical_root).parts + target_parts | ||
| # Resolve the link target strictly, even when creating a new child. | ||
| current = self._walk(target_parts, links=links + (candidate,)) | ||
| elif stat.S_ISDIR(mode) or stat.S_ISREG(mode): | ||
| current = candidate | ||
| else: | ||
| raise UnsupportedPathError(f"Unsupported filesystem object: {candidate}") | ||
| return current | ||
|
|
||
| def _path( | ||
| self, path: Path | str, *, allow_missing: bool = False, follow_leaf: bool = True | ||
| ) -> Path: | ||
| original = Path(path) | ||
| parts = self._parts(original) | ||
| if ".." in parts: | ||
| raise PathEscapeError(f"Parent traversal is not an operation path: {original}") | ||
| return self._walk( | ||
| parts, allow_missing=allow_missing, follow_leaf=follow_leaf | ||
| ) | ||
|
|
||
| @staticmethod | ||
| def _require_file(path: Path) -> None: | ||
| if not stat.S_ISREG(path.stat().st_mode): | ||
| raise UnsupportedPathError(f"Expected a regular file: {path}") | ||
|
|
||
| def mkdir( | ||
| self, path: Path | str, *, parents: bool = False, exist_ok: bool = False | ||
| ) -> None: | ||
| """Create a directory, preflighting the entire requested hierarchy.""" | ||
| destination = self._path(path, allow_missing=True) | ||
| destination.mkdir(parents=parents, exist_ok=exist_ok) | ||
|
|
||
| def read_bytes(self, path: Path | str) -> bytes: | ||
| destination = self._path(path) | ||
| self._require_file(destination) | ||
| return destination.read_bytes() | ||
|
|
||
| def read_text(self, path: Path | str, *, encoding: str = "utf-8") -> str: | ||
| return self.read_bytes(path).decode(encoding) | ||
|
|
||
| def create_bytes(self, path: Path | str, content: bytes, *, mode: int = 0o644) -> None: | ||
| """Exclusively create a file; never overwrite an existing entry.""" | ||
| destination = self._path(path, allow_missing=True, follow_leaf=False) | ||
| try: | ||
| self._entry_mode(destination) | ||
| except FileNotFoundError: | ||
| pass | ||
| else: | ||
| raise FileExistsError(errno.EEXIST, os.strerror(errno.EEXIST), str(destination)) | ||
| flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | ||
| flags |= getattr(os, "O_NOFOLLOW", 0) | ||
| fd = os.open(destination, flags, mode) | ||
| with os.fdopen(fd, "wb") as stream: | ||
| stream.write(content) | ||
|
|
||
| def create_text( | ||
| self, path: Path | str, content: str, *, encoding: str = "utf-8", mode: int = 0o644 | ||
| ) -> None: | ||
| self.create_bytes(path, content.encode(encoding), mode=mode) | ||
|
|
||
| def write_bytes(self, path: Path | str, content: bytes, *, mode: int = 0o644) -> None: | ||
| """Atomically create/replace a file, preserving any allowed leaf link.""" | ||
| destination = self._path(path, allow_missing=True) | ||
| if destination.exists(): | ||
| self._require_file(destination) | ||
| fd, temporary = tempfile.mkstemp(prefix=f".{destination.name}.", dir=destination.parent) | ||
| temporary_path = Path(temporary) | ||
| try: | ||
| with os.fdopen(fd, "wb") as stream: | ||
| stream.write(content) | ||
| temporary_path.chmod(mode) | ||
| if self._path(path, allow_missing=True) != destination: | ||
| raise FileHelperError(f"Destination changed during write: {path}") | ||
| os.replace(temporary_path, destination) | ||
| finally: | ||
| temporary_path.unlink(missing_ok=True) | ||
|
|
||
| def write_text( | ||
| self, path: Path | str, content: str, *, encoding: str = "utf-8", mode: int = 0o644 | ||
| ) -> None: | ||
| self.write_bytes(path, content.encode(encoding), mode=mode) | ||
|
|
||
| def symlink(self, target: Path | str, path: Path | str) -> None: | ||
| """Create a link to an existing contained regular file or directory. | ||
|
|
||
| Relative targets use the link's parent, matching ``Path.symlink_to``. | ||
| """ | ||
| if not self.allow_symlinks: | ||
| raise SymlinkDeniedError("Symlink creation requires explicit allow_symlinks=True") | ||
| destination = self._path(path, allow_missing=True, follow_leaf=False) | ||
| target = Path(target) | ||
| target_parts = self._parts(target) | ||
| if not target.is_absolute(): | ||
| target_parts = destination.parent.relative_to(self._canonical_root).parts + target_parts | ||
| resolved = self._walk(target_parts) | ||
| destination.symlink_to(target, target_is_directory=resolved.is_dir()) | ||
|
|
||
| def _deletion_plan(self, path: Path, *, recursive: bool) -> list[tuple[Path, bool]]: | ||
| mode = self._entry_mode(path) | ||
| if stat.S_ISLNK(mode): | ||
| if not self.allow_symlinks: | ||
| raise SymlinkDeniedError(f"Refusing to delete symlink: {path}") | ||
| return [(path, False)] | ||
| if stat.S_ISREG(mode): | ||
| return [(path, False)] | ||
| if not stat.S_ISDIR(mode): | ||
| raise UnsupportedPathError(f"Unsupported filesystem object: {path}") | ||
| plan: list[tuple[Path, bool]] = [] | ||
| if recursive: | ||
| with os.scandir(path) as entries: | ||
| children = sorted((Path(entry.path) for entry in entries)) | ||
| for child in children: | ||
| plan.extend(self._deletion_plan(child, recursive=True)) | ||
| plan.append((path, True)) | ||
| return plan | ||
|
|
||
| def delete(self, path: Path | str, *, recursive: bool = False) -> None: | ||
| """Delete an entry; recursive deletion never follows descendant links.""" | ||
| destination = self._path(path, follow_leaf=False) | ||
| if destination == self._canonical_root: | ||
| raise FileHelperError(f"Refusing to delete the established root: {self.root}") | ||
| plan = self._deletion_plan(destination, recursive=recursive) | ||
| for entry, directory in plan: | ||
| # Recheck accessed parents/leaf before each mutation, not just preflight. | ||
| checked = self._path(entry, follow_leaf=False) | ||
| if checked != entry: | ||
| raise FileHelperError(f"Destination changed during deletion: {entry}") | ||
| if directory: | ||
| entry.rmdir() | ||
| else: | ||
| entry.unlink() | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.