diff --git a/strix/fix/locations.py b/strix/fix/locations.py index d5f56c8cc..b401a6b49 100644 --- a/strix/fix/locations.py +++ b/strix/fix/locations.py @@ -45,14 +45,26 @@ def _find_blocks(content: str, block: str) -> tuple[int, ...]: ) +def _read_anchor_source(root: Path, file: str) -> str | None: + """Read a location's source, refusing symlinks that escape ``root`` and + files that cannot be decoded as UTF-8.""" + resolved_root = root.resolve() + file_path = (root / file).resolve() + if not file_path.is_relative_to(resolved_root) or not file_path.is_file(): + return None + try: + return file_path.read_text(encoding="utf-8") + except (OSError, UnicodeError): + return None + + def anchor_location( root: Path, location: CandidateLocation | FixEdit, ) -> AnchorResult: - file_path = root / location.file - if not file_path.is_file(): + content = _read_anchor_source(root, location.file) + if content is None: return AnchorResult(AnchorStatus.MISSING, location) - content = file_path.read_text(encoding="utf-8") block = location.before if isinstance(location, FixEdit) else location.snippet if not block: return AnchorResult(AnchorStatus.UNIQUE, location, (location.start_line,)) diff --git a/strix/fix/prepare.py b/strix/fix/prepare.py index 4fdf36e72..cdd6d51fc 100644 --- a/strix/fix/prepare.py +++ b/strix/fix/prepare.py @@ -3,11 +3,12 @@ from __future__ import annotations import asyncio +import functools import hashlib import os import subprocess import time -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Iterable from dataclasses import dataclass from pathlib import Path from typing import Literal @@ -62,7 +63,57 @@ IndependentVerifier = Callable[ SourceVerifier = Callable[[PreparationContext], Awaitable[bool]] -async def run_command(workspace: Path, command: CommandSpec) -> CheckResult: +_COMMAND_ENV_ALLOWLIST = frozenset( + { + "HOME", + "LANG", + "LC_ALL", + "PATH", + "PYTHONHOME", + "PYTHONPATH", + "SYSTEMROOT", + "TEMP", + "TMP", + "TMPDIR", + "VIRTUAL_ENV", + "SystemRoot", + } +) + + +@functools.lru_cache(maxsize=1) +def _network_isolation_prefix() -> tuple[str, ...] | None: + """Return a working ``unshare`` prefix that creates an empty network + namespace, or None when the platform cannot isolate egress.""" + for prefix in (("unshare", "-Urn"), ("unshare", "-n")): + try: + probe = subprocess.run( # noqa: S603 + [*prefix, "true"], + capture_output=True, + timeout=15, + check=False, + ) + except (OSError, subprocess.SubprocessError): + continue + if probe.returncode == 0: + return prefix + return None + + +def _command_environment(credentials_allowed: Iterable[str]) -> dict[str, str]: + allowed = _COMMAND_ENV_ALLOWLIST | set(credentials_allowed) + env = {key: value for key, value in os.environ.items() if key in allowed} + env["PYTHONDONTWRITEBYTECODE"] = "1" + return env + + +async def run_command( + workspace: Path, + command: CommandSpec, + *, + credentials_allowed: Iterable[str] = (), + network_allowed: bool = False, +) -> CheckResult: started = time.monotonic() cwd = (workspace / command.cwd).resolve() if not cwd.is_relative_to(workspace.resolve()) or not cwd.is_dir(): @@ -74,12 +125,25 @@ async def run_command(workspace: Path, command: CommandSpec) -> CheckResult: output="The command working directory is unavailable.", required=command.required, ) + argv = list(command.argv) + if not network_allowed: + prefix = _network_isolation_prefix() + if prefix is None: + return CheckResult( + name=command.name, + argv=command.argv, + status=CheckStatus.UNAVAILABLE, + duration_seconds=time.monotonic() - started, + output="Network isolation is unavailable, so the command was not run.", + required=command.required, + ) + argv = [*prefix, *argv] process: asyncio.subprocess.Process | None = None try: process = await asyncio.create_subprocess_exec( - *command.argv, + *argv, cwd=cwd, - env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}, + env=_command_environment(credentials_allowed), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, ) @@ -123,8 +187,11 @@ def _apply_edits(workspace: Path, candidate: FixCandidateV1) -> None: (edit.start_line, edit.end_line, edit.before, edit.after) ) + workspace_resolved = workspace.resolve() for file_path, edits in by_file.items(): - path = workspace / file_path + path = (workspace / file_path).resolve() + if not path.is_relative_to(workspace_resolved): + raise ValueError(f"Draft edit path escapes the workspace: {file_path}") content = path.read_text(encoding="utf-8") lines = content.splitlines(keepends=True) newline = "\r\n" if "\r\n" in content else "\n" @@ -148,6 +215,7 @@ async def build_git_manifest( "git", "status", "--porcelain=v1", + "--untracked-files=all", "-z", cwd=workspace, stdout=asyncio.subprocess.PIPE, @@ -160,6 +228,7 @@ async def build_git_manifest( entries: list[FileManifestEntry] = [] changed_files: list[str] = [] records = [record for record in output.decode(errors="replace").split("\0") if record] + workspace_resolved = workspace.resolve() index = 0 while index < len(records): record = records[index] @@ -176,7 +245,31 @@ async def build_git_manifest( operation = "delete" else: operation = "modify" - resulting = hashlib.sha256(path.read_bytes()).hexdigest() if path.is_file() else None + resolved = path.resolve() + contained = resolved.is_relative_to(workspace_resolved) + if operation == "add" and contained and resolved.is_dir(): + for child in sorted(resolved.rglob("*")): + child_resolved = child.resolve() + if ( + not child_resolved.is_file() + or not child_resolved.is_relative_to(workspace_resolved) + or ".git" in child.relative_to(resolved).parts + ): + continue + entries.append( + FileManifestEntry( + path=child_resolved.relative_to(workspace_resolved).as_posix(), + operation="add", + resulting_sha256=hashlib.sha256(child_resolved.read_bytes()).hexdigest(), + ) + ) + index += 1 + continue + resulting = ( + hashlib.sha256(resolved.read_bytes()).hexdigest() + if contained and resolved.is_file() + else None + ) original: str | None = None if operation != "add": original_process = await asyncio.create_subprocess_exec( @@ -215,18 +308,31 @@ async def build_git_manifest( async def _verify_source(context: PreparationContext) -> bool: identity = context.candidate.source_identity - if identity is None or identity.kind != "commit": - return identity is not None - process = await asyncio.create_subprocess_exec( + if identity is None: + return False + if identity.kind == "commit": + process = await asyncio.create_subprocess_exec( + "git", + "rev-parse", + "HEAD", + cwd=context.workspace, + stdout=asyncio.subprocess.PIPE, + stderr=subprocess.DEVNULL, + ) + output, _ = await process.communicate() + if process.returncode != 0 or output.decode().strip().lower() != identity.value: + return False + status_process = await asyncio.create_subprocess_exec( "git", - "rev-parse", - "HEAD", + "status", + "--porcelain=v1", + "-z", cwd=context.workspace, stdout=asyncio.subprocess.PIPE, stderr=subprocess.DEVNULL, ) - output, _ = await process.communicate() - return process.returncode == 0 and output.decode().strip().lower() == identity.value + status_output, _ = await status_process.communicate() + return status_process.returncode == 0 and not status_output.strip(b"\x00") def _result( @@ -279,6 +385,13 @@ async def prepare_fix( timeout_seconds=request.timeout_seconds, ) context = PreparationContext(request=request, workspace=workspace, candidate=request.candidate) + runner: CommandRunner = command_runner + if runner is run_command: + runner = functools.partial( + run_command, + credentials_allowed=request.credentials_allowed, + network_allowed=request.network_allowed, + ) async def execute() -> FixPreparationResultV1: # noqa: PLR0911, PLR0912 if cancelled(): @@ -323,9 +436,9 @@ async def prepare_fix( if cancelled(): raise PreparationCancelledError await repair(context, checks) - checks = [await command_runner(workspace, check) for check in request.checks] + checks = [await runner(workspace, check) for check in request.checks] if context.candidate.reproduction and context.candidate.reproduction.command: - reproduction = await command_runner( + reproduction = await runner( workspace, context.candidate.reproduction.command, ) @@ -355,6 +468,11 @@ async def prepare_fix( for result in checks if result.required and result.status is CheckStatus.UNAVAILABLE ] + gaps.extend( + f"{result.name}: optional check {result.status}" + for result in checks + if not result.required and result.status is not CheckStatus.PASSED + ) if reproduction is None and not verifier.reproduction_executed: gaps.append("No executable security reproduction was available.") elif reproduction is not None and reproduction.status is CheckStatus.UNAVAILABLE: diff --git a/strix/report/sarif.py b/strix/report/sarif.py index ab97598b8..fcf9c3f77 100644 --- a/strix/report/sarif.py +++ b/strix/report/sarif.py @@ -28,9 +28,10 @@ Design notes: * File locations must be repo-relative POSIX paths. Paths that look like URIs, absolute paths, or traversal patterns are rejected rather than emitted as invalid code-scanning alerts. - * Findings with a fix suggestion (``code_locations[].fix_before`` + - ``fix_after``) are emitted as SARIF ``fixes`` so code-scanning can - render a one-click suggested change. + * Findings whose fix candidate completed verified preparation + (``fix_preparation.state == "ready"`` with a matching + ``candidate_digest``) are emitted as SARIF ``fixes`` so code-scanning + can render a one-click suggested change. * Endpoint / target-only findings (typical of DAST) carry a SARIF ``logicalLocations`` entry so the finding keeps a meaningful anchor even without a source file + line. @@ -56,6 +57,10 @@ import re from pathlib import Path, PurePosixPath from typing import Any, cast +from pydantic import ValidationError + +from strix.fix.contracts import FixCandidateV1 + logger = logging.getLogger(__name__) @@ -583,46 +588,42 @@ def _result_properties( def _build_fixes(report: dict[str, Any]) -> list[dict[str, Any]] | None: - """Build SARIF ``fixes`` from a prepared finding. + """Build SARIF ``fixes`` from a verified prepared finding. SARIF consumers can apply ``fixes`` automatically. Strix emits them only - after the preparation stage records a ``ready`` result. + when preparation recorded a ``ready`` result whose ``candidate_digest`` + still matches the stored fix candidate — so the emitted replacements are + exactly what preparation verified, never a stale or diverged draft. """ preparation = report.get("fix_preparation") if not isinstance(preparation, dict) or preparation.get("state") != "ready": return None - raw_locations = report.get("code_locations") - if not isinstance(raw_locations, list): + raw_candidate = report.get("fix_candidate") + if not isinstance(raw_candidate, dict): + return None + try: + candidate = FixCandidateV1.model_validate(raw_candidate) + except ValidationError: + return None + if preparation.get("candidate_digest") != candidate.digest(): return None artifact_changes: list[dict[str, Any]] = [] - for location in raw_locations: - if not isinstance(location, dict): - continue - file_path = _string_value(location.get("file")) - fix_before = _string_value(location.get("fix_before")) - fix_after = _string_value(location.get("fix_after")) - start_line = location.get("start_line") - if not (file_path and fix_before and fix_after): - continue - if type(start_line) is not int or start_line < 1: - continue - uri = _sarif_uri(file_path) + for edit in candidate.draft_edits: + uri = _sarif_uri(edit.file) if uri is None: continue - - deleted_region: dict[str, Any] = {"startLine": start_line} - end_line = location.get("end_line") - if type(end_line) is int and end_line >= start_line: - deleted_region["endLine"] = end_line - + deleted_region: dict[str, Any] = { + "startLine": edit.start_line, + "endLine": edit.end_line, + } artifact_changes.append( { "artifactLocation": {"uri": uri}, "replacements": [ { "deletedRegion": deleted_region, - "insertedContent": {"text": fix_after}, + "insertedContent": {"text": edit.after}, } ], } diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index ffc6b40fb..a9888364a 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -413,15 +413,24 @@ def _build_fix_candidate( if candidate is None: return None if repo_path is not None: - anchored, results = anchor_candidate(repo_path, candidate) - candidate = anchored - gaps = [ - f"{result.location.file}: {result.status}" - for result in results - if result.status is not AnchorStatus.UNIQUE - ] - if gaps: - candidate = candidate.model_copy(update={"known_gaps": [*candidate.known_gaps, *gaps]}) + try: + anchored, results = anchor_candidate(repo_path, candidate) + except Exception as exc: # noqa: BLE001 + # Anchoring must never block the finding from being stored. + candidate = candidate.model_copy( + update={"known_gaps": [*candidate.known_gaps, f"Candidate anchoring failed: {exc}"]} + ) + else: + candidate = anchored + gaps = [ + f"{result.location.file}: {result.status}" + for result in results + if result.status is not AnchorStatus.UNIQUE + ] + if gaps: + candidate = candidate.model_copy( + update={"known_gaps": [*candidate.known_gaps, *gaps]} + ) return cast("dict[str, object]", candidate.model_dump(mode="json")) diff --git a/tests/test_fix_preparation.py b/tests/test_fix_preparation.py index 02933cedd..1d10d4855 100644 --- a/tests/test_fix_preparation.py +++ b/tests/test_fix_preparation.py @@ -27,7 +27,13 @@ from strix.fix.contracts import ( candidate_from_legacy_report, ) from strix.fix.locations import AnchorStatus, anchor_location -from strix.fix.prepare import PreparationContext, prepare_fix +from strix.fix.prepare import ( + PreparationContext, + _network_isolation_prefix, + build_git_manifest, + prepare_fix, + run_command, +) if TYPE_CHECKING: @@ -382,3 +388,149 @@ async def test_prepare_fix_rejects_wrong_source_commit(tmp_path: Path) -> None: assert result.state is PreparationState.STALE assert "source identity" in result.stop_reason + + +def test_anchor_location_rejects_symlink_escape(tmp_path: Path) -> None: + outside = tmp_path / "outside.txt" + outside.write_text("target\n", encoding="utf-8") + root = tmp_path / "repo" + root.mkdir() + (root / "link.txt").symlink_to(outside) + location = CandidateLocation( + file="link.txt", + start_line=1, + end_line=1, + snippet="target", + ) + + assert anchor_location(root, location).status is AnchorStatus.MISSING + + +def test_anchor_location_treats_unreadable_file_as_missing(tmp_path: Path) -> None: + (tmp_path / "blob.bin").write_bytes(b"\x89PNG\r\n\x1a\n\x00\xff\xfe") + location = CandidateLocation( + file="blob.bin", + start_line=1, + end_line=1, + snippet="blob", + ) + + assert anchor_location(tmp_path, location).status is AnchorStatus.MISSING + + +@pytest.mark.asyncio +async def test_prepare_fix_rejects_edit_escaping_workspace(tmp_path: Path) -> None: + workspace, _commit = _workspace(tmp_path) + outside = tmp_path / "outside.txt" + outside.write_text("target\n", encoding="utf-8") + (workspace / "link.txt").symlink_to(outside) + _git(workspace, "add", "link.txt") + _git(workspace, "commit", "-m", "add link") + commit = _git(workspace, "rev-parse", "HEAD") + candidate = FixCandidateV1( + source_identity=SourceIdentity(kind=SourceIdentityKind.COMMIT, value=commit), + security_invariant="Replace target.", + draft_edits=[ + FixEdit( + file="link.txt", + start_line=1, + end_line=1, + before="target", + after="safe", + ) + ], + ) + + result = await prepare_fix( + _request(candidate), + workspace, + repair=_noop_repair, + verify=_verified, + ) + + assert result.state is PreparationState.NEEDS_REVIEW + assert outside.read_text(encoding="utf-8") == "target\n" + + +@pytest.mark.asyncio +async def test_prepare_fix_rejects_dirty_workspace(tmp_path: Path) -> None: + workspace, commit = _workspace(tmp_path) + (workspace / "stray.txt").write_text("unrelated\n", encoding="utf-8") + + result = await prepare_fix( + _request(_candidate(commit)), + workspace, + repair=_noop_repair, + verify=_verified, + ) + + assert result.state is PreparationState.STALE + assert "source identity" in result.stop_reason + + +@pytest.mark.asyncio +async def test_build_git_manifest_lists_files_inside_new_directory(tmp_path: Path) -> None: + workspace, _commit = _workspace(tmp_path) + package = workspace / "pkg" + package.mkdir() + (package / "mod.py").write_text("x = 1\n", encoding="utf-8") + + entries, _summary, _artifact = await build_git_manifest(workspace) + + paths = {entry.path for entry in entries} + assert "pkg/mod.py" in paths + entry = next(entry for entry in entries if entry.path == "pkg/mod.py") + assert entry.operation == "add" + assert entry.resulting_sha256 is not None + + +@pytest.mark.asyncio +async def test_run_command_drops_ambient_credentials( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("STRIX_AMBIENT_TOKEN", "hunter2") + command = CommandSpec( + name="env probe", + argv=[ + sys.executable, + "-c", + "import os; print(os.environ.get('STRIX_AMBIENT_TOKEN', ''))", + ], + ) + + sealed = await run_command(tmp_path, command, network_allowed=True) + assert sealed.status is CheckStatus.PASSED + assert "" in sealed.output + + granted = await run_command( + tmp_path, + command, + credentials_allowed=["STRIX_AMBIENT_TOKEN"], + network_allowed=True, + ) + assert granted.status is CheckStatus.PASSED + assert "hunter2" in granted.output + + +@pytest.mark.asyncio +async def test_run_command_blocks_egress_when_network_not_allowed(tmp_path: Path) -> None: + command = CommandSpec( + name="egress probe", + argv=[ + sys.executable, + "-c", + ( + "import socket, sys; s = socket.socket(); s.settimeout(3); " + "sys.exit(0 if s.connect_ex(('1.1.1.1', 53)) == 0 else 1)" + ), + ], + ) + + result = await run_command(tmp_path, command) + + if _network_isolation_prefix() is None: + assert result.status is CheckStatus.UNAVAILABLE + assert "not run" in result.output + else: + assert result.status is CheckStatus.FAILED diff --git a/tests/test_sarif.py b/tests/test_sarif.py index a81bc129f..affe5034a 100644 --- a/tests/test_sarif.py +++ b/tests/test_sarif.py @@ -5,6 +5,7 @@ from __future__ import annotations import json from typing import TYPE_CHECKING, Any +from strix.fix.contracts import FixCandidateV1 from strix.report.sarif import write_sarif @@ -129,15 +130,37 @@ def test_write_sarif_never_embeds_poc_script(tmp_path: Path) -> None: assert poc["description"] == "Send a crafted request to trigger the sink." +def _fix_candidate(**overrides: Any) -> dict[str, Any]: + candidate: dict[str, Any] = { + "security_invariant": "Use a parameterized query.", + "draft_edits": [ + { + "file": "app.py", + "start_line": 4, + "end_line": 4, + "before": 'query = "SELECT * FROM u WHERE id=" + uid', + "after": 'query = "SELECT * FROM u WHERE id=%s"', + } + ], + } + candidate.update(overrides) + return candidate + + def test_write_sarif_builds_fixes_from_ready_preparation(tmp_path: Path) -> None: # SARIF fixes are automatically applicable, so only a prepared result can - # expose the candidate as an artifactChange. + # expose the verified candidate as an artifactChange. + candidate = _fix_candidate() write_sarif( tmp_path, [ _finding( remediation_steps="Use a parameterized query.", - fix_preparation={"state": "ready"}, + fix_candidate=candidate, + fix_preparation={ + "state": "ready", + "candidate_digest": FixCandidateV1.model_validate(candidate).digest(), + }, code_locations=[ { "file": "app.py", @@ -160,6 +183,35 @@ def test_write_sarif_builds_fixes_from_ready_preparation(tmp_path: Path) -> None assert replacement["insertedContent"]["text"] == 'query = "SELECT * FROM u WHERE id=%s"' +def test_write_sarif_suppresses_fixes_when_prepared_candidate_diverged( + tmp_path: Path, +) -> None: + # A ready result whose digest no longer matches the recorded candidate + # means repair changed what was verified — emitting the draft would offer + # the wrong replacement, so SARIF stays silent. + write_sarif( + tmp_path, + [ + _finding( + fix_candidate=_fix_candidate(), + fix_preparation={"state": "ready", "candidate_digest": "0" * 64}, + code_locations=[ + { + "file": "app.py", + "start_line": 4, + "end_line": 4, + "fix_before": "unsafe(value)", + "fix_after": "safe(value)", + } + ], + ) + ], + ) + + result = _read(tmp_path)["runs"][0]["results"][0] + assert "fixes" not in result + + def test_write_sarif_hides_unprepared_fix_candidate(tmp_path: Path) -> None: write_sarif( tmp_path,