diff --git a/docs/fix-preparation.md b/docs/fix-preparation.md index 4c9c127bb..0c719dbb1 100644 --- a/docs/fix-preparation.md +++ b/docs/fix-preparation.md @@ -87,8 +87,11 @@ strix fix --repo ./repo --request request.json --output ./fix-result/result.json `--workspace` is an alias for `--repo`. `--artifact` overrides the archive path; `--max-agent-turns`, `--timeout`, and `--max-budget` override request budgets. Outputs are result JSON, a readable Markdown review, a patch, and the full ZIP -artifact. Without `--output`, they go in a new `strix_runs/fix-…` directory. Use an -output directory outside the source checkout to keep it clean for the next run. +artifact. Without `--output`, they go in a new `~/.strix/fixes/fix-…` directory +outside the source checkout. If that location is itself inside the repository, +choose an external directory with `--output`. Explicit output directories can be +shared: result files and the ZIP are private from creation (0600 on POSIX) and +published atomically. Existing directory permissions are left unchanged. Exit codes: 0 approved, 2 incomplete/blocked/stale, 1 startup or input failure, 130 interrupted. Interruptions save any checkpointed work in the ZIP archive. Partial patches and their limitations are retained when review cannot approve. diff --git a/strix/fix/runtime.py b/strix/fix/runtime.py index 8e75c6255..bbb9262a7 100644 --- a/strix/fix/runtime.py +++ b/strix/fix/runtime.py @@ -68,6 +68,7 @@ from strix.fix.workspace import ( from strix.report.usage import LLMUsageLedger from strix.runtime import session_manager from strix.tools.thinking.tool import think +from strix.utils.secret_files import open_secret_file if TYPE_CHECKING: @@ -603,13 +604,11 @@ async def run_fix_preparation( if artifact_path is None: return manifest, summary, None destination = artifact_path.resolve() - destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700) patch_output = await build_git_patch(root, manifest) - with zipfile.ZipFile( - destination, - mode="w", - compression=zipfile.ZIP_DEFLATED, - ) as archive: + with ( + open_secret_file(destination) as stream, + zipfile.ZipFile(stream, mode="w", compression=zipfile.ZIP_DEFLATED) as archive, + ): archive.writestr( "manifest.json", json.dumps( @@ -641,7 +640,6 @@ async def run_fix_preparation( continue source = environment.resolve(entry.path) archive.write(source, f"files/{entry.path}") - destination.chmod(0o600) return manifest, summary, str(destination) environment.max_agent_turns = request.max_agent_turns diff --git a/strix/interface/fix_cli.py b/strix/interface/fix_cli.py index d859f4a94..5bc74747b 100644 --- a/strix/interface/fix_cli.py +++ b/strix/interface/fix_cli.py @@ -15,7 +15,6 @@ from rich.console import Console from strix.config import load_settings from strix.config.models import configure_sdk_model_defaults -from strix.core.paths import run_dir_for from strix.fix import ( FixCandidateV1, FixPreparationRequestV1, @@ -25,6 +24,7 @@ from strix.fix import ( from strix.fix.runtime import run_isolated_fix_preparation from strix.interface.environment import check_docker_installed, pull_docker_image from strix.interface.scan_setup import preflight_model_connection +from strix.utils.secret_files import open_secret_file, write_secret_text def _parser() -> argparse.ArgumentParser: @@ -35,7 +35,7 @@ def _parser() -> argparse.ArgumentParser: parser.add_argument("--finding-id", help="Finding ID to select from vulnerabilities.json.") parser.add_argument("--repo", "--workspace", dest="repo", type=Path, required=True) parser.add_argument( - "--output", type=Path, help="Result JSON; defaults to a new strix_runs folder." + "--output", type=Path, help="Result JSON; defaults to a new ~/.strix/fixes/ folder." ) parser.add_argument( "--artifact", type=Path, help="Patch/log archive; defaults beside the result." @@ -130,23 +130,34 @@ async def _execute( ) -> FixPreparationResultV1: await _preflight() result = await run_isolated_fix_preparation(request, repo, artifact_path=artifact) - output.parent.mkdir(parents=True, exist_ok=True) - output.write_text(result.model_dump_json(indent=2) + "\n", encoding="utf-8") - output.with_suffix(".md").write_text(_summary(result), encoding="utf-8") - with zipfile.ZipFile(artifact) as archive: - output.with_suffix(".patch").write_bytes(archive.read("changes.patch")) + write_secret_text(output, result.model_dump_json(indent=2) + "\n") + write_secret_text(output.with_suffix(".md"), _summary(result)) + with ( + zipfile.ZipFile(artifact) as archive, + open_secret_file(output.with_suffix(".patch")) as stream, + ): + stream.write(archive.read("changes.patch")) return result +def _default_output(repo: Path) -> Path: + output = ( + Path.home() / ".strix" / "fixes" / f"fix-{uuid.uuid4().hex[:12]}" / "result.json" + ).resolve() + if output.is_relative_to(repo.resolve()): + raise ValueError( + "The default results directory is inside this repository; set --output outside it." + ) + return output + + def run_fix(argv: list[str]) -> int: """Return 0 approved, 2 incomplete, 1 startup failure, or 130 interrupted.""" args = _parser().parse_args(argv) console = Console() try: request = _load_request(args) - output = ( - args.output or run_dir_for(f"fix-{uuid.uuid4().hex[:12]}") / "result.json" - ).resolve() + output = (args.output or _default_output(args.repo)).resolve() artifact = (args.artifact or output.with_suffix(".zip")).resolve() # Result files must not overwrite source or a previous preparation's evidence. paths = [output, artifact, output.with_suffix(".md"), output.with_suffix(".patch")] diff --git a/strix/utils/secret_files.py b/strix/utils/secret_files.py index 023aa87f1..de41e64b7 100644 --- a/strix/utils/secret_files.py +++ b/strix/utils/secret_files.py @@ -1,39 +1,41 @@ from __future__ import annotations -import contextlib import os +import tempfile +from contextlib import contextmanager +from pathlib import Path from typing import TYPE_CHECKING if TYPE_CHECKING: - from pathlib import Path + from collections.abc import Generator + from typing import BinaryIO SECRET_FILE_MODE = 0o600 -def write_secret_text(path: Path, text: str) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - - tmp = path.with_suffix(path.suffix + ".tmp") - with contextlib.suppress(FileNotFoundError): - tmp.unlink() - - fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, SECRET_FILE_MODE) - try: - with os.fdopen(fd, "w", encoding="utf-8") as handle: - handle.write(text) - except BaseException as exc: - _cleanup_tmp(tmp, exc) - raise - +@contextmanager +def open_secret_file(path: Path) -> Generator[BinaryIO, None, None]: + """Write through a private sibling file and publish it atomically on success.""" + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + # mkstemp creates the inode with mode 0600, before any caller writes content. + fd, name = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp") + tmp = Path(name) try: + with os.fdopen(fd, "wb") as handle: + yield handle tmp.replace(path) except BaseException as exc: _cleanup_tmp(tmp, exc) raise +def write_secret_text(path: Path, text: str) -> None: + with open_secret_file(path) as handle: + handle.write(text.encode("utf-8")) + + def _cleanup_tmp(tmp: Path, cause: BaseException) -> None: """Delete the temporary secret file. A failed delete must not stay silent.""" try: diff --git a/tests/test_fix_cli.py b/tests/test_fix_cli.py index 9360c38ac..33bf58096 100644 --- a/tests/test_fix_cli.py +++ b/tests/test_fix_cli.py @@ -5,8 +5,11 @@ from __future__ import annotations import asyncio import importlib import json +import os +import stat import sys import zipfile +from pathlib import Path import pytest from agents import RunConfig @@ -206,3 +209,82 @@ def test_legacy_empty_credential_field_is_accepted_but_forwarding_is_rejected(): FixPreparationRequestV1.model_validate({**request, "credentials_allowed": []}) with pytest.raises(ValueError, match="credentials_allowed"): FixPreparationRequestV1.model_validate({**request, "credentials_allowed": ["ANY_HOST_KEY"]}) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX permission bits") +def test_cli_outputs_and_in_progress_archive_are_private_in_shared_directory(tmp_path, monkeypatch): + workspace, _ = _workspace(tmp_path) + commit = existing_suite(workspace) + request_path = tmp_path / "request.json" + request_path.write_text(_request(commit).model_dump_json()) + shared = tmp_path / "shared-output" + shared.mkdir() + shared.chmod(0o777) + output = shared / "result.json" + model = ScriptedModel([*patch(), finish("done")], [*suite_commands(), finish("approved")]) + _local_runtime(monkeypatch, tmp_path, model) + original_writestr = zipfile.ZipFile.writestr + writes_checked = [] + + def private_writestr(archive, name, data, *args, **kwargs): + # Check the open archive before source/log bytes enter it, not just after close. + assert stat.S_IMODE(os.fstat(archive.fp.fileno()).st_mode) == 0o600 + writes_checked.append(name) + return original_writestr(archive, name, data, *args, **kwargs) + + monkeypatch.setattr(zipfile.ZipFile, "writestr", private_writestr) + previous = os.umask(0) + try: + assert ( + fix_cli.run_fix( + [ + "--request", + str(request_path), + "--repo", + str(workspace), + "--output", + str(output), + ] + ) + == 0 + ) + finally: + os.umask(previous) + assert "changes.patch" in writes_checked + assert "agent-sessions.json" in writes_checked + assert stat.S_IMODE(shared.stat().st_mode) == 0o777 + assert {p.name for p in shared.iterdir()} == { + "result.json", + "result.md", + "result.patch", + "result.zip", + } + assert all(stat.S_IMODE(p.stat().st_mode) == 0o600 for p in shared.iterdir()) + + +def test_default_outputs_allow_repeated_runs_from_inside_the_repository(tmp_path, monkeypatch): + workspace, _ = _workspace(tmp_path) + commit = existing_suite(workspace) + request_path = tmp_path / "request.json" + request_path.write_text(_request(commit).model_dump_json()) + home = tmp_path / "home" + home.mkdir() + monkeypatch.setattr(Path, "home", lambda: home) + monkeypatch.chdir(workspace) + + for attempt in range(2): + model = ScriptedModel([*patch(), finish("done")], [*suite_commands(), finish("approved")]) + with monkeypatch.context() as runtime_patch: + _local_runtime(runtime_patch, tmp_path / f"attempt-{attempt}", model) + assert fix_cli.run_fix(["--request", str(request_path), "--repo", "."]) == 0 + assert _git(workspace, "status", "--porcelain") == "" + + assert len(list((home / ".strix/fixes").glob("fix-*/result.json"))) == 2 + assert not (workspace / "strix_runs").exists() + + +def test_default_output_cannot_resolve_inside_source_checkout(tmp_path, monkeypatch): + monkeypatch.setattr(Path, "home", lambda: tmp_path) + with pytest.raises(ValueError, match="set --output outside"): + fix_cli._default_output(tmp_path) + assert not (tmp_path / ".strix").exists() diff --git a/tests/test_secret_files.py b/tests/test_secret_files.py index f93c35418..f93478fc7 100644 --- a/tests/test_secret_files.py +++ b/tests/test_secret_files.py @@ -8,7 +8,7 @@ from typing import TYPE_CHECKING import pytest -from strix.utils.secret_files import SECRET_FILE_MODE, write_secret_text +from strix.utils.secret_files import SECRET_FILE_MODE, open_secret_file, write_secret_text if TYPE_CHECKING: @@ -63,3 +63,31 @@ def test_overwriting_an_existing_record_keeps_it_restricted(tmp_path: Path) -> N assert json.loads(target.read_text(encoding="utf-8"))["v"] == 2 if sys.platform != "win32": assert stat.S_IMODE(target.stat().st_mode) == SECRET_FILE_MODE + + +@posix_only +def test_stream_is_private_before_writing_and_atomic_on_completion(tmp_path: Path) -> None: + target = tmp_path / "shared" / "result.zip" + target.parent.mkdir(mode=0o777) + previous = os.umask(0) + try: + with open_secret_file(target) as stream: + assert stat.S_IMODE(os.fstat(stream.fileno()).st_mode) == SECRET_FILE_MODE + assert not target.exists() + stream.write(b"private customer source\x00\xff") + stream.flush() + assert not target.exists() + assert target.read_bytes() == b"private customer source\x00\xff" + assert stat.S_IMODE(target.stat().st_mode) == SECRET_FILE_MODE + finally: + os.umask(previous) + + +def test_failed_stream_preserves_previous_result_and_removes_partial_file(tmp_path: Path) -> None: + target = tmp_path / "result.zip" + write_secret_text(target, "previous result") + with pytest.raises(RuntimeError, match="interrupted write"), open_secret_file(target) as stream: + stream.write(b"partial customer source") + raise RuntimeError("interrupted write") + assert target.read_text() == "previous result" + assert list(tmp_path.iterdir()) == [target]