Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
138 changes: 138 additions & 0 deletions design/file-helper.md
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.
259 changes: 259 additions & 0 deletions src/specify_cli/file_helper.py
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)
Comment thread
Copilot marked this conversation as resolved.
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()
Loading
Loading