Skip to content
Open
Show file tree
Hide file tree
Changes from 6 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
60 changes: 56 additions & 4 deletions src/specify_cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3641,17 +3641,69 @@ def extension_add(
console.print("Only install extensions from sources you trust.\n")
console.print(f"Downloading from {from_url}...")

# Download ZIP to temp location
# Download ZIP to temp location.
# Validate and create each ancestor without following symlinks
# (consistent with shared_infra._ensure_safe_shared_directory):
# `mkdir(parents=True)` would silently traverse a symlinked ancestor,
# so each component is checked and created individually instead.
# Each existing component is also re-resolved and required to land
# under project_root, which catches non-symlink directory aliases
# (e.g. Windows junctions / mount points) that resolve outside.
import uuid as _uuid
download_dir = project_root / ".specify" / "extensions" / ".cache" / "downloads"
download_dir.mkdir(parents=True, exist_ok=True)
zip_path = download_dir / f"{extension}-url-download.zip"
project_root_resolved = project_root.resolve()
Comment thread
mnriem marked this conversation as resolved.
Outdated
_current = project_root
for _part in download_dir.relative_to(project_root).parts:
_current = _current / _part
if _current.is_symlink():
console.print("[red]Error:[/red] Refusing to use symlinked download cache directory")
raise typer.Exit(1)
if _current.exists():
try:
_current.resolve().relative_to(project_root_resolved)
except (OSError, ValueError):
console.print("[red]Error:[/red] Download cache directory escapes project root")
raise typer.Exit(1)
Comment thread
mnriem marked this conversation as resolved.
Outdated
continue
_current.mkdir()
if _current.is_symlink():
console.print("[red]Error:[/red] Refusing to use symlinked download cache directory")
raise typer.Exit(1)
Comment thread
mnriem marked this conversation as resolved.
Outdated

safe_name = Path(extension).name.replace("/", "_").replace("\\", "_") or "download"
safe_name = safe_name[:64] # cap length to avoid filesystem errors
zip_path = download_dir / f"{safe_name}-{_uuid.uuid4().hex[:8]}.zip"

# Guard: resolved path must stay inside download_dir (CWE-22)
try:
zip_path.resolve().relative_to(download_dir.resolve())
Comment thread
mnriem marked this conversation as resolved.
Outdated
except ValueError:
Comment thread
mnriem marked this conversation as resolved.
Outdated
console.print("[red]Error:[/red] Invalid extension name (path traversal detected)")
raise typer.Exit(1)

try:
from specify_cli.authentication.http import open_url as _open_url

with _open_url(from_url, timeout=60) as response:
zip_data = response.read()
zip_path.write_bytes(zip_data)
# Open with O_NOFOLLOW | O_EXCL to defeat any TOCTOU symlink
# swap that may have happened between the cache validation
# above and this write. O_EXCL also guarantees the unique
# UUID filename is freshly created, never reused. O_NOFOLLOW
# is a no-op on Windows where the constant is absent.
_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
_fd = os.open(zip_path, _flags, 0o600)
Comment thread
mnriem marked this conversation as resolved.
Outdated
Comment thread
mnriem marked this conversation as resolved.
Outdated
try:
with os.fdopen(_fd, "wb") as _zf:
_zf.write(zip_data)
except BaseException:
# os.fdopen took ownership only on success; on failure
# before the with-block, the fd may still be open.
try:
os.close(_fd)
except OSError:
pass
raise

# Install from downloaded ZIP
manifest = manager.install_from_zip(zip_path, speckit_version, priority=priority)
Expand Down
232 changes: 232 additions & 0 deletions tests/test_extension_add_path_traversal.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,232 @@
"""Tests for path traversal guard in `specify extension add --from` (GHSA-67q9-p54f-7cpr).

The extension argument is used to construct a temporary ZIP download path.
Without sanitisation, absolute paths or ``../`` segments in the argument can
escape the intended cache directory, causing arbitrary file writes and deletes.
"""

from pathlib import Path
Comment thread
mnriem marked this conversation as resolved.
from unittest.mock import MagicMock, patch

import pytest
from typer.testing import CliRunner

from specify_cli import app

runner = CliRunner()

TRAVERSAL_PAYLOADS = [
"../pwned",
"../../etc/passwd",
"subdir/../../escape",
Comment thread
mnriem marked this conversation as resolved.
"..\\pwned",
"..\\..\\etc\\passwd",
"/tmp/evil",
Comment thread
mnriem marked this conversation as resolved.
]


@pytest.fixture()
def project_dir(tmp_path, monkeypatch):
proj = tmp_path / "project"
proj.mkdir()
(proj / ".specify").mkdir()
monkeypatch.chdir(proj)
return proj


def _mock_open_url():
"""Return a mock open_url that yields fake ZIP bytes."""
mock_response = MagicMock()
mock_response.read.return_value = b"PK\x03\x04fake"
mock_response.__enter__ = MagicMock(return_value=mock_response)
mock_response.__exit__ = MagicMock(return_value=False)
return mock_response


class TestExtensionAddFromPathTraversal:
"""Path traversal payloads in the extension name must not escape the download cache."""

@pytest.mark.parametrize("bad_name", TRAVERSAL_PAYLOADS)
def test_traversal_payload_writes_inside_cache(self, project_dir, bad_name):
"""The zip_path passed to install_from_zip must resolve inside the cache dir."""
cache_dir = project_dir / ".specify" / "extensions" / ".cache" / "downloads"
with patch("specify_cli.authentication.http.open_url", return_value=_mock_open_url()), \
patch("specify_cli.extensions.ExtensionManager.install_from_zip", return_value=MagicMock(id="x", name="X", version="1.0.0")) as mock_install:
result = runner.invoke(
app,
["extension", "add", bad_name, "--from", "https://example.com/ext.zip"],
)
assert result.exit_code == 0, result.output
mock_install.assert_called_once()
zip_arg = mock_install.call_args[0][0] # positional arg: zip_path
zip_arg.resolve().relative_to(cache_dir.resolve()) # raises ValueError if outside

@pytest.mark.parametrize("bad_name", [p for p in TRAVERSAL_PAYLOADS if not p.startswith(("/", "\\"))])
def test_traversal_payload_cannot_delete_outside_cache(self, project_dir, bad_name):
"""The finally-block cleanup must not delete files outside the cache dir.

Absolute-path payloads are excluded here to avoid writing sentinels
outside the pytest sandbox (e.g. ``/tmp``); the write-side test above
already proves the production guard contains absolute payloads.
"""
# Place a sentinel at the path the pre-fix code would have constructed
cache_dir = project_dir / ".specify" / "extensions" / ".cache" / "downloads"
cache_dir.mkdir(parents=True, exist_ok=True)
pre_fix_path = cache_dir / f"{bad_name}-url-download.zip"
pre_fix_path.parent.mkdir(parents=True, exist_ok=True)
pre_fix_path.write_text("sentinel")

with patch("specify_cli.authentication.http.open_url", return_value=_mock_open_url()), \
patch("specify_cli.extensions.ExtensionManager.install_from_zip", return_value=MagicMock(id="x", name="X", version="1.0.0")):
runner.invoke(
app,
["extension", "add", bad_name, "--from", "https://example.com/ext.zip"],
)
Comment thread
mnriem marked this conversation as resolved.
Outdated
assert pre_fix_path.exists(), f"Sentinel deleted by cleanup: {pre_fix_path}"
assert pre_fix_path.read_text() == "sentinel"

def test_clean_name_is_unaffected(self, project_dir):
"""A normal extension name should not trigger the guard."""
with patch("specify_cli.authentication.http.open_url", return_value=_mock_open_url()), \
patch("specify_cli.extensions.ExtensionManager.install_from_zip") as mock_install:
mock_install.return_value = MagicMock(id="my-ext", name="My Ext", version="1.0.0")
result = runner.invoke(
app,
["extension", "add", "my-ext", "--from", "https://example.com/ext.zip"],
)

assert result.exit_code == 0, result.output
mock_install.assert_called_once()
# Verify the zip path is inside the cache
zip_arg = mock_install.call_args[0][0]
cache_dir = project_dir / ".specify" / "extensions" / ".cache" / "downloads"
zip_arg.resolve().relative_to(cache_dir.resolve())
assert "path traversal" not in (result.output or "").lower()


class TestExtensionAddFromSymlinkedCache:
"""A symlinked download-cache ancestor must be rejected before any write."""

@pytest.mark.parametrize(
"ancestor_parts",
[
(".specify",),
(".specify", "extensions"),
(".specify", "extensions", ".cache"),
(".specify", "extensions", ".cache", "downloads"),
],
)
def test_symlinked_ancestor_is_refused(self, tmp_path, monkeypatch, ancestor_parts):
"""Symlinking any cache ancestor outside the project must abort the command."""
project_dir = tmp_path / "project"
project_dir.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
monkeypatch.chdir(project_dir)

# Build the parent of the symlinked component using real directories.
parent = project_dir
for part in ancestor_parts[:-1]:
parent = parent / part
parent.mkdir()

# Replace the final ancestor component with a symlink pointing outside the project.
symlink_path = parent / ancestor_parts[-1]
symlink_path.symlink_to(outside, target_is_directory=True)
Comment thread
mnriem marked this conversation as resolved.

with patch("specify_cli.authentication.http.open_url", return_value=_mock_open_url()), \
patch("specify_cli.extensions.ExtensionManager.install_from_zip") as mock_install:
result = runner.invoke(
app,
["extension", "add", "my-ext", "--from", "https://example.com/ext.zip"],
)

assert result.exit_code != 0
assert "symlinked download cache" in (result.output or "").lower()
mock_install.assert_not_called()
# No download was written through the symlink
assert list(outside.iterdir()) == []


class TestExtensionAddFromAncestorEscape:
"""A non-symlink ancestor whose ``.resolve()`` lands outside the project must be refused.

Simulates the Windows junction / mount-point case using a manually-pre-resolved
directory swap: we cannot create real junctions on POSIX, but we can patch the
``.resolve()`` of an existing ancestor to land outside the project root. The
production guard must reject it before any download happens.
"""

def test_ancestor_resolves_outside_project_is_refused(self, tmp_path, monkeypatch):
project_dir = tmp_path / "project"
project_dir.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
monkeypatch.chdir(project_dir)

# Pre-create the cache ancestors as real directories so the per-component
# walk reaches the resolve()/relative_to() check on each existing one.
cache_root = project_dir / ".specify" / "extensions" / ".cache"
cache_root.mkdir(parents=True)

real_resolve = Path.resolve

def fake_resolve(self, *a, **kw):
# Pretend the .cache ancestor resolves outside the project.
if self == cache_root:
return real_resolve(outside, *a, **kw)
return real_resolve(self, *a, **kw)

with patch.object(Path, "resolve", fake_resolve), \
patch("specify_cli.authentication.http.open_url", return_value=_mock_open_url()), \
patch("specify_cli.extensions.ExtensionManager.install_from_zip") as mock_install:
result = runner.invoke(
app,
["extension", "add", "my-ext", "--from", "https://example.com/ext.zip"],
)

assert result.exit_code != 0
assert "escapes project root" in (result.output or "").lower()
mock_install.assert_not_called()
assert list(outside.iterdir()) == []


class TestExtensionAddFromTOCTOUWrite:
"""The download write must not follow a symlink swapped in after validation."""

def test_swapped_zip_path_symlink_is_refused(self, project_dir):
"""If zip_path is replaced by a symlink between validation and write, refuse."""
cache_dir = project_dir / ".specify" / "extensions" / ".cache" / "downloads"
outside = project_dir.parent / "outside"
outside.mkdir(exist_ok=True)
target = outside / "stolen.bin"

# Patch os.open to simulate the TOCTOU window: just before the real
# os.open runs, replace the target path with a symlink pointing outside.
# With O_NOFOLLOW the os.open call must raise instead of writing through.
Comment thread
mnriem marked this conversation as resolved.
Outdated
import os as _os
real_open = _os.open

def racing_open(path, flags, mode=0o777):
try:
p = Path(path)
if p.parent == cache_dir and not p.exists():
p.symlink_to(target)
except OSError:
pass
Comment thread
mnriem marked this conversation as resolved.
Outdated
return real_open(path, flags, mode)

with patch("specify_cli.authentication.http.open_url", return_value=_mock_open_url()), \
patch("specify_cli.extensions.ExtensionManager.install_from_zip") as mock_install, \
patch("os.open", side_effect=racing_open):
result = runner.invoke(
app,
["extension", "add", "my-ext", "--from", "https://example.com/ext.zip"],
)

# Either O_NOFOLLOW (POSIX) or O_EXCL (all platforms) must reject the
# swapped symlink. The command must fail and nothing must be written
# through the symlink to `outside`.
assert result.exit_code != 0
mock_install.assert_not_called()
assert not target.exists(), "write followed swapped symlink"
Loading