Skip to content
Merged
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
54 changes: 41 additions & 13 deletions backend/app/scanners/manifest.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,19 +82,29 @@ class ManifestScanner:
"""Discover and parse Python dependency manifests in a repository tree."""

def __init__(self, repo_path: str | Path) -> None:
self.repo_path = Path(repo_path)
self.repo_path = Path(repo_path).resolve()

def _resolve_in_repo(self, path: str | Path) -> Path | None:
"""Resolve an existing path and reject paths outside the repository root."""
try:
resolved = Path(path).resolve(strict=True)
resolved.relative_to(self.repo_path)
except (OSError, RuntimeError, ValueError):
return None
return resolved

def scan(self) -> ManifestScanResult:
result = ManifestScanResult()
if not self.repo_path.exists():
logger.warning("repo_path_missing", path=str(self.repo_path))
return result

visited_requirements: set[Path] = set()
for path in self._discover_manifests():
rel = str(path.relative_to(self.repo_path))
result.files_found.append(rel)
try:
deps = self._parse_file(path)
deps = self._parse_file(path, visited_requirements)
for d in deps:
d.manifest_source = rel
result.dependencies.extend(deps)
Expand Down Expand Up @@ -125,16 +135,21 @@ def _discover_manifests(self) -> list[Path]:
if any(part in skip_dirs for part in path.parts):
continue
name = path.name
if name in MANIFEST_NAMES:
found.append(path)
elif name.startswith("requirements") and name.endswith(".txt"):
is_manifest = name in MANIFEST_NAMES or (
name.startswith("requirements") and name.endswith(".txt")
)
if is_manifest and self._resolve_in_repo(path) is not None:
found.append(path)
return sorted(found)

def _parse_file(self, path: Path) -> list[ParsedDependency]:
def _parse_file(
self,
path: Path,
visited_requirements: set[Path] | None = None,
) -> list[ParsedDependency]:
name = path.name
if name.endswith(".txt") and "requirements" in name.lower():
return self._parse_requirements_txt(path)
return self._parse_requirements_txt(path, visited_requirements)
if name == "pyproject.toml":
return self._parse_pyproject(path)
if name == "poetry.lock":
Expand All @@ -151,18 +166,31 @@ def _parse_file(self, path: Path) -> list[ParsedDependency]:
return self._parse_setup_py(path)
return []

def _parse_requirements_txt(self, path: Path) -> list[ParsedDependency]:
def _parse_requirements_txt(
self,
path: Path,
visited: set[Path] | None = None,
) -> list[ParsedDependency]:
deps: list[ParsedDependency] = []
is_dev = any(x in path.name.lower() for x in ("dev", "test", "lint"))
text = read_text_auto(path)
visited = visited if visited is not None else set()
resolved_path = self._resolve_in_repo(path)
if resolved_path is None or resolved_path in visited:
return []
visited.add(resolved_path)

is_dev = any(x in resolved_path.name.lower() for x in ("dev", "test", "lint"))
text = read_text_auto(resolved_path)
for line in text.splitlines():
# Handle nested -r includes
stripped = line.strip()
if stripped.startswith("-r ") or stripped.startswith("--requirement "):
include = stripped.split(maxsplit=1)[1]
include_path = (path.parent / include).resolve()
if include_path.exists() and self.repo_path in include_path.parents or include_path.parent == self.repo_path:
deps.extend(self._parse_requirements_txt(include_path))
include_path = Path(include)
if not include_path.is_absolute():
include_path = resolved_path.parent / include_path
resolved_include = self._resolve_in_repo(include_path)
if resolved_include is not None and resolved_include.is_file():
deps.extend(self._parse_requirements_txt(resolved_include, visited))
continue
parsed = parse_requirement_line(line)
if not parsed:
Expand Down
52 changes: 52 additions & 0 deletions backend/tests/scanners/test_manifest.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,3 +114,55 @@ def test_malformed_requirements_are_skipped(tmp_path: Path):
names = {d.normalized_name for d in result.merged_by_name()}
assert "requests" in names
assert all("!!!" not in (d.package_name or "") for d in result.dependencies)


def test_requirements_nested_repeated_and_circular_includes_are_safe(tmp_path: Path):
(tmp_path / "requirements.txt").write_text(
"-r nested/base.txt\n-r nested/base.txt\nrequests==2.32.0\n",
encoding="utf-8",
)
(tmp_path / "nested").mkdir()
(tmp_path / "nested/base.txt").write_text(
"-r ../requirements.txt\npytest==8.3.0\n", encoding="utf-8"
)

result = ManifestScanner(tmp_path).scan()
names = [dep.normalized_name for dep in result.dependencies]

assert names.count("requests") == 1
assert names.count("pytest") == 1


def test_requirements_include_cannot_escape_repository(tmp_path: Path):
outside = tmp_path.parent / f"{tmp_path.name}-outside-requirements.txt"
outside.write_text("should-not-be-installed==1.0.0\n", encoding="utf-8")
try:
(tmp_path / "requirements.txt").write_text(
f"-r ../{outside.name}\nrequests==2.32.0\n", encoding="utf-8"
)

result = ManifestScanner(tmp_path).scan()
names = {dep.normalized_name for dep in result.dependencies}

assert "requests" in names
assert "should-not-be-installed" not in names
finally:
outside.unlink(missing_ok=True)


def test_requirements_include_rejects_symlink_escape(tmp_path: Path):
outside = tmp_path.parent / f"{tmp_path.name}-outside-link.txt"
outside.write_text("should-not-be-installed==1.0.0\n", encoding="utf-8")
try:
(tmp_path / "requirements.txt").write_text(
"-r linked-requirements.txt\nrequests==2.32.0\n", encoding="utf-8"
)
(tmp_path / "linked-requirements.txt").symlink_to(outside)

result = ManifestScanner(tmp_path).scan()
names = {dep.normalized_name for dep in result.dependencies}

assert "requests" in names
assert "should-not-be-installed" not in names
finally:
outside.unlink(missing_ok=True)
57 changes: 57 additions & 0 deletions backend/tests/scanners/test_manifest_include_safety.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
"""Requirements include safety regression tests."""

from pathlib import Path

from app.scanners.manifest import ManifestScanner


def test_requirements_include_cycle_is_processed_once(tmp_path: Path):
(tmp_path / "requirements.txt").write_text(
"-r requirements-dev.txt\nrequests==2.32.0\n", encoding="utf-8"
)
(tmp_path / "requirements-dev.txt").write_text(
"--requirement requirements.txt\npytest==8.3.0\n", encoding="utf-8"
)

result = ManifestScanner(tmp_path).scan()
versions = {dep.normalized_name: dep.version for dep in result.dependencies}

assert versions["requests"] == "2.32.0"
assert versions["pytest"] == "8.3.0"
assert len([dep for dep in result.dependencies if dep.normalized_name == "requests"]) == 1
assert len([dep for dep in result.dependencies if dep.normalized_name == "pytest"]) == 1


def test_requirements_include_cannot_escape_repository(tmp_path: Path):
outside = tmp_path.parent / f"{tmp_path.name}-outside-requirements.txt"
outside.write_text("should-not-be-installed==1.0.0\n", encoding="utf-8")
try:
(tmp_path / "requirements.txt").write_text(
f"-r ../{outside.name}\nrequests==2.32.0\n", encoding="utf-8"
)

result = ManifestScanner(tmp_path).scan()
names = {dep.normalized_name for dep in result.dependencies}

assert "requests" in names
assert "should-not-be-installed" not in names
finally:
outside.unlink(missing_ok=True)


def test_requirements_include_rejects_symlink_escape(tmp_path: Path):
outside = tmp_path.parent / f"{tmp_path.name}-outside-link.txt"
outside.write_text("should-not-be-installed==1.0.0\n", encoding="utf-8")
try:
(tmp_path / "requirements.txt").write_text(
"-r linked-requirements.txt\nrequests==2.32.0\n", encoding="utf-8"
)
(tmp_path / "linked-requirements.txt").symlink_to(outside)

result = ManifestScanner(tmp_path).scan()
names = {dep.normalized_name for dep in result.dependencies}

assert "requests" in names
assert "should-not-be-installed" not in names
finally:
outside.unlink(missing_ok=True)
Loading