Compare commits

..
38 changed files with 203 additions and 2481 deletions

No files matched your search

+11 -3
View File
@@ -1,4 +1,4 @@
.PHONY: help install dev-install format lint type-check security check-all clean pre-commit setup-dev dev viewer wheel tui-build tui-test tui-lint
.PHONY: help install dev-install format lint format-check lint-check type-check security check-all clean pre-commit setup-dev dev viewer wheel tui-build tui-test tui-lint
TUI_BINARY := build/sidecar/strix-tui$(if $(filter Windows_NT,$(OS)),.exe)
@@ -10,7 +10,9 @@ help:
@echo ""
@echo "Code Quality:"
@echo " format - Format code with ruff"
@echo " lint - Lint code with ruff"
@echo " lint - Lint code with ruff and apply fixes"
@echo " format-check - Check formatting without modifying files"
@echo " lint-check - Check lint without modifying files"
@echo " type-check - Run type checking with mypy and pyright"
@echo " security - Run security checks with bandit"
@echo " check-all - Run all code quality checks"
@@ -45,6 +47,12 @@ lint:
uv run ruff check . --fix
@echo "✅ Linting complete!"
format-check:
uv run ruff format --check .
lint-check:
uv run ruff check .
type-check:
@echo "🔍 Type checking with mypy..."
uv run mypy strix/
@@ -57,7 +65,7 @@ security:
uv run bandit -r strix/ -c pyproject.toml
@echo "✅ Security checks complete!"
check-all: format lint type-check security
check-all: format-check lint-check type-check security
@echo "✅ All code quality checks passed!"
pre-commit:
+9 -12
View File
@@ -153,9 +153,8 @@ WHITE-BOX TESTING (code provided):
- Local execution, unit/integration testing, patch verification, and HTTP requests against locally started in-scope services are normal authorized white-box validation
- If dynamically running the code proves impossible after exhaustive attempts, pivot to comprehensive static analysis.
- Try to infer how to run the code based on its structure and content.
- Draft the initial fix candidate when you file the report. Use `code_locations` with verbatim `fix_before` and `fix_after`, plus `fix_pr_body`.
- Treat the inline changes as a candidate, not as a completed fix. A later preparation stage can inspect and modify any required repository file.
- Record checks you ran in `fix_verification`. Do not describe reasoned checks as executed checks.
- Derive the code fix as PART OF reporting, not as a separate later pass: create_vulnerability_report already requires the concrete patch inline (`code_locations` with verbatim `fix_before`/`fix_after` and `fix_pr_body`), so the reporting agent that analyzes the root cause is the one that produces the fix. Do NOT spawn a downstream agent afterwards to re-derive/re-apply the same patch.
- If you also apply and verify the patch in the repo (edit the file, re-test that the vulnerability is gone), do it in the same agent/turn while the analysis is fresh — right before or as part of filing the report — never as a second re-analysis pass.
COMBINED MODE (code + deployed target present):
- Treat this as static analysis plus dynamic testing simultaneously
@@ -241,9 +240,7 @@ VALIDATION REQUIREMENTS:
- THREAT MODEL: before you start testing, call `get_threat_model` on the target you were pointed at — it is the scan's shared answer to who the attacker is, where the trust boundaries sit, and what counts as critical here. It is scoped to this scan and nothing carries over from an earlier run, so `found: false` means no agent on this run has derived one yet. Read it instead of re-deriving trust boundaries yourself; where your testing disproves it — a boundary it calls trusted turns out to be attacker-reachable, a role it did not know about, a host or endpoint it never listed — record that with `amend_threat_model` so the agents after you inherit the correction. Amending is not optional politeness: a model nobody corrects turns the first agent's guesses into everyone's assumptions.
- Before filing any report, run the counterevidence pass: argue the strongest case AGAINST the finding, record what you found in the `counterevidence` field, set `confidence` honestly (a static-only trace you couldn't execute is at best `medium`), and state what evidence would change the severity. See the counterevidence and severity-calibration knowledge above.
- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
- When source is available, the reporting agent files an initial fix candidate with the report. The candidate uses `code_locations` with `fix_before` and `fix_after`, plus `fix_pr_body`.
- Do not treat the candidate as a prepared fix. The fix preparation stage applies, repairs, tests, and independently verifies it after finding discovery closes.
- Do not silently patch a finding without filing a report.
- Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via create_vulnerability_report (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)
- DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent. If your evidence proves more than the finding it matched (a working exploit where that one had only a static trace, a chain that raises the impact), revise that finding with update_vulnerability_report using the duplicate_of id — never re-file it.
- HTTP EVIDENCE: a finding you validated through the proxy is not fully filed until `http_exchange_ids` carries the proxy request ids of the exchanges that prove it — the request that triggers the vulnerability plus the baseline/control request it differs from (an unauthenticated success next to the authenticated one, the payload response next to the benign one). Copy the ids exactly as `list_requests`/`view_request` show them, never invent or guess one, and never omit the field to bypass validation. Leave it out only when there is no captured HTTP exchange at all (static-only code findings, dependency CVEs). If you filed before the proving exchanges existed, attach them afterwards with update_vulnerability_report. Without the ids, the finding ships as prose nobody can replay.
- REVISING A FINDING: use update_vulnerability_report (report id + the fields you want to replace + update_reason) when you learn something a finding already on file does not carry — you built the PoC after filing it, a chain raised its impact, further testing weakened it, or its counterevidence/remediation/code locations were wrong. Editing a finding needs no duplicate verdict, and it is always better than filing a second report for the same issue. Read the finding first with get_report, and pass only the fields that change.
@@ -363,7 +360,7 @@ ROOT AGENT ROLE:
1. **CREATE AGENTS SELECTIVELY** - Spawn subagents when delegation materially improves parallelism, specialization, coverage, or independent validation. Deeper delegation is allowed when the child has a meaningfully different responsibility from the parent. Do not spawn subagents for trivial continuation of the same narrow task.
2. **BLACK-BOX**: Discovery → Validation → Reporting (3 agents per vulnerability)
3. **WHITE-BOX**: Discovery → Validation → Reporting with an initial fix candidate. The later preparation phase repairs and verifies the candidate.
3. **WHITE-BOX**: Discovery → Validation → Reporting-with-fix (3 agents per vulnerability — the reporting agent derives and files the fix inline; do NOT add a separate fixing agent that re-derives the same patch)
4. **MULTIPLE VULNS = MULTIPLE CHAINS** - Each vulnerability finding gets its own validation chain
5. **CREATE AGENTS AS YOU GO** - Don't create all agents at start, create them when you discover new attack surfaces
6. **ONE JOB PER AGENT** - Each agent has ONE specific task only
@@ -382,7 +379,7 @@ BLACK-BOX (domain/URL only):
WHITE-BOX (source code provided):
- Found authentication code issues? → Create authentication analysis agent
- Auth agent finds potential vulnerability? → Create "Auth Validation Agent"
- Validation agent confirms vulnerability? → Create "Auth Reporting Agent" that files the report and an initial fix candidate (`code_locations` + `fix_pr_body`)
- Validation agent confirms vulnerability? → Create "Auth Reporting Agent" that files the report AND its inline fix (`code_locations` + `fix_pr_body`) in one shot — no separate fixing agent
VULNERABILITY WORKFLOW (MANDATORY FOR EVERY FINDING):
@@ -404,10 +401,10 @@ Authentication Code Agent finds weak password validation
Spawns "Auth Validation Agent" (proves it's exploitable)
↓
If valid → Spawns "Auth Reporting Agent" (creates the vulnerability report
with the initial fix candidate: code_locations fix_before/fix_after
+ fix_pr_body)
WITH the fix inline: code_locations fix_before/fix_after + fix_pr_body,
applying/verifying the patch in the same turn if desired)
↓
STOP - the scan preparation phase handles repository-wide repair and verification
STOP - no separate fixing agent; the fix was derived once, at report time
```
CRITICAL RULES:
@@ -443,7 +440,7 @@ FOCUS PRINCIPLES:
REALISTIC TESTING OUTCOMES:
- **No Findings**: Agent completes testing but finds no vulnerabilities
- **Validation Failed**: Initial finding was false positive, validation agent confirms it's not exploitable
- **Valid Vulnerability**: Validation succeeds. The reporting agent files the report and an initial fix candidate for white-box findings.
- **Valid Vulnerability**: Validation succeeds, spawns a reporting agent that files the report with the fix inline (white-box) — no separate fixing agent
PERSISTENCE IS MANDATORY:
- Real vulnerabilities take TIME - expect to need 2000+ steps minimum
-64
View File
@@ -1,64 +0,0 @@
"""Verified fix preparation contracts and runtime."""
from strix.fix.contracts import (
BlockerKind,
CandidateLocation,
CheckResult,
CheckStatus,
CommandSpec,
FileManifestEntry,
FixCandidateV1,
FixEdit,
FixPreparationAttempt,
FixPreparationRequestV1,
FixPreparationResultV1,
PreparationBlocker,
PreparationState,
RepairOutcome,
RepairStatus,
ReproductionSpec,
SourceIdentity,
SourceIdentityKind,
VerificationDecision,
VerificationTarget,
VerifierResult,
candidate_from_legacy_report,
)
from strix.fix.prepare import (
PreparationCancelledError,
PreparationContext,
PreparationPolicy,
build_git_manifest,
prepare_fix,
)
__all__ = [
"BlockerKind",
"CandidateLocation",
"CheckResult",
"CheckStatus",
"CommandSpec",
"FileManifestEntry",
"FixCandidateV1",
"FixEdit",
"FixPreparationAttempt",
"FixPreparationRequestV1",
"FixPreparationResultV1",
"PreparationBlocker",
"PreparationCancelledError",
"PreparationContext",
"PreparationPolicy",
"PreparationState",
"RepairOutcome",
"RepairStatus",
"ReproductionSpec",
"SourceIdentity",
"SourceIdentityKind",
"VerificationDecision",
"VerificationTarget",
"VerifierResult",
"build_git_manifest",
"candidate_from_legacy_report",
"prepare_fix",
]
-331
View File
@@ -1,331 +0,0 @@
"""Versioned contracts for verified fix preparation."""
from __future__ import annotations
import hashlib
import json
from enum import StrEnum
from pathlib import PurePosixPath
from typing import TYPE_CHECKING, Literal, cast
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
if TYPE_CHECKING:
from collections.abc import Mapping
class ContractModel(BaseModel):
model_config = ConfigDict(extra="forbid")
class SourceIdentityKind(StrEnum):
COMMIT = "commit"
ARCHIVE = "archive"
class PreparationState(StrEnum):
PREPARING = "preparing"
READY = "ready"
BLOCKED = "blocked"
FAILED = "failed"
STALE = "stale"
class CheckStatus(StrEnum):
PASSED = "passed"
FAILED = "failed"
UNAVAILABLE = "unavailable"
CANCELLED = "cancelled"
class VerificationTarget(StrEnum):
BASE = "base"
PATCHED = "patched"
class VerificationDecision(StrEnum):
VERIFIED = "verified"
REJECTED = "rejected"
INCONCLUSIVE = "inconclusive"
class RepairStatus(StrEnum):
COMPLETE = "complete"
BLOCKED = "blocked"
BUDGET_EXHAUSTED = "budget_exhausted"
INCOMPLETE = "incomplete"
class BlockerKind(StrEnum):
SOURCE = "source"
ENVIRONMENT = "environment"
REPOSITORY_BASELINE = "repository_baseline"
CREDENTIAL = "credential"
EXTERNAL_CONFIGURATION = "external_configuration"
SECURITY_EVIDENCE = "security_evidence"
class PreparationBlocker(ContractModel):
kind: BlockerKind
summary: str = Field(min_length=1)
user_action: str = Field(min_length=1)
details: list[str] = []
class SourceIdentity(ContractModel):
kind: SourceIdentityKind
value: str = Field(min_length=1)
repository: str | None = None
@field_validator("value")
@classmethod
def validate_value(cls, value: str) -> str:
normalized = value.strip().lower()
if not normalized or any(char not in "0123456789abcdef" for char in normalized):
raise ValueError("source identity must be a hexadecimal digest")
return normalized
@model_validator(mode="after")
def validate_digest_length(self) -> SourceIdentity:
valid_lengths = {40, 64} if self.kind is SourceIdentityKind.COMMIT else {64}
if len(self.value) not in valid_lengths:
expected = "40 or 64" if self.kind is SourceIdentityKind.COMMIT else "64"
raise ValueError(f"{self.kind} identity must contain {expected} hexadecimal characters")
return self
def _validate_relative_path(value: str) -> str:
path = value.strip()
parsed = PurePosixPath(path)
if not path or parsed.is_absolute() or ".." in parsed.parts or path.startswith("./"):
raise ValueError("path must be relative to the repository root")
return path
class CandidateLocation(ContractModel):
file: str
start_line: int = Field(ge=1)
end_line: int = Field(ge=1)
snippet: str | None = None
label: str | None = None
_relative_file = field_validator("file")(_validate_relative_path)
@model_validator(mode="after")
def validate_range(self) -> CandidateLocation:
if self.end_line < self.start_line:
raise ValueError("end_line must be greater than or equal to start_line")
return self
class FixEdit(CandidateLocation):
before: str = Field(min_length=1)
after: str
original_sha256: str | None = None
class CommandSpec(ContractModel):
name: str = Field(min_length=1, max_length=120)
argv: list[str] = Field(min_length=1)
required: bool = True
timeout_seconds: int = Field(default=300, ge=1, le=3600)
cwd: str = "."
_relative_cwd = field_validator("cwd")(_validate_relative_path)
@field_validator("argv")
@classmethod
def validate_argv(cls, value: list[str]) -> list[str]:
if any(not argument or "\x00" in argument for argument in value):
raise ValueError("command arguments must be non-empty and cannot contain NUL")
return value
class ReproductionSpec(ContractModel):
instructions: str = Field(min_length=1)
command: CommandSpec | None = None
class ReportedCheck(ContractModel):
name: str
result: str
executed: bool = False
class FixCandidateV1(ContractModel):
version: Literal["1"] = "1"
source_identity: SourceIdentity | None = None
security_invariant: str = Field(min_length=1)
finding_locations: list[CandidateLocation] = []
draft_edits: list[FixEdit] = []
reproduction: ReproductionSpec | None = None
reported_checks: list[ReportedCheck] = []
known_gaps: list[str] = []
def digest(self) -> str:
payload = json.dumps(
self.model_dump(mode="json"),
sort_keys=True,
separators=(",", ":"),
).encode()
return hashlib.sha256(payload).hexdigest()
class FixPreparationRequestV1(ContractModel):
version: Literal["1"] = "1"
organization_id: str | None = None
scan_id: str
finding_id: str
repository_id: str | None = None
candidate: FixCandidateV1
checks: list[CommandSpec] = []
max_repair_attempts: int = Field(default=2, ge=1, le=2)
timeout_seconds: int = Field(default=1800, ge=30, le=14400)
network_allowed: bool = False
credentials_allowed: list[str] = []
class CheckResult(ContractModel):
name: str
argv: list[str]
status: CheckStatus
exit_code: int | None = None
duration_seconds: float = Field(ge=0)
output: str = ""
required: bool = True
target: VerificationTarget | None = None
baseline_status: CheckStatus | None = None
baseline_output: str | None = None
class VerifierResult(ContractModel):
decision: VerificationDecision
summary: str
security_invariant_closed: bool = False
reproduction_executed: bool = False
reproduction_summary: str | None = None
sibling_paths_reviewed: list[str] = []
preserved_behaviors: list[str] = []
gaps: list[str] = []
security_tests: list[CheckResult] = []
repairable: bool = False
blocker: PreparationBlocker | None = None
class RepairOutcome(ContractModel):
status: RepairStatus
summary: str = Field(min_length=1)
gaps: list[str] = []
reproduction_command: CommandSpec | None = None
turns_used: int = Field(default=0, ge=0)
blocker: PreparationBlocker | None = None
class FixPreparationAttempt(ContractModel):
attempt: int = Field(ge=1)
repair: RepairOutcome
checks: list[CheckResult] = []
security_reproduction: CheckResult | None = None
verifier: VerifierResult | None = None
workspace_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
class FileManifestEntry(ContractModel):
path: str
operation: Literal["add", "modify", "delete"]
original_sha256: str | None = None
resulting_sha256: str | None = None
artifact_ref: str | None = None
_relative_path = field_validator("path")(_validate_relative_path)
class FixPreparationResultV1(ContractModel):
version: Literal["1"] = "1"
state: PreparationState
stop_reason: str
source_identity: SourceIdentity | None
candidate: FixCandidateV1
candidate_digest: str
final_file_manifest: list[FileManifestEntry] = []
artifact_ref: str | None = None
changed_files: list[str] = []
diff_summary: str = ""
checks: list[CheckResult] = []
security_reproduction: CheckResult | None = None
verifier: VerifierResult | None = None
attempt_history: list[FixPreparationAttempt] = []
gaps: list[str] = []
blocker: PreparationBlocker | None = None
attempts: int = Field(default=0, ge=0)
elapsed_seconds: float = Field(default=0, ge=0)
cost_usd: float | None = Field(default=None, ge=0)
def candidate_from_legacy_report(
report: Mapping[str, object],
*,
source_identity: SourceIdentity | None = None,
) -> FixCandidateV1 | None:
raw_locations = report.get("code_locations")
if not isinstance(raw_locations, list):
return None
location_values = cast("list[object]", raw_locations)
locations: list[CandidateLocation] = []
edits: list[FixEdit] = []
for raw_value in location_values:
if not isinstance(raw_value, dict):
continue
raw = cast("dict[str, object]", raw_value)
file_path = raw.get("file")
start_line = raw.get("start_line")
end_line = raw.get("end_line")
if not isinstance(file_path, str) or type(start_line) is not int:
continue
if type(end_line) is not int:
end_line = start_line
common = {
"file": file_path,
"start_line": start_line,
"end_line": end_line,
"snippet": raw.get("snippet") if isinstance(raw.get("snippet"), str) else None,
"label": raw.get("label") if isinstance(raw.get("label"), str) else None,
}
try:
locations.append(CandidateLocation.model_validate(common))
before = raw.get("fix_before")
after = raw.get("fix_after")
if isinstance(before, str) and isinstance(after, str):
edits.append(FixEdit.model_validate({**common, "before": before, "after": after}))
except ValueError:
continue
if not locations:
return None
invariant = str(
report.get("remediation_steps") or report.get("technical_analysis") or ""
).strip()
if not invariant:
invariant = "Resolve the reported security finding without changing legitimate behavior."
reported_checks: list[ReportedCheck] = []
verification = str(report.get("fix_verification") or "").strip()
if verification:
reported_checks.append(
ReportedCheck(name="reporting-agent verification", result=verification, executed=False)
)
return FixCandidateV1(
source_identity=source_identity,
security_invariant=invariant,
finding_locations=locations,
draft_edits=edits,
reproduction=ReproductionSpec(
instructions=str(report.get("poc_description") or report.get("evidence") or invariant)
),
reported_checks=reported_checks,
known_gaps=["The reporting-agent verification is not independent."],
)
-109
View File
@@ -1,109 +0,0 @@
"""Deterministic source anchoring for finding locations and draft edits."""
from __future__ import annotations
import hashlib
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING
from strix.fix.contracts import CandidateLocation, FixCandidateV1, FixEdit
if TYPE_CHECKING:
from pathlib import Path
class AnchorStatus(StrEnum):
UNIQUE = "unique"
MISSING = "missing"
AMBIGUOUS = "ambiguous"
STALE = "stale"
@dataclass(frozen=True, slots=True)
class AnchorResult:
status: AnchorStatus
location: CandidateLocation | FixEdit
matches: tuple[int, ...] = ()
def sha256_text(value: str) -> str:
return hashlib.sha256(value.encode()).hexdigest()
def _find_blocks(content: str, block: str) -> tuple[int, ...]:
source_lines = content.splitlines()
block_lines = block.splitlines()
if not block_lines:
return ()
width = len(block_lines)
return tuple(
index + 1
for index in range(len(source_lines) - width + 1)
if source_lines[index : index + width] == block_lines
)
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:
content = _read_anchor_source(root, location.file)
if content is None:
return AnchorResult(AnchorStatus.MISSING, location)
block = location.before if isinstance(location, FixEdit) else location.snippet
if not block:
return AnchorResult(AnchorStatus.UNIQUE, location, (location.start_line,))
matches = _find_blocks(content, block)
if not matches:
if (
isinstance(location, FixEdit)
and location.original_sha256
and sha256_text(content) != location.original_sha256
):
return AnchorResult(AnchorStatus.STALE, location)
return AnchorResult(AnchorStatus.MISSING, location)
if len(matches) > 1:
return AnchorResult(AnchorStatus.AMBIGUOUS, location, matches)
start_line = next(iter(matches))
end_line = start_line + len(block.splitlines()) - 1
anchored = location.model_copy(update={"start_line": start_line, "end_line": end_line})
return AnchorResult(AnchorStatus.UNIQUE, anchored, matches)
def anchor_candidate(
root: Path,
candidate: FixCandidateV1,
) -> tuple[FixCandidateV1, list[AnchorResult]]:
results = [
*(anchor_location(root, location) for location in candidate.finding_locations),
*(anchor_location(root, edit) for edit in candidate.draft_edits),
]
finding_count = len(candidate.finding_locations)
anchored_locations = [result.location for result in results[:finding_count]]
anchored_edits = [result.location for result in results[finding_count:]]
return (
candidate.model_copy(
update={
"finding_locations": anchored_locations,
"draft_edits": anchored_edits,
}
),
results,
)
-755
View File
@@ -1,755 +0,0 @@
"""Repository fix preparation engine."""
from __future__ import annotations
import asyncio
import functools
import hashlib
import json
import os
import subprocess
import time
from collections.abc import Awaitable, Callable, Iterable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal
from strix.fix.contracts import (
BlockerKind,
CheckResult,
CheckStatus,
CommandSpec,
FileManifestEntry,
FixCandidateV1,
FixPreparationAttempt,
FixPreparationRequestV1,
FixPreparationResultV1,
PreparationBlocker,
PreparationState,
RepairOutcome,
RepairStatus,
VerificationDecision,
VerificationTarget,
VerifierResult,
)
from strix.fix.locations import AnchorStatus, anchor_location
class PreparationCancelledError(RuntimeError):
pass
CommandRunner = Callable[[Path, CommandSpec], Awaitable[CheckResult]]
ManifestBuilder = Callable[[Path], Awaitable[tuple[list[FileManifestEntry], str, str | None]]]
CancellationCheck = Callable[[], bool]
@dataclass(slots=True)
class PreparationPolicy:
max_repair_attempts: int = 2
timeout_seconds: int = 1800
max_output_chars: int = 20000
def __post_init__(self) -> None:
if not 1 <= self.max_repair_attempts <= 2:
raise ValueError("max_repair_attempts must be between 1 and 2")
@dataclass(slots=True)
class PreparationContext:
request: FixPreparationRequestV1
workspace: Path
candidate: FixCandidateV1
attempt: int = 0
feedback: list[FixPreparationAttempt] = field(default_factory=list)
RepairAgent = Callable[
[PreparationContext, list[CheckResult]],
Awaitable[RepairOutcome | None],
]
IndependentVerifier = Callable[
[PreparationContext, list[CheckResult]],
Awaitable[VerifierResult],
]
SourceVerifier = Callable[[PreparationContext], Awaitable[bool]]
_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():
return CheckResult(
name=command.name,
argv=command.argv,
status=CheckStatus.UNAVAILABLE,
duration_seconds=time.monotonic() - started,
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(
*argv,
cwd=cwd,
env=_command_environment(credentials_allowed),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
)
output, _ = await asyncio.wait_for(process.communicate(), command.timeout_seconds)
except (FileNotFoundError, PermissionError) as exc:
return CheckResult(
name=command.name,
argv=command.argv,
status=CheckStatus.UNAVAILABLE,
duration_seconds=time.monotonic() - started,
output=str(exc),
required=command.required,
)
except TimeoutError:
if process is not None:
process.kill()
await process.wait()
return CheckResult(
name=command.name,
argv=command.argv,
status=CheckStatus.FAILED,
duration_seconds=time.monotonic() - started,
output=f"Timed out after {command.timeout_seconds} seconds.",
required=command.required,
)
return CheckResult(
name=command.name,
argv=command.argv,
status=CheckStatus.PASSED if process.returncode == 0 else CheckStatus.FAILED,
exit_code=process.returncode,
duration_seconds=time.monotonic() - started,
output=output.decode(errors="replace")[-20000:],
required=command.required,
)
async def build_git_manifest(
workspace: Path,
) -> tuple[list[FileManifestEntry], str, str | None]:
process = await asyncio.create_subprocess_exec(
"git",
"status",
"--porcelain=v1",
"--untracked-files=all",
"-z",
cwd=workspace,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
output, error = await process.communicate()
if process.returncode != 0:
raise RuntimeError(error.decode(errors="replace"))
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]
status = record[:2]
path_text = record[3:]
if "R" in status or "C" in status:
index += 1
path = workspace / path_text
changed_files.append(path_text)
operation: Literal["add", "modify", "delete"]
if status == "??" or "A" in status:
operation = "add"
elif "D" in status:
operation = "delete"
else:
operation = "modify"
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(
"git",
"show",
f"HEAD:{path_text}",
cwd=workspace,
stdout=asyncio.subprocess.PIPE,
stderr=subprocess.DEVNULL,
)
original_bytes, _ = await original_process.communicate()
if original_process.returncode == 0:
original = hashlib.sha256(original_bytes).hexdigest()
entries.append(
FileManifestEntry(
path=path_text,
operation=operation,
original_sha256=original,
resulting_sha256=resulting,
)
)
index += 1
summary_process = await asyncio.create_subprocess_exec(
"git",
"diff",
"--stat",
"--",
cwd=workspace,
stdout=asyncio.subprocess.PIPE,
stderr=subprocess.DEVNULL,
)
summary, _ = await summary_process.communicate()
return entries, summary.decode(errors="replace").strip(), None
async def _workspace_digest(workspace: Path) -> str:
manifest, _, _ = await build_git_manifest(workspace)
payload = json.dumps(
[entry.model_dump(mode="json") for entry in manifest],
sort_keys=True,
separators=(",", ":"),
).encode()
return hashlib.sha256(payload).hexdigest()
async def _verify_source(context: PreparationContext) -> bool:
identity = context.candidate.source_identity
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",
"status",
"--porcelain=v1",
"-z",
cwd=context.workspace,
stdout=asyncio.subprocess.PIPE,
stderr=subprocess.DEVNULL,
)
status_output, _ = await status_process.communicate()
return status_process.returncode == 0 and not status_output.strip(b"\x00")
def _result(
context: PreparationContext,
*,
state: PreparationState,
reason: str,
started: float,
checks: list[CheckResult] | None = None,
reproduction: CheckResult | None = None,
verifier: VerifierResult | None = None,
gaps: list[str] | None = None,
manifest: list[FileManifestEntry] | None = None,
diff_summary: str = "",
artifact_ref: str | None = None,
attempt_history: list[FixPreparationAttempt] | None = None,
blocker: PreparationBlocker | None = None,
) -> FixPreparationResultV1:
return FixPreparationResultV1(
state=state,
stop_reason=reason,
source_identity=context.candidate.source_identity,
candidate=context.candidate,
candidate_digest=context.candidate.digest(),
final_file_manifest=manifest or [],
artifact_ref=artifact_ref,
changed_files=[entry.path for entry in manifest or []],
diff_summary=diff_summary,
checks=checks or [],
security_reproduction=reproduction,
verifier=verifier,
attempt_history=attempt_history or [],
gaps=gaps or [],
blocker=blocker,
attempts=context.attempt,
elapsed_seconds=time.monotonic() - started,
)
def _repair_outcome(value: RepairOutcome | None) -> RepairOutcome:
if value is not None:
return value
return RepairOutcome(
status=RepairStatus.COMPLETE,
summary="The repair implementation returned control for independent evaluation.",
)
def _required_checks_pass(checks: list[CheckResult]) -> bool:
required = [result for result in checks if result.required]
return bool(required) and all(result.status is CheckStatus.PASSED for result in required)
def _verification_passes(
checks: list[CheckResult],
verifier: VerifierResult,
) -> bool:
return (
_required_checks_pass(checks)
and verifier.decision is VerificationDecision.VERIFIED
and verifier.security_invariant_closed
and verifier.reproduction_executed
and any(
result.target is VerificationTarget.PATCHED and result.status is CheckStatus.PASSED
for result in verifier.security_tests
)
)
def _verification_gaps(
repair_outcome: RepairOutcome,
checks: list[CheckResult],
verifier: VerifierResult,
) -> list[str]:
gaps = list(repair_outcome.gaps)
required = [result for result in checks if result.required]
if not required:
gaps.append("No required repository check was configured.")
gaps.extend(
f"{result.name}: required check {result.status}"
for result in required
if result.status is not CheckStatus.PASSED
)
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 not verifier.security_tests:
gaps.append("The independent verifier did not execute a security test.")
elif not any(
result.target is VerificationTarget.PATCHED and result.status is CheckStatus.PASSED
for result in verifier.security_tests
):
gaps.append("The independent verifier did not pass a security test on the patch.")
if not verifier.reproduction_executed:
gaps.append("The independent verifier did not execute the security reproduction.")
if not verifier.security_invariant_closed:
gaps.append("The independent verifier did not prove the security invariant.")
gaps.extend(verifier.gaps)
return list(dict.fromkeys(gaps))
def _matches_baseline_failure(result: CheckResult) -> bool:
if result.baseline_status is not CheckStatus.FAILED:
return False
candidate_output = " ".join(result.output.split())
baseline_output = " ".join((result.baseline_output or "").split())
return bool(candidate_output and candidate_output == baseline_output)
async def prepare_fix( # noqa: PLR0915
request: FixPreparationRequestV1,
workspace: Path,
*,
repair: RepairAgent,
verify: IndependentVerifier,
command_runner: CommandRunner = run_command,
manifest_builder: ManifestBuilder = build_git_manifest,
source_verifier: SourceVerifier = _verify_source,
cancelled: CancellationCheck = lambda: False,
policy: PreparationPolicy | None = None,
) -> FixPreparationResultV1:
started = time.monotonic()
resolved_policy = policy or PreparationPolicy(
max_repair_attempts=request.max_repair_attempts,
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, PLR0915
if cancelled():
raise PreparationCancelledError
if not await source_verifier(context):
blocker = PreparationBlocker(
kind=BlockerKind.SOURCE,
summary="The repository no longer matches the finding source.",
user_action="Refresh the finding against the current repository revision.",
)
return _result(
context,
state=PreparationState.STALE,
reason=blocker.summary,
blocker=blocker,
started=started,
)
anchors = [
anchor_location(workspace, location) for location in context.candidate.finding_locations
]
if any(result.status is not AnchorStatus.UNIQUE for result in anchors):
details = [
f"{result.location.file}: {result.status}"
for result in anchors
if result.status is not AnchorStatus.UNIQUE
]
blocker = PreparationBlocker(
kind=BlockerKind.SOURCE,
summary="The reported finding locations could not be resolved uniquely.",
user_action="Refresh the finding or identify the affected source location.",
details=details,
)
return _result(
context,
state=PreparationState.BLOCKED,
reason=blocker.summary,
gaps=details,
blocker=blocker,
started=started,
)
context.candidate = context.candidate.model_copy(
update={"finding_locations": [result.location for result in anchors]}
)
checks: list[CheckResult] = []
verifier: VerifierResult | None = None
attempt_history: list[FixPreparationAttempt] = []
for attempt in range(1, resolved_policy.max_repair_attempts + 1):
context.attempt = attempt
if cancelled():
raise PreparationCancelledError
repair_outcome = _repair_outcome(await repair(context, checks))
if repair_outcome.status is RepairStatus.BLOCKED:
manifest, summary, artifact_ref = await manifest_builder(workspace)
blocker = repair_outcome.blocker or PreparationBlocker(
kind=BlockerKind.EXTERNAL_CONFIGURATION,
summary=repair_outcome.summary,
user_action=(
"Provide the missing repository, credential, or production "
"configuration prerequisite."
),
details=repair_outcome.gaps,
)
return _result(
context,
state=PreparationState.BLOCKED,
reason=blocker.summary,
gaps=repair_outcome.gaps,
manifest=manifest,
diff_summary=summary,
artifact_ref=artifact_ref,
attempt_history=attempt_history,
blocker=blocker,
started=started,
)
if repair_outcome.status is not RepairStatus.COMPLETE:
manifest, summary, artifact_ref = await manifest_builder(workspace)
return _result(
context,
state=PreparationState.FAILED,
reason=repair_outcome.summary,
gaps=repair_outcome.gaps,
manifest=manifest,
diff_summary=summary,
artifact_ref=artifact_ref,
attempt_history=attempt_history,
started=started,
)
checks = [await runner(workspace, check) for check in request.checks]
workspace_digest = await _workspace_digest(workspace)
attempt_record = FixPreparationAttempt(
attempt=attempt,
repair=repair_outcome,
checks=checks,
workspace_digest=workspace_digest,
)
attempt_history.append(attempt_record)
context.feedback = list(attempt_history)
required = [result for result in checks if result.required]
unavailable = [
result for result in required if result.status is CheckStatus.UNAVAILABLE
]
if not required or unavailable:
details = (
["No required repository quality check was configured."]
if not required
else [f"{result.name}: {result.output}" for result in unavailable]
)
blocker = PreparationBlocker(
kind=BlockerKind.ENVIRONMENT,
summary="The repository quality gate could not run.",
user_action=(
"Provide the missing tool or repository setup needed to run "
"the required checks."
),
details=details,
)
manifest, summary, artifact_ref = await manifest_builder(workspace)
return _result(
context,
state=PreparationState.BLOCKED,
reason=blocker.summary,
checks=checks,
gaps=blocker.details,
manifest=manifest,
diff_summary=summary,
artifact_ref=artifact_ref,
attempt_history=attempt_history,
blocker=blocker,
started=started,
)
failed = [result for result in required if result.status is CheckStatus.FAILED]
baseline_failures = [result for result in failed if _matches_baseline_failure(result)]
if baseline_failures:
blocker = PreparationBlocker(
kind=BlockerKind.REPOSITORY_BASELINE,
summary="Required checks also fail on the unchanged repository.",
user_action=(
"Repair the repository baseline or identify authoritative "
"replacement checks."
),
details=[result.name for result in baseline_failures],
)
manifest, summary, artifact_ref = await manifest_builder(workspace)
return _result(
context,
state=PreparationState.BLOCKED,
reason=blocker.summary,
checks=checks,
gaps=blocker.details,
manifest=manifest,
diff_summary=summary,
artifact_ref=artifact_ref,
attempt_history=attempt_history,
blocker=blocker,
started=started,
)
if failed:
if attempt < resolved_policy.max_repair_attempts:
continue
manifest, summary, artifact_ref = await manifest_builder(workspace)
return _result(
context,
state=PreparationState.FAILED,
reason="The repair did not pass the repository quality gate.",
checks=checks,
gaps=[f"{result.name}: {result.output}" for result in failed],
manifest=manifest,
diff_summary=summary,
artifact_ref=artifact_ref,
attempt_history=attempt_history,
started=started,
)
verifier = await verify(context, checks)
attempt_record.verifier = verifier
attempt_record.security_reproduction = next(
(
result
for result in verifier.security_tests
if result.target is VerificationTarget.PATCHED
),
None,
)
context.feedback = list(attempt_history)
gaps = _verification_gaps(repair_outcome, checks, verifier)
if verifier.blocker is not None:
manifest, summary, artifact_ref = await manifest_builder(workspace)
return _result(
context,
state=PreparationState.BLOCKED,
reason=verifier.blocker.summary,
checks=checks,
reproduction=attempt_record.security_reproduction,
verifier=verifier,
gaps=gaps,
manifest=manifest,
diff_summary=summary,
artifact_ref=artifact_ref,
attempt_history=attempt_history,
blocker=verifier.blocker,
started=started,
)
if _verification_passes(checks, verifier):
manifest, summary, artifact_ref = await manifest_builder(workspace)
if not manifest:
return _result(
context,
state=PreparationState.FAILED,
reason="The repair did not change repository source.",
checks=checks,
verifier=verifier,
gaps=["No prepared source change was produced."],
attempt_history=attempt_history,
started=started,
)
return _result(
context,
state=PreparationState.READY,
reason="The fix passed repository checks and security verification.",
checks=checks,
reproduction=attempt_record.security_reproduction,
verifier=verifier,
manifest=manifest,
diff_summary=summary,
artifact_ref=artifact_ref,
attempt_history=attempt_history,
started=started,
)
if verifier.repairable and attempt < resolved_policy.max_repair_attempts:
continue
manifest, summary, artifact_ref = await manifest_builder(workspace)
return _result(
context,
state=PreparationState.FAILED,
reason="The independent security verifier did not approve the repair.",
checks=checks,
reproduction=attempt_record.security_reproduction,
verifier=verifier,
gaps=gaps,
manifest=manifest,
diff_summary=summary,
artifact_ref=artifact_ref,
attempt_history=attempt_history,
started=started,
)
manifest, summary, artifact_ref = await manifest_builder(workspace)
return _result(
context,
state=PreparationState.FAILED,
reason="The repair did not satisfy the required gates.",
checks=checks,
verifier=verifier,
manifest=manifest,
diff_summary=summary,
artifact_ref=artifact_ref,
attempt_history=attempt_history,
started=started,
)
try:
async with asyncio.timeout(resolved_policy.timeout_seconds):
return await execute()
except PreparationCancelledError:
return _result(
context,
state=PreparationState.FAILED,
reason="Fix preparation was cancelled.",
attempt_history=context.feedback,
started=started,
)
except TimeoutError:
return _result(
context,
state=PreparationState.FAILED,
reason="Fix preparation exceeded its time limit.",
attempt_history=context.feedback,
started=started,
)
+1 -3
View File
@@ -209,9 +209,7 @@ def format_vulnerability_report(report: dict[str, Any]) -> Text: # noqa: PLR091
text.append("\n ")
text.append(loc["snippet"], style="dim")
if loc.get("fix_before") or loc.get("fix_after"):
preparation = report.get("fix_preparation")
prepared = isinstance(preparation, dict) and preparation.get("state") == "ready"
text.append("\n Prepared fix:" if prepared else "\n Draft fix candidate:")
text.append("\n Fix:")
if loc.get("fix_before"):
text.append("\n - ", style="dim")
text.append(loc["fix_before"], style="dim")
+30 -33
View File
@@ -28,10 +28,9 @@ 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 whose fix candidate completed verified preparation
(``fix_preparation.state == "ready"`` with a ``candidate_digest``
matching the candidate the result carries) are emitted as SARIF
``fixes`` so code-scanning can render a one-click suggested change.
* 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.
* 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.
@@ -57,10 +56,6 @@ 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__)
@@ -588,44 +583,46 @@ def _result_properties(
def _build_fixes(report: dict[str, Any]) -> list[dict[str, Any]] | None:
"""Build SARIF ``fixes`` from a verified prepared finding.
"""Build SARIF ``fixes`` from a finding's code-location fix pairs.
SARIF consumers can apply ``fixes`` automatically. Strix emits them only
when preparation recorded a ``ready`` result. The result carries the
candidate preparation verified — after anchoring corrected its reported
lines — and its ``candidate_digest`` must match that candidate, so the
emitted replacements are exactly what preparation verified, never a
stale or diverged draft.
Strix findings carry the suggested change inline on each code
location as ``fix_before`` + ``fix_after``. We map every location
that has both (and a safe repo-relative URI + start line) into a
SARIF ``artifactChange``, replacing the finding's region with the
fixed text. Returns None when no location carries a usable fix pair.
"""
preparation = report.get("fix_preparation")
if not isinstance(preparation, dict) or preparation.get("state") != "ready":
return None
raw_candidate = preparation.get("candidate") or 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():
raw_locations = report.get("code_locations")
if not isinstance(raw_locations, list):
return None
artifact_changes: list[dict[str, Any]] = []
for edit in candidate.draft_edits:
uri = _sarif_uri(edit.file)
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)
if uri is None:
continue
deleted_region: dict[str, Any] = {
"startLine": edit.start_line,
"endLine": edit.end_line,
}
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
artifact_changes.append(
{
"artifactLocation": {"uri": uri},
"replacements": [
{
"deletedRegion": deleted_region,
"insertedContent": {"text": edit.after},
"insertedContent": {"text": fix_after},
}
],
}
-12
View File
@@ -93,8 +93,6 @@ UPDATABLE_REPORT_FIELDS = frozenset(
"http_exchange_ids",
"fix_verification",
"fix_pr_body",
"fix_candidate",
"fix_preparation",
}
)
@@ -359,8 +357,6 @@ class ReportState:
fix_pr_body: str | None = None,
finding_class: str | None = None,
dependency_metadata: dict[str, str] | None = None,
fix_candidate: dict[str, Any] | None = None,
fix_preparation: dict[str, Any] | None = None,
agent_id: str | None = None,
agent_name: str | None = None,
) -> str:
@@ -424,10 +420,6 @@ class ReportState:
report["finding_class"] = (finding_class or "dynamic").strip().lower()
if dependency_metadata:
report["dependency_metadata"] = dependency_metadata
if fix_candidate:
report["fix_candidate"] = fix_candidate
if fix_preparation:
report["fix_preparation"] = fix_preparation
if agent_id:
report["agent_id"] = agent_id
if agent_name:
@@ -921,9 +913,6 @@ class ReportState:
context["ref"] = f"refs/heads/{branch}"
return context
def get_repository_context(self) -> dict[str, Any] | None:
return self._derive_repository_context()
def _sync_llm_usage_record(self) -> None:
self.run_record["llm_usage"] = self._build_llm_usage_record()
@@ -944,7 +933,6 @@ def openrouter_stream_cost(usage: Any) -> float | None:
"""
if not isinstance(usage, dict):
return None
total = 0.0
cost = usage.get("cost")
if isinstance(cost, int | float) and cost > 0:
+2 -14
View File
@@ -332,10 +332,7 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
lines.extend(f" {ln}" for ln in snippet.splitlines())
lines.append(f" {fence}")
if loc.get("fix_before") or loc.get("fix_after"):
preparation = report.get("fix_preparation")
prepared = isinstance(preparation, dict) and preparation.get("state") == "ready"
label = "Prepared Fix" if prepared else "Draft Fix Candidate"
lines.append(f"\n **{label}:**")
lines.append("\n **Suggested Fix:**")
lines.append("```diff")
if loc.get("fix_before"):
lines.extend(f"- {ln}" for ln in str(loc["fix_before"]).splitlines())
@@ -350,19 +347,10 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
lines.append("")
if report.get("fix_verification"):
lines.append("## Reported Candidate Checks\n")
lines.append("## Fix Verification\n")
lines.append(str(report["fix_verification"]))
lines.append("")
if isinstance(report.get("fix_preparation"), dict):
preparation = report["fix_preparation"]
lines.append("## Fix Preparation\n")
lines.append(f"State: `{preparation.get('state', 'unknown')}`")
if preparation.get("stop_reason"):
lines.append("")
lines.append(str(preparation["stop_reason"]))
lines.append("")
if report.get("assumptions"):
lines.append("## Assumptions\n")
lines.append(str(report["assumptions"]))
+2
View File
@@ -94,6 +94,8 @@ curl https://BUCKET.s3.amazonaws.com/
### IAM Privilege Escalation
**Resource control policies:** AWS Organizations RCPs impose an organization-level limit on supported resources, including access by external principals. Map resource-account RCPs as well as principal-account SCPs, identity/resource policies, session policies, and permission boundaries when assessing a cross-account or leaked-key path. RCPs do not grant access and do not apply uniformly to every service; an identity-policy allow alone does not describe the effective permission ([AWS RCP documentation](https://docs.aws.amazon.com/organizations/latest/userguide/orgs_manage_policies_rcps.html)).
Common escalation paths (verify with `aws iam simulate-principal-policy` when possible):
| Permission | Escalation |
+4
View File
@@ -32,6 +32,10 @@ Kubernetes clusters expose a large attack surface through their API server, kube
## Key Vulnerabilities
### Ingress-NGINX Admission
Inventory validating-webhook reachability from the pod network and the controller's service-account permissions. IngressNightmare illustrates controller execution without a Kubernetes account ([advisory](https://kubernetes.io/blog/2025/03/24/ingress-nginx-cve-2025-1974/)). Identify the controller implementation/image, then verify affected builds, backports, and maintenance status against current project or vendor notices; distinguish community ingress-nginx from other NGINX controllers and the Ingress API ([project notice](https://kubernetes.io/blog/2026/01/29/ingress-nginx-statement/)).
### RBAC Misconfigurations
- Wildcard verbs or resources in ClusterRole/Role bindings: `verbs: ["*"]`, `resources: ["*"]`
+6 -1
View File
@@ -57,6 +57,11 @@ Map endpoints, authentication classes, and permission classes per route.
## Key Vulnerabilities
### Request and Spatial Input Handling
- Under ASGI, compare underscore/hyphen forms of proxy-injected identity headers through `ASGIRequest` normalization into `request.META`. Resolve the installed branch, vendor backports, and support status from package metadata and current Django release/advisory information ([header-collision advisory](https://www.djangoproject.com/weblog/2026/apr/07/security-releases/)).
- For GeoDjango, trace attacker-controlled spatial lookup strings/dictionaries into `GDALRaster`: driver behavior can turn them into server-side requests or file writes. Include custom filter APIs and admin changelist filtering; direct model-field assignment is a separate input path ([spatial lookup advisory](https://www.djangoproject.com/weblog/2026/aug/04/security-releases/)).
### Authentication & Authorization
**Permission Class Gaps**
@@ -207,7 +212,7 @@ Static analysis is the fastest way to reach the sinks above in white-box scope.
- **pip-audit** (PyPA) — dependency CVE scanner for known-vuln Django/DRF/simplejwt versions: `pipx install pip-audit && pip-audit -r requirements.txt`
- **ast-grep** (preinstalled) — quick structural grep for risky calls without a full SAST run: `ast-grep run -p 'mark_safe($X)' -l python`
For the `SECRET_KEY` → signed-cookie/reset-token forgery path noted under Session Issues, Django's own `django.core.signing` is the "tool": with a leaked key you can mint valid `signing.dumps()` values (session cookies, password-reset tokens, and `PickleSerializer`-backed session RCE).
For the `SECRET_KEY` → signed-cookie forgery path noted under Session Issues, Django's own `django.core.signing` is the "tool": with a leaked key you can mint valid `signing.dumps()` values using the consumer's serializer and signing salt. Inspect `SESSION_SERIALIZER` and the installed implementation: the session-RCE path requires a reachable pickle-backed serializer; JSON sessions do not provide it. Password-reset tokens instead use `PasswordResetTokenGenerator`, with user state and timestamp in the digest, rather than the generic `signing.dumps()` format ([Django 5.0 removals](https://docs.djangoproject.com/en/5.0/releases/5.0/#features-removed-in-5-0), [token implementation](https://github.com/django/django/blob/stable/5.2.x/django/contrib/auth/tokens.py)).
## Summary
+6
View File
@@ -58,6 +58,12 @@ For each route, identify:
## Key Vulnerabilities
### Starlette Request Handling
Resolve Starlette's version independently of FastAPI and identify the ASGI server/front proxy. Check current upstream advisories and any vendor backports for the installed build before treating a parser issue as applicable. Compare middleware authorization against the actual routed path: malformed Host values can alter reconstructed `request.url` without changing routing ([URL parsing advisory](https://github.com/Kludex/starlette/security/advisories/GHSA-86qp-5c8j-p5mr)).
For `request.form()`, test URL-encoded and multipart limits separately; a limit enforced on one parser may not constrain the other. Check whether crossing an upload's memory-to-disk spool threshold blocks the event loop ([form limits](https://github.com/Kludex/starlette/security/advisories/GHSA-82w8-qh3p-5jfq), [file spooling](https://github.com/Kludex/starlette/security/advisories/GHSA-2c2j-9gv5-cj73)).
### Authentication & Authorization
**Dependency Injection Gaps**
+7
View File
@@ -78,6 +78,13 @@ For each controller and method, identify:
## Key Vulnerabilities
### Adapter and Schema Version Boundaries
- **Express adapter:** resolve the installed Nest/Express versions and effective query-parser setting from configuration or a harmless nested-query probe. Apply `qs`/nested-operator attacks only when the parser constructs nested objects. Check the matching router documentation and test both root and child paths against authentication middleware ([migration reference](https://nestjs.io/tutorials/what-s-new-in-express-5-eeb51579)).
- **Fastify adapter:** inspect the installed version's schema requirements and any custom validator configuration. Verify whether the route schema is actually enforced before relying on a DTO or shorthand-schema declaration ([migration reference](https://fastify.dev/docs/v5.0.x/Guides/Migration-Guide-V5/)).
- **Standard Schema, where supported:** `@Body({ schema })`, `@Query({ schema })`, `@Param(..., { schema })`, and `@RawBody({ schema })` attach metadata; enforcement requires `StandardSchemaValidationPipe`. Probe schema-decorated routes for accepted invalid values when the pipe is absent or transport-local. Outgoing schema enforcement uses `StandardSchemaSerializerInterceptor`; check actual output rather than relying only on class-transformer decorators.
- **GraphQL transport:** inspect installed packages, enabled IDE, and negotiated WebSocket subprotocol. Verify supported/deprecated transports against the matching Nest documentation, then recheck connection and per-operation authentication on the deployed transport ([migration guide](https://docs.nestjs.com/migration-guide)).
### Guard Bypass
**Decorator Stack Gaps**
+10 -3
View File
@@ -12,7 +12,7 @@ Security testing for Next.js applications. Focus on authorization drift across r
**Routers**
- App Router (`app/`) and Pages Router (`pages/`) often coexist
- Route Handlers (`app/api/**`) and API routes (`pages/api/**`)
- Middleware: `middleware.ts` at project root
- Middleware/proxy: inspect `middleware.ts`, `proxy.ts`, installed Next.js metadata, and runtime configuration. Check the matching framework docs for supported filenames, Node/Edge behavior, and feature status before selecting runtime-specific probes ([migration guide](https://nextjs.org/docs/app/guides/upgrading/version-16)).
**Runtimes**
- Node.js (full API access)
@@ -43,6 +43,8 @@ Security testing for Next.js applications. Focus on authorization drift across r
## Reconnaissance
At each assessment, establish the deployed build from package metadata, lockfiles, or runtime evidence. Verify release/support status and applicable fixes through current official docs, advisories, or upstream source; linked advisories are starting points, not a complete or permanently current list. Recheck when the target build or proposed remediation changes, and mark status unverified if evidence is unavailable.
**Route Discovery**
```javascript
@@ -83,10 +85,15 @@ Inspect Network tab for POST requests with `Next-Action` header. Extract action
## Key Vulnerabilities
### RSC and Image Processing
- **React2Shell (CVE-2025-55182):** App Router applications can expose the vulnerable RSC decoder without explicit Server Actions. Identify the bundled `react-server-dom-*` implementation and trace requests into decoding; client-only React and Pages-only apps have different exposure. Check the framework's patched branch, including source-disclosure and DoS fixes ([RCE advisory](https://react.dev/blog/2025/12/03/critical-security-vulnerability-in-react-server-components), [RSC advisories](https://react.dev/blog/2025/12/11/denial-of-service-and-source-code-exposure-in-react-server-components)).
- Treat `/_next/image` optimization and Node `next/og` `ImageResponse` as separate input paths. Trace attacker-controlled image bytes into native decoders, and SVG content/attributes/styles into image construction. URL allowlisting does not make image contents trusted; match the decoder, runtime, and Next.js version to the relevant advisory ([optimizer](https://nextjs.org/blog/august-2026-security-release), [ImageResponse](https://github.com/vercel/next.js/security/advisories/GHSA-vcvr-r3jv-pc5j)).
### Middleware Bypass
**Known Techniques**
- `x-middleware-subrequest` header crafting (CVE-class bypass)
- `x-middleware-subrequest` header crafting (CVE-2025-29927): middleware-only authorization can be skipped on affected deployments. Resolve the installed branch and hosting protections from the advisory and provider configuration, then check whether external headers reach the origin and the destination enforces authorization independently ([advisory](https://github.com/vercel/next.js/security/advisories/GHSA-f82v-jwr5-mffw)).
- `x-nextjs-data` probing
- Look for 307 + `x-middleware-rewrite`/`x-nextjs-redirect` headers
@@ -118,7 +125,7 @@ Middleware checks first value, handler uses last or array.
**Cache Boundary Failures**
- User-bound data cached without identity keys (ETag/Set-Cookie unaware)
- Personalized content served from shared cache/CDN
- Missing `no-store` on sensitive fetches
- Missing `no-store` on sensitive fetches matters only when they enter a shared cache. Determine `fetch` and GET Route Handler defaults for the deployed version from docs/configuration and observed responses. Where Cache Components are enabled, inspect `use cache` arguments/closed-over values, `cacheTag`, `cacheLife`, and invalidation for user/tenant separation ([caching documentation](https://nextjs.org/docs/app/getting-started/cache-components)).
**Flight Data Leakage**
+4
View File
@@ -134,6 +134,10 @@ Parser precedence varies; may bypass validation. Also test default argument valu
Send unexpected keys in input objects; backends may pass them to resolvers or downstream logic.
### OneOf Inputs
Discover `@oneOf` input types through the schema or `__Type.isOneOf`. They require exactly one non-null field, with every other field omitted. Test both inline literals and variables with zero fields, two selectors, and an extra null-valued selector; all must fail coercion. Then test authorization independently for each valid selector (ID, username, organization/email), since schema exclusivity does not ensure identical tenant checks in each resolver branch. Confirm the deployed implementation supports OneOf before treating acceptance as a specification violation ([GraphQL September 2025](https://spec.graphql.org/September2025/#sec-OneOf-Input-Objects)).
### Cursor Manipulation
Decode cursors (usually base64) to:
+6 -2
View File
@@ -77,7 +77,7 @@ com.app://callback (mobile custom scheme)
### State and Nonce
- Missing, predictable, or reusable `state` → CSRF on OAuth login (session fixation, account linking)
- Missing, predictable, or reusable `state` → test CSRF on OAuth login (session fixation, account linking); RFC 9700 also permits correctly bound PKCE, or OIDC `nonce`, to provide CSRF protection, so establish whether that protection survives a cross-session callback
- Missing `nonce` in OIDC → ID token injection/replay
- `state` not bound to client session or PKCE verifier
@@ -102,7 +102,7 @@ com.app://callback (mobile custom scheme)
### Scope and Token Issues
- Scope escalation: request `admin`/`offline_access`/`openid profile email` beyond app need; server grants all requested scopes
- Refresh token not rotated or reuse not detected → persistent access
- Public-client refresh tokens neither sender-constrained nor rotated with reuse detection → persistent access; a non-rotating token bound to the client's key is permitted by RFC 9700, so test replay without that key
- Access token accepted across services (missing audience/resource binding)
- Token introspection returns `active:true` without proper auth on introspection endpoint
@@ -113,6 +113,10 @@ com.app://callback (mobile custom scheme)
- Userinfo endpoint returns PII without matching access token scope
- `sub` collision across issuers if `iss` not validated
### OAuth Security BCP (RFC 9700)
Authorization servers must support PKCE, public clients must use it, and confidential clients are also recommended to use it. Test challenge stripping and verifier injection across both client types. Resource-owner-password grants must not be used; implicit access-token responses are discouraged because tokens can leak or be replayed. For mix-up defenses, bind the selected issuer and endpoints to the authorization transaction; validate an authorization-response `iss` when that defense is used, rather than merely checking the issuer of a later ID token ([RFC 9700](https://www.rfc-editor.org/rfc/rfc9700.html)).
## Advanced Techniques
**Referer Leakage**
@@ -100,6 +100,13 @@ certipy find -u <USER>@<DOMAIN> -p <PASS> -dc-ip <DC_IP> -vulnerable -stdout
- **ESC8** — NTLM relay to the CA web-enrollment endpoint (coerce a DC, relay to `/certsrv`) → DC certificate → DCSync.
- **ESC others** — ESC2/3 (any-purpose/enrollment-agent), ESC4 (writable template DACL → make it ESC1), ESC6 (`EDITF_ATTRIBUTESUBJECTALTNAME2` on the CA), ESC7 (CA officer rights), ESC9/10 (weak cert mapping), ESC11 (RPC relay), ESC13 (issuance-policy→group), ESC15 (app-policy on v1 templates). `certipy find -vulnerable` flags each.
**Strong certificate mapping:** establish the DC patch level, mapping policy, and vendor backports; consult Microsoft's current enforcement guidance before assuming Compatibility mode is available. Inspect template SID extensions, explicit mappings, and the principal selected at authentication. A requested privileged UPN alone does not establish a working ESC chain ([KB5014754](https://support.microsoft.com/en-us/servicing/os/windows-server/2022/05/kb5014754-certificate-based-authentication-changes-on-windows-domain-controllers)).
### Windows Server 2025 dMSA / BadSuccessor
- **CVE-2025-53779:** on unpatched Server 2025 DCs, control sufficient to create/modify a delegated Managed Service Account can establish a one-way migration link to another principal and obtain its authority/keys through the KDC. Enumerate dMSA creation rights, object ACLs, and migration attributes rather than limiting discovery to conventional service accounts and delegation flags.
- **Patched behavior:** the KDC requires mutual dMSA↔target linkage. Writing the dMSA-side attribute still succeeds, so LDAP write success is not proof of exploitation. Determine whether the tester also controls the target object's reciprocal link and whether the KDC actually issues the relevant ticket. Post-patch abuse is a different prerequisite chain from the original low-privilege OU-control escalation ([researchers' patch analysis](https://www.akamai.com/blog/security-research/badsuccessor-is-dead-analyzing-badsuccessor-patch)).
### NTLM Coercion & Relay
Force a privileged machine to authenticate to you, then relay that NTLM auth to a service that doesn't enforce signing/EPA (LDAP, AD CS, SMB).
+3 -1
View File
@@ -86,12 +86,14 @@ Authorization: Bearer <token_with_audience_A>
### Rules and Actions Abuse
Inventory which Rules, Hooks, and Actions actually execute in the tenant. Check the current Auth0 lifecycle notice and tenant capabilities for retirement or read-only restrictions. During migration, compare claim assignment, MFA, denial decisions, and account-linking checks across every connection; distinguish secret/configuration access from source-code modification ([lifecycle notice](https://auth0.com/docs/troubleshoot/product-lifecycle/deprecations-and-migrations#rules-and-hooks-deprecations)).
**Post-Login Rule/Action Injection**
- Rules that add claims based on unvalidated user metadata:
```javascript
user.app_metadata.role = 'admin' // if user can set app_metadata via signup/API
```
- `context.authorization` manipulation in Actions
- Rules use `context`; Actions receive `event` and enforce token/MFA/denial changes through `api` methods. Test migrated checks that only mutate event data: those mutations do not propagate to other Actions or substitute for the corresponding enforcement API ([migration behavior](https://auth0.com/docs/customize/actions/migrate/migrate-from-rules-to-actions)).
- Secrets in Rule code exposed to tenant admins or via Management API leak
**Signup / Registration Actions**
@@ -79,6 +79,13 @@ Unauthenticated view (and, with `public_mode`, delete) of the lowest-key snapsho
### Prometheus / Alertmanager — exposure is the vuln (no auth by default)
Prometheus and Alertmanager ship with **no authentication**; the docs explicitly say do not expose them. There is rarely a CVE — reachability itself is the finding, and the payoff is recon + credential leakage + pivoting (below).
## Plugin and MCP Boundaries
Resolve installed Grafana, plugin, and MCP server builds separately; check current upstream advisories and vendor backports before applying a listed attack path or recommending a release.
- Plugin archives are extracted before signature verification. Test chained symlinks and containment before trusting a signature failure to prevent filesystem writes; archive installation is the required trigger ([extraction advisory](https://grafana.com/security/security-advisories/cve-2026-15815/)).
- Inventory mcp-grafana separately from Grafana. Trace `grafana_api_request` and `X-Grafana-URL` into destination selection; preventing token forwarding alone does not prevent SSRF ([MCP advisory](https://grafana.com/security/security-advisories/cve-2026-19516/)).
## Pivoting: Observability → Deeper Compromise
This is the core value. Chain each exposure into something that matters. Always articulate the pivot in the finding, not just the exposed endpoint.
@@ -78,11 +78,11 @@ Record both forward and reverse reachability: attacker-controlled input to privi
## Optional Tool Routing
Use tools only when they match the deployed surface. Treat generated cases and scanner labels as leads until the application-side boundary is validated.
Use tools only when they match the deployed surface. Resolve compatible versions, runtime requirements, maintenance status, and advisories from registry metadata and current upstream documentation, then pin the exact reviewed versions for the run. Treat generated cases and scanner labels as leads until the application-side boundary is validated.
- **[Promptfoo](https://github.com/promptfoo/promptfoo)** — use for repeatable model/application trials, custom adversarial cases, graders, provider comparisons, and success-rate regression. Install the reviewed version locally with `npm install --save-dev --save-exact promptfoo@0.122.0`, then invoke `./node_modules/.bin/promptfoo redteam run`. Define explicit plugins, assertions, `numTests`, `maxConcurrency`, and `delay`; provider calls may transmit test data and incur cost. Its `owasp:llm` preset still uses the 2025 category mapping in version 0.122.0, so build or select tests from the 2026 matrix above and do not present the preset report as complete 2026 coverage.
- **[MCP Inspector](https://github.com/modelcontextprotocol/inspector)** — use for LLM01/LLM03 surface mapping when MCP servers are present. Install the reviewed version with `npm install --save-dev --save-exact @modelcontextprotocol/inspector@2.2.0`, then use `./node_modules/.bin/mcp-inspector --cli --config <reviewed-config> --server <name> --method tools/list` and the equivalent `resources/list` / `prompts/list` operations. Starting a stdio server executes that configured process, initialization/list handlers may have side effects, and `tools/call` can perform the real action; inspect the target and credentials before invoking it.
- **[ModelScan](https://github.com/protectai/modelscan)** — use for LLM04 static triage of supported H5, Pickle, and SavedModel artifacts before loading them, for example `uvx modelscan==0.8.8 -p <artifact>`. Run it as an untrusted-file parser in an isolated analysis environment. A clean result covers only the scanner's supported formats and signatures; it does not establish artifact provenance, integrity, or absence of behavioral backdoors.
- **[Promptfoo](https://github.com/promptfoo/promptfoo)** — use for repeatable model/application trials, custom adversarial cases, graders, provider comparisons, and success-rate regression. Install the reviewed version locally with `npm install --save-dev --save-exact promptfoo@<reviewed-version>`, then invoke `./node_modules/.bin/promptfoo redteam run`. Define explicit plugins, assertions, `numTests`, `maxConcurrency`, and `delay`; provider calls may transmit test data and incur cost. Inspect the installed `owasp:llm` preset's category mapping and actual tests before claiming coverage; add cases for gaps against the assessment's chosen taxonomy.
- **[MCP Inspector](https://github.com/modelcontextprotocol/inspector)** — use for LLM01/LLM03 surface mapping when MCP servers are present. Install the reviewed version with `npm install --save-dev --save-exact @modelcontextprotocol/inspector@<reviewed-version>`, then use `./node_modules/.bin/mcp-inspector --cli --config <reviewed-config> --server <name> --method tools/list` and the equivalent `resources/list` / `prompts/list` operations. Starting a stdio server executes that configured process, initialization/list handlers may have side effects, and `tools/call` can perform the real action; inspect the target and credentials before invoking it.
- **[ModelScan](https://github.com/protectai/modelscan)** — use for LLM04 static triage of supported H5, Pickle, and SavedModel artifacts before loading them, for example `uvx modelscan==<reviewed-version> -p <artifact>`. Run it as an untrusted-file parser in an isolated analysis environment. A clean result covers only the scanner's supported formats and signatures; it does not establish artifact provenance, integrity, or absence of behavioral backdoors.
## LLM01:2026 Prompt Injection
+8 -1
View File
@@ -35,7 +35,7 @@ Security testing for Supabase applications. Focus on mis-scoped Row Level Securi
- Functions: `https://<ref>.functions.supabase.co/`
**Headers**
- `apikey: <anon-or-service>` — identifies project
- `apikey: <publishable-or-secret-key>` — identifies the application component; legacy `anon` / `service_role` JWT keys can still coexist
- `Authorization: Bearer <JWT>` — binds user context
**Roles**
@@ -45,6 +45,13 @@ Security testing for Supabase applications. Focus on mis-scoped Row Level Securi
**Key Principle**
`auth.uid()` returns current user UUID from JWT. Policies must never trust client-supplied IDs over server context.
### API Keys and Signing Keys
- Opaque keys start with `sb_publishable_` or `sb_secret_`; JWT decoding will not classify them. Search bundles, server configuration, and responses for both formats. Publishable keys are intended for public clients; secret keys use `service_role` and bypass RLS. A browser 401 for a leaked secret key is not revocation: the browser restriction uses `User-Agent` and does not stop server-side use.
- Creating new keys does not disable legacy keys. After migration, test whether the old `anon` / `service_role` credential remains accepted; dashboard revocation is a separate action. Verify supported key formats and migration/deprecation status from project settings and current Supabase documentation ([API keys](https://supabase.com/docs/guides/getting-started/api-keys)).
- Edge Functions use user JWTs in `Authorization` and API keys in `apikey`. Verify the deployed platform's `verify_jwt` behavior with current docs and controlled requests, including API keys on either header; passing a platform check with a publishable key does not establish a signed-in user. Test whether the handler verifies user identity before using a privileged client. Conversely, `verify_jwt=false` with explicit secret-key or webhook verification is not necessarily public. Trace the configured `@supabase/server` auth mode or custom verification ([function authorization](https://supabase.com/docs/guides/functions/auth-headers)).
- Asymmetric signing keys are discoverable at `/auth/v1/.well-known/jwks.json`. Test custom API/Edge Function validators for issuer/algorithm binding and stale cached keys after rotation/revocation. Measure cache lifetimes from response headers, SDK configuration, and controlled rotation tests; check service-specific revocation behavior in current documentation rather than assuming platform and custom validators share a cache ([signing keys](https://supabase.com/docs/guides/auth/signing-keys)).
## High-Value Targets
- Tables with sensitive data (users, orders, payments, PII)
@@ -106,10 +106,16 @@ Classify each discovered integration by data read, data write, external communic
- For each tool, validate the same authorization and argument checks through every supported transport.
- Treat server-launched subprocess configuration, environment variables, and working directories as sensitive executable configuration.
- For HTTP/SSE transports, validate OAuth issuer, signature, expiry, audience/resource, tenant, and scope claims at the server boundary. Reject tokens minted for the wrong audience, and do not treat a session ID as identity.
- For downstream APIs, do not pass through the same bearer token unless the target explicitly authorizes that audience and principal. Separate upstream MCP authentication from downstream target authorization.
- MCP forbids token passthrough: reject access tokens not issued for the MCP server, and use a separate downstream authorization flow instead of forwarding a client's token to another API ([MCP authorization](https://modelcontextprotocol.io/specification/2026-07-28/basic/authorization)).
- For browser or loopback OAuth, review redirect URI, state/PKCE handling, localhost binding, and consent proxying. Treat metadata fetches and tool discovery on remote servers as SSRF-relevant surfaces.
- For stdio servers, the launch command and environment are already code execution. Discovery must not execute an unreviewed server binary or mutable package tag.
#### MCP Protocol Boundaries
- Identify the protocol revision and client/server SDK builds from configuration and wire traffic, then consult the matching specification and current SDK advisories. Determine whether the transport uses initialization/session IDs or independent requests with client metadata and discovery. Test authorization on every request; client-supplied identity/capabilities in `_meta` are not authenticated identity.
- Where multi-round tool requests use `input_required` and `inputResponses`, trace the retry/resume binding. Test whether a response from another user, tool invocation, or approval round can be substituted, and whether a retry repeats a consequential action ([MCP protocol specification](https://blog.modelcontextprotocol.io/posts/2026-07-28/)).
- For OAuth callbacks, bind the expected issuer from validated discovery to the PKCE transaction. A present `iss` must match exactly, even if metadata did not advertise support; an absent `iss` must be rejected when `authorization_response_iss_parameter_supported` is true. Test issuer changes between discovery, registration, and callback, including error callbacks. Bind client credentials to their authorization-server issuer. Require the MCP server's `resource` in authorization and token requests and validate the token's audience at the server ([authorization requirements](https://modelcontextprotocol.io/specification/2026-07-28/basic/authorization)).
### Executable Component Supply Chain
Every skill, plugin, MCP server, model adapter, package, and update channel is an executable or behavior-shaping dependency. Record:
@@ -158,7 +164,7 @@ npx @modelcontextprotocol/inspector@<reviewed-version> --cli \
--method tools/list --format json
```
- Current upstream requirements should be checked before pinning; as of August 12, 2026, MCP Inspector 2.1.0 requires Node.js `>=22.19.0`.
- Before choosing MCP Inspector, inspect package-registry metadata, engine requirements, release notes, and advisories. Select a compatible reviewed version and pin that exact version for the run; do not infer safety from a release tag.
- Prefer CLI/TUI and loopback binding over exposing the web UI.
- Preserve the generated API token; never disable authentication or bind the process-spawning backend to an external interface.
- Do not publish ports 6274/6277 or pass through the Docker socket/host devices.
@@ -173,7 +179,7 @@ npx @modelcontextprotocol/inspector@<reviewed-version> --cli \
npx promptfoo@<reviewed-version> eval
```
- Current upstream engine constraints should be checked before pinning; as of August 12, 2026, Promptfoo documents Node.js `^20.20.0` or `>=22.22.0`.
- Resolve Promptfoo's runtime requirements, supported features, and security status from registry metadata and current upstream docs before choosing and pinning a version.
- Use synthetic prompts/data and a dedicated test provider/project.
- Provider calls transmit data externally and can incur cost even when evaluation orchestration is local. Set request/concurrency and spending ceilings.
- Pin model, provider, prompt, tool schema, retrieval corpus revision, and evaluator versions.
@@ -104,6 +104,12 @@ Prove the strongest reliable capability first. If escalation requires a user ges
- Treat scriptless disclosure of a nonce or trusted URL as a primitive; prove a second controllable sink before claiming bypass.
- For response splitting, consider whether a same-origin endpoint can be turned into a script resource with a controlled body length or framing.
### Local Network and Loopback Access
Identify the browser build, enabled flags, permission state, and enterprise policy; consult its current Local Network Access documentation and verify actual behavior with controlled requests. Test DNS rebinding and browser-to-local-service chains in both permission-denied and permission-granted states. A successful server-side request does not establish browser reachability ([Chrome LNA](https://developer.chrome.com/blog/local-network-access)).
Discover which permission names and aliases the target supports, including `local-network`, `loopback-network`, and `local-network-access`. Record which destination class was permitted before claiming access to both LAN and localhost services ([permission reference](https://developer.chrome.com/release-notes/145#local_network_access_split_permissions)).
### JavaScript Gadget Discovery
- When direct calls are blocked, inspect implicit coercions (`toString`, `valueOf`, iterators, getters, proxies) and callbacks invoked by accessible library functions.
@@ -97,6 +97,12 @@ transfer-encoding: chunked
SMUGGLED
```
### CL.0, 0.CL, and Double Desync
In CL.0, the front end honors `Content-Length` while the back end ignores the body; in 0.CL the front end ignores it while the back end expects it. The latter commonly deadlocks until an early-response gadget responds without consuming the body and keeps the connection open. Test redirects and early errors for that specific connection behavior. A subsequent controlled request can expose the boundary shift; two successive desyncs can convert 0.CL into CL.0. A lone timeout or 400 does not establish that chain.
Include `Expect: 100-continue` handling and interim/final response sequencing in differential tests. Calculate offsets from what the back end receives, including proxy-added headers. Use controlled follow-up requests to establish which bytes are consumed and which response belongs to each request ([HTTP/1.1 must die](https://portswigger.net/research/http1-must-die), [0.CL walkthrough](https://portswigger.net/blog/http-1-1-must-die-conquering-the-0-cl-challenge)).
## Key Vulnerabilities
### Front-End Security Control Bypass
@@ -79,6 +79,8 @@ JNDI injection is not itself a serialization format. It becomes part of this wor
### Python Pickle
**Model checkpoints:** `torch.load(..., weights_only=True)` does not rule out parser/storage memory corruption. Resolve the installed loader build and check current upstream advisories/backports before trusting that flag. Trace untrusted checkpoints through the exact loader and inspect safe-global allowlists ([example advisory](https://github.com/pytorch/pytorch/security/advisories/GHSA-63cw-57p8-fm3p)).
Pickle executes arbitrary code during unpickling by design:
```python
import pickle, os, base64
@@ -101,13 +103,15 @@ When `yaml.load` used instead of `yaml.safe_load`.
- POP chains through framework classes (Laravel, Symfony, WordPress plugins)
**Phar Deserialization**
- Upload or reference `phar://` wrapper triggering metadata deserialization on file operations
- Trace `phar://` file operations and explicit `Phar::getMetadata()` / `PharFileInfo::getMetadata()` calls. Verify the installed PHP version's metadata-deserialization behavior and `allowed_classes` handling from source/docs; archive opening alone does not establish a deserialization sink ([migration reference](https://www.php.net/manual/en/migration80.incompatible.php#migration80.incompatible.phar)).
### .NET Deserialization
**BinaryFormatter / LosFormatter**
- Never safe on untrusted input; full RCE with known gadget chains (ysoserial.net)
Inspect the target framework and resolved serialization packages to establish whether `BinaryFormatter` executes, throws, or is restored through a compatibility package. Check the matching runtime documentation before choosing gadget chains; `LosFormatter` and other serializers are separate surfaces ([Microsoft guide](https://learn.microsoft.com/en-us/dotnet/standard/serialization/binaryformatter-migration-guide/)).
**Json.NET**
```json
{"$type":"System.Windows.Data.ObjectDataProvider, PresentationFramework", ...}
@@ -193,7 +197,7 @@ Payload generation is the practitioner's core tool here. The sandbox has `git`/`
| **ysoserial** (frohoff) | Java native | Gadget-chain payloads: `CommonsCollections1-7`, `Groovy1`, `Spring1/2`, and `URLDNS` for a safe no-exec DNS oracle. Needs a JRE. |
| **phpggc** (ambionics) | PHP `unserialize` / Phar | Framework POP chains (Laravel, Symfony, WordPress, Drupal, Monolog). Needs `php-cli`. |
| **ysoserial.net** | .NET `BinaryFormatter` / Json.NET | Windows/.NET gadget payloads. Needs .NET/mono — usually out of scope in a Linux sandbox. |
| **marshalsec** | Java Hessian/Burlap, Kryo, JSON, and JNDI reference tooling | Use only from a reviewed, pinned upstream commit when a non-native Java marshaller requires it. It has no stable release and intentionally bundles historical gadget dependencies; do not treat it as a globally installed default tool. |
| **marshalsec** | Java Hessian/Burlap, Kryo, JSON, and JNDI reference tooling | Use only from a reviewed, pinned upstream commit when a non-native Java marshaller requires it. Check upstream release status and bundled gadget dependencies before selecting a commit; do not treat it as a globally installed default tool. |
```
# Java: prove the sink with a no-exec DNS oracle BEFORE any RCE chain
@@ -91,7 +91,7 @@ Binary search the character space to minimize requests. Works on any string fiel
### `$where` JavaScript Injection
If `$where` operator is enabled (disabled by default in MongoDB 7.0+; MongoDB 4.4–6.x deprecated it but left `javascriptEnabled` defaulting to `true`), inject arbitrary server-side JavaScript:
If `$where` is enabled, inject server-side JavaScript. Inspect `security.javascriptEnabled` / `--noscripting`, managed-service restrictions, and the installed engine version. Verify operator availability, defaults, and supported functions against that build's documentation or controlled probes; deprecation alone does not mean execution is disabled ([MongoDB documentation](https://www.mongodb.com/docs/manual/reference/operator/query/where/)):
```json
{"$where": "function(){return this.role == 'admin'}"} // direct filter — returns matching documents
{"$where": "function(){return this.username == 'admin' && sleep(2000)}"} // timing oracle only — sleep() returns undefined (falsy), so no documents are returned; observe latency
@@ -148,6 +148,8 @@ Improper file path handling and dynamic inclusion enable sensitive file disclosu
- Verify symlink handling and path canonicalization prior to write
- Impact: overwrite config/templates or drop webshells into served directories
**Python extraction filters:** Determine the effective `filter` and `TarFile.extraction_filter` from the installed Python runtime, caller configuration, and matching documentation; verify patch/backport status against current advisories ([documentation](https://docs.python.org/3/library/tarfile.html#extraction-filters)). Filters do not eliminate link-handling bugs: test archive link ordering and post-extraction reads for outside-file disclosure or permission/timestamp changes, separately from content overwrite ([security advisory](https://mail.python.org/archives/list/[email protected]/thread/EFJWGAZJA56AKSBR2WHMHQZO7RRLZPRH/)).
### File Write to Execution
Characterize the write primitive before choosing a payload:
@@ -51,7 +51,7 @@ Prototype pollution corrupts shared object prototypes (`Object.prototype`, `Arra
**Common Sinks**
- `lodash.merge`, `lodash.defaultsDeep`, `deep-extend`, `merge-options`
- Express/query parsers accepting nested objects
- YAML `load()` (not `safeLoad`) with prototype keys
- YAML merge-key handling with prototype keys: identify the js-yaml version/schema and check applicable upstream advisories. Test whether `<<` merges alter the parsed result's prototype, then trace inherited values into a sensitive consumer; global `Object.prototype` modification is not required. Verify the installed API rather than assuming `load` versus `safeLoad` determines safety ([merge advisory](https://github.com/nodeca/js-yaml/security/advisories/GHSA-mh29-5h37-fv8m)).
- JSON.parse → merge into existing object without null prototype
**RCE Gadget Chains (Node.js)**
+3 -1
View File
@@ -76,6 +76,8 @@ When output isn't reflected:
### Jinja2 / Mako (Python)
**Jinja sandbox:** resolve the installed Jinja build and check current sandbox advisories before selecting indirect `str.format` or `|attr` gadgets. Establish template-source control and inspect custom filters; user data passed only as a variable is a different surface ([release notes](https://jinja.palletsprojects.com/en/stable/changes/)).
The classic Python class walk — every object exposes its method-resolution-order, which leads to `object`, which exposes every subclass loaded in the interpreter, which includes things like `subprocess.Popen`:
```jinja
@@ -137,7 +139,7 @@ Twig sandbox bypasses are version-specific. The canonical historical gadget (Twi
{{_self.env.registerUndefinedFilterCallback("system")}}{{_self.env.getFilter("id")}}
```
This was patched — in Twig 2.x / 3.x `_self` returns the template name as a string and no longer exposes `.env`. Modern bypasses depend on which extensions are loaded and the active sandbox policy; consult Twig's published security advisories for the current state and probe with the version-specific gadgets (filter/function abuse, reflection on `_context` in some configs).
In Twig 2.x / 3.x, `_self` returns the template name as a string and does not expose `.env`. Bypasses depend on which extensions are loaded and the active sandbox policy; consult Twig's published security advisories and probe with the version-specific gadgets (filter/function abuse, reflection on `_context` in some configs).
Smarty `{php}...{/php}` was the historical RCE primitive; deprecated in Smarty 3 and removed in 4. On modern Smarty, the surface is static-method invocation and template-object reflection — `{$smarty.template_object->smarty->...}` walks back to the Smarty engine, and direct static calls on whitelisted classes (e.g. `{Smarty_Internal_Write_File::writeFile(...)}` on misconfigured installs) reach the filesystem. Probe both before assuming Smarty is hardened.
@@ -49,13 +49,15 @@ Weak or default credentials remain one of the most prevalent and high-impact vul
### Weak Password Policies
- No minimum length or complexity requirements
- Insufficient minimum length for the authentication mode; NIST SP 800-63B-4 requires 15 characters for single-factor passwords and permits a minimum of eight when the password is only used as part of MFA
- Allowing common passwords: `password`, `123456`, `qwerty`, `admin`, `letmein`
- Not checking against breached password databases (Have I Been Pwned)
- Case-insensitive password storage
- No password history enforcement
- Forced periodic changes without compromise evidence, which encourage predictable password changes
- Excessively short maximum length (indicates plaintext or weak hashing)
NIST SP 800-63B-4 recommends allowing a maximum of at least 64 characters and requires screening against common/compromised passwords and checking the entire password without truncation. Missing character-class rules is not a weakness: the standard prohibits mandatory composition rules and periodic resets without evidence of compromise. Apply these requirements when that standard is the target's policy baseline ([NIST SP 800-63B-4](https://pages.nist.gov/800-63-4/sp800-63b.html#passwordver)).
### Default and Hardcoded Credentials
- Vendor defaults: `admin/admin`, `admin/password`, `root/root`, `guest/guest`
@@ -191,7 +193,7 @@ No password wordlists ship in the sandbox by default — download what you need
5. Test for password spraying (one password, many users) before targeted brute-force
6. Check for concurrent session limits; successful logins may kick out legitimate users
7. GraphQL batching can test multiple credentials in a single request, bypassing per-request limits
8. Document the password policy and recommend minimum standards (length, complexity, breach checking)
8. Document the password policy and recommend minimum standards (length appropriate to single-factor/MFA use, breach checking, and resistance to online guessing)
9. For web logins prefer `ffuf`; for other services use `nmap` NSE `*-brute` scripts or custom scripts with equivalent logic
10. Combine with MFA testing: weak passwords plus missing MFA is a critical finding
+4
View File
@@ -53,6 +53,10 @@ Cross-site scripting persists because context, parser, and framework edges are c
## Key Vulnerabilities
### DOMPurify Live-Node Sanitization
Distinguish HTML-string input from live DOM objects sanitized with `IN_PLACE`. Resolve the deployed DOMPurify build and check relevant upstream advisories; a controlled observable `nodeName` is one live-node bypass mechanism. Trace same-origin foreign nodes or `adoptNode()` into this mode before selecting the bypass ([example advisory](https://github.com/cure53/DOMPurify/security/advisories/GHSA-x4vx-rjvf-j5p4)).
### DOM XSS
**Sources**
+29 -131
View File
@@ -11,8 +11,8 @@ import asyncio
import json
import logging
import re
from pathlib import Path, PurePosixPath
from typing import TYPE_CHECKING, Any, cast
from pathlib import PurePosixPath
from typing import TYPE_CHECKING, Any
from agents import RunContextWrapper, function_tool
@@ -21,9 +21,6 @@ from strix.tools.proxy.tools import existing_request_ids
if TYPE_CHECKING:
from collections.abc import Mapping
from strix.fix.contracts import SourceIdentity
from strix.report.state import ReportState
@@ -345,11 +342,16 @@ def _validate_fix_verification(
if str(fix_verification or "").strip():
return []
return [
"fix_verification is REQUIRED when any code_location carries a 'fix_after'. "
"Describe the checks you performed on this draft candidate. Separate executed "
"checks from reasoned checks and list every gap. This statement does not make "
"the candidate ready for automatic application. If the candidate fails a check, "
"revise it or remove 'fix_after'."
"fix_verification is REQUIRED when any code_location carries a 'fix_after' - "
"a suggestion a reviewer can click to apply must be verified first. State, in "
"order: (1) security closure - re-trace the source->sink path through the "
"PATCHED code and say why it is now blocked; (2) bypass review - re-read the "
"diff without your original rationale and name the equivalent sinks, sibling "
"call sites, and alternate malicious input classes you checked; (3) preserved "
"behavior - the legitimate inputs, APIs, and error semantics that still work; "
"(4) how each was checked (executed vs. reasoned), naming any unrun check as "
"an explicit gap. If you cannot make these statements, drop 'fix_after' and "
"leave the location informational."
]
@@ -369,110 +371,6 @@ def _finding_class_of(report: dict[str, Any]) -> str:
return "dynamic"
def _fix_source_context(
report_state: ReportState,
) -> tuple[SourceIdentity | None, Path | None]:
from strix.fix.contracts import SourceIdentity, SourceIdentityKind
raw_context = report_state.get_repository_context()
if raw_context is None:
return None, None
context = cast("dict[str, object]", raw_context)
commit = context.get("commitSha")
if not isinstance(commit, str) or not commit:
return None, None
raw_targets = cast("object", report_state.run_record.get("targets_info"))
targets = cast("list[object]", raw_targets) if isinstance(raw_targets, list) else []
if len(targets) != 1:
return SourceIdentity(kind=SourceIdentityKind.COMMIT, value=commit), None
raw_target = targets[0]
target = cast("dict[str, object]", raw_target) if isinstance(raw_target, dict) else {}
raw_details = target.get("details")
details = cast("dict[str, object]", raw_details) if isinstance(raw_details, dict) else {}
repo_path = details.get("cloned_repo_path")
repository = context.get("repositoryUri")
return (
SourceIdentity(
kind=SourceIdentityKind.COMMIT,
value=commit,
repository=repository if isinstance(repository, str) else None,
),
Path(repo_path) if isinstance(repo_path, str) and repo_path else None,
)
def _build_fix_candidate(
report_state: ReportState,
report_fields: Mapping[str, object],
) -> dict[str, object] | None:
from strix.fix.contracts import candidate_from_legacy_report
from strix.fix.locations import AnchorStatus, anchor_candidate
source_identity, repo_path = _fix_source_context(report_state)
candidate = candidate_from_legacy_report(report_fields, source_identity=source_identity)
if candidate is None:
return None
if repo_path is not None:
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"))
_FIX_CANDIDATE_FIELDS = frozenset(
{
"code_locations",
"remediation_steps",
"technical_analysis",
"poc_description",
"evidence",
"fix_verification",
}
)
def _refresh_fix_candidate(
report_state: ReportState,
report_id: str,
changes: dict[str, Any],
) -> None:
if not _FIX_CANDIDATE_FIELDS.intersection(changes):
return
existing = next(
(
report
for report in report_state.get_existing_vulnerabilities()
if report.get("id") == report_id
),
None,
)
if existing is None:
return
candidate = _build_fix_candidate(report_state, {**existing, **changes})
if candidate is not None:
changes["fix_candidate"] = candidate
if candidate is not None or existing.get("fix_candidate") or existing.get("fix_preparation"):
changes["fix_preparation"] = {
"state": "stale",
"stop_reason": "The finding or draft candidate changed after preparation.",
}
_UPDATE_TEXT_FIELDS = (
"title",
"description",
@@ -753,7 +651,6 @@ def _do_update(
class_error = _fit_revision_to_class(report_state, report_id, changes)
if class_error is not None:
return class_error
_refresh_fix_candidate(report_state, report_id, changes)
try:
updated = report_state.update_vulnerability_report(
@@ -1015,9 +912,6 @@ async def _do_create(
"fix_pr_body": fix_pr_body,
"http_exchange_ids": normalized_http_exchange_ids,
}
fix_candidate = _build_fix_candidate(report_state, report_fields)
if fix_candidate:
report_fields["fix_candidate"] = fix_candidate
dedupe = await check_duplicate(candidate, existing)
if dedupe.get("is_duplicate"):
@@ -1365,10 +1259,11 @@ async def create_vulnerability_report(
``update_vulnerability_report`` once the proxy responds.
Keep IDs out of ``evidence`` and all other report text.
**How ``fix_before`` / ``fix_after`` work**: they describe an
initial fix candidate. A later preparation stage applies,
tests, repairs, and independently verifies the candidate before
Strix can offer an automatic pull request. Therefore:
**How ``fix_before`` / ``fix_after`` work**: they're used as
literal GitHub/GitLab PR suggestion blocks. When a reviewer
accepts the suggestion, the platform replaces the **exact
lines from ``start_line`` to ``end_line``** with
``fix_after``. Therefore:
1. ``fix_before`` must be a **VERBATIM** copy of the source
at those lines — same whitespace, indentation, line
@@ -1389,9 +1284,9 @@ async def create_vulnerability_report(
before SQL"``). Order primary fix first, supporting
changes (imports, config) after.
**Informational vs candidate**:
**Informational vs actionable**:
- With ``fix_before`` / ``fix_after``: actionable fix
candidate for the preparation stage.
(renders as a PR suggestion block).
- Without them: informational context (e.g. showing the
source of tainted data, or a sink that doesn't need
direct editing).
@@ -1426,10 +1321,11 @@ async def create_vulnerability_report(
that aren't part of the fix.
- Duplicating the same change across multiple locations.
fix_verification: REQUIRED whenever any ``code_locations`` entry
carries a ``fix_after``. This field records the reporting
agent's checks on the draft candidate. It is not an independent
verification result. Before writing this field, work the gates
**in order**:
carries a ``fix_after``. A reviewer can apply that
suggestion with one click, so an unverified fix ships
straight into the codebase. Before writing this field, work
the gates **in order** and never trade an earlier one for a
later one:
1. **Security closure** — re-trace the source → sink path
through the *patched* code and state why it is now
@@ -1448,9 +1344,11 @@ async def create_vulnerability_report(
Then write what you did: the commands you ran and their
results, and every gate you could only reason about rather
than execute, marked explicitly as a gap. Do not claim that a
gate passed because the candidate looks correct. If a gate
fails, revise the candidate or drop ``fix_after``.
than execute, marked explicitly as a gap. Do not claim a
gate passed because it looks right. If a gate fails, revise
the patch or drop ``fix_after`` and leave the location
informational — never compensate for a failed security
closure with a smaller diff or extra prose.
Also use this field to record the narrowest-complete-change
judgement: prefer the smallest repository-native fix that
-854
View File
@@ -1,854 +0,0 @@
"""Tests for fix candidate anchoring and repository preparation."""
from __future__ import annotations
import hashlib
import subprocess
import sys
from typing import TYPE_CHECKING
import pytest
from strix.fix.contracts import (
BlockerKind,
CandidateLocation,
CheckResult,
CheckStatus,
CommandSpec,
FixCandidateV1,
FixEdit,
FixPreparationRequestV1,
PreparationBlocker,
PreparationState,
RepairOutcome,
RepairStatus,
ReproductionSpec,
SourceIdentity,
SourceIdentityKind,
VerificationDecision,
VerificationTarget,
VerifierResult,
candidate_from_legacy_report,
)
from strix.fix.locations import AnchorStatus, anchor_location
from strix.fix.prepare import (
PreparationContext,
_network_isolation_prefix,
build_git_manifest,
prepare_fix,
run_command,
)
if TYPE_CHECKING:
from pathlib import Path
def _git(workspace: Path, *args: str) -> str:
return subprocess.run( # noqa: S603
["/usr/bin/git", *args],
cwd=workspace,
check=True,
capture_output=True,
text=True,
).stdout.strip()
def _workspace(tmp_path: Path) -> tuple[Path, str]:
workspace = tmp_path / "repo"
workspace.mkdir()
_git(workspace, "init")
_git(workspace, "config", "user.email", "[email protected]")
_git(workspace, "config", "user.name", "Test")
(workspace / "app.py").write_text("def result():\n return 'unsafe'\n", encoding="utf-8")
_git(workspace, "add", "app.py")
_git(workspace, "commit", "-m", "initial")
return workspace, _git(workspace, "rev-parse", "HEAD")
def _candidate(
commit: str,
*,
reproduction: ReproductionSpec | None = None,
) -> FixCandidateV1:
if reproduction is None:
reproduction = ReproductionSpec(
instructions="Confirm that result returns safe.",
command=CommandSpec(
name="security reproduction",
argv=[
sys.executable,
"-c",
"from app import result; assert result() == 'safe'",
],
),
)
return FixCandidateV1(
source_identity=SourceIdentity(kind=SourceIdentityKind.COMMIT, value=commit),
security_invariant="Return a safe value.",
finding_locations=[
CandidateLocation(
file="app.py",
start_line=1,
end_line=2,
snippet="def result():\n return 'unsafe'",
)
],
draft_edits=[
FixEdit(
file="app.py",
start_line=2,
end_line=2,
before=" return 'unsafe'",
after=" return 'safe'",
)
],
reproduction=reproduction,
)
def _request(candidate: FixCandidateV1, *, attempts: int = 2) -> FixPreparationRequestV1:
return FixPreparationRequestV1(
scan_id="scan-1",
finding_id="finding-1",
candidate=candidate,
checks=[
CommandSpec(
name="compile",
argv=[
sys.executable,
"-c",
"compile(open('app.py', encoding='utf-8').read(), 'app.py', 'exec')",
],
)
],
max_repair_attempts=attempts,
)
async def _noop_repair(
context: PreparationContext,
_checks: list[CheckResult],
) -> RepairOutcome:
path = context.workspace / "app.py"
if path.exists():
path.write_text(
path.read_text(encoding="utf-8").replace("'unsafe'", "'safe'"),
encoding="utf-8",
)
return RepairOutcome(
status=RepairStatus.COMPLETE,
summary="The repository fix is ready for independent evaluation.",
)
async def _verified(
_context: PreparationContext,
_checks: list[CheckResult],
) -> VerifierResult:
return VerifierResult(
decision=VerificationDecision.VERIFIED,
summary="The invariant is closed.",
security_invariant_closed=True,
reproduction_executed=True,
reproduction_summary="The vulnerable input is rejected.",
sibling_paths_reviewed=["app.py"],
preserved_behaviors=["The module compiles."],
security_tests=[
CheckResult(
name="security reproduction",
argv=[sys.executable, "-c", "assert True"],
status=CheckStatus.PASSED,
exit_code=0,
duration_seconds=0,
target=VerificationTarget.PATCHED,
)
],
)
def test_candidate_from_legacy_report_preserves_draft_and_checks() -> None:
candidate = candidate_from_legacy_report(
{
"remediation_steps": "Reject unsafe input.",
"poc_description": "Call the vulnerable function.",
"fix_verification": "Reasoned about the patched branch.",
"code_locations": [
{
"file": "src/app.py",
"start_line": 9,
"end_line": 9,
"snippet": "sink(value)",
"fix_before": "sink(value)",
"fix_after": "sink(clean(value))",
}
],
}
)
assert candidate is not None
assert candidate.security_invariant == "Reject unsafe input."
assert candidate.draft_edits[0].after == "sink(clean(value))"
assert candidate.reported_checks[0].executed is False
assert candidate.known_gaps == ["The reporting-agent verification is not independent."]
def test_anchor_location_rewrites_invented_line_numbers(tmp_path: Path) -> None:
(tmp_path / "app.py").write_text("first\nsecond\ntarget\nlast\n", encoding="utf-8")
location = CandidateLocation(
file="app.py",
start_line=99,
end_line=99,
snippet="target",
)
result = anchor_location(tmp_path, location)
assert result.status is AnchorStatus.UNIQUE
assert result.location.start_line == 3
assert result.location.end_line == 3
@pytest.mark.parametrize(
("content", "expected"),
[
("first\nlast\n", AnchorStatus.MISSING),
("target\nmiddle\ntarget\n", AnchorStatus.AMBIGUOUS),
],
)
def test_anchor_location_rejects_non_unique_source(
tmp_path: Path,
content: str,
expected: AnchorStatus,
) -> None:
(tmp_path / "app.py").write_text(content, encoding="utf-8")
edit = FixEdit(
file="app.py",
start_line=1,
end_line=1,
before="target",
after="safe",
)
assert anchor_location(tmp_path, edit).status is expected
def test_anchor_location_detects_stale_file_digest(tmp_path: Path) -> None:
original = "target\n"
(tmp_path / "app.py").write_text("changed\n", encoding="utf-8")
edit = FixEdit(
file="app.py",
start_line=1,
end_line=1,
before="target",
after="safe",
original_sha256=hashlib.sha256(original.encode()).hexdigest(),
)
assert anchor_location(tmp_path, edit).status is AnchorStatus.STALE
@pytest.mark.asyncio
async def test_prepare_fix_returns_ready_with_manifest(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=_noop_repair,
verify=_verified,
)
assert result.state is PreparationState.READY
assert result.attempts == 1
assert [entry.path for entry in result.final_file_manifest] == ["app.py"]
assert result.security_reproduction is not None
assert result.security_reproduction.target is VerificationTarget.PATCHED
@pytest.mark.asyncio
async def test_prepare_fix_does_not_apply_draft_before_repair(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
observed_source: list[str] = []
async def repair(
context: PreparationContext,
_checks: list[CheckResult],
) -> RepairOutcome:
observed_source.append((context.workspace / "app.py").read_text(encoding="utf-8"))
(context.workspace / "app.py").write_text(
"def result():\n return 'safe'\n",
encoding="utf-8",
)
return RepairOutcome(
status=RepairStatus.COMPLETE,
summary="Implemented the repair from repository context.",
)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=repair,
verify=_verified,
)
assert result.state is PreparationState.READY
assert observed_source == ["def result():\n return 'unsafe'\n"]
@pytest.mark.asyncio
async def test_quality_gate_retries_once_before_verification(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
compile_calls = 0
repair_calls = 0
verification_calls = 0
async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
nonlocal compile_calls
compile_calls += 1
failed = compile_calls == 1
return CheckResult(
name=command.name,
argv=command.argv,
status=CheckStatus.FAILED if failed else CheckStatus.PASSED,
exit_code=1 if failed else 0,
duration_seconds=0,
required=command.required,
)
async def repair(
context: PreparationContext,
checks: list[CheckResult],
) -> RepairOutcome:
nonlocal repair_calls
repair_calls += 1
assert checks == [] if repair_calls == 1 else checks[0].status is CheckStatus.FAILED
(context.workspace / "app.py").write_text(
"def result():\n return 'safe'\n",
encoding="utf-8",
)
return RepairOutcome(status=RepairStatus.COMPLETE, summary="Repair complete.")
async def verify(
context: PreparationContext,
checks: list[CheckResult],
) -> VerifierResult:
nonlocal verification_calls
verification_calls += 1
return await _verified(context, checks)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=repair,
verify=verify,
command_runner=runner,
)
assert result.state is PreparationState.READY
assert repair_calls == 2
assert compile_calls == 2
assert verification_calls == 1
@pytest.mark.asyncio
async def test_failed_quality_gate_never_runs_security_verifier(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
verifier_called = False
async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
return CheckResult(
name=command.name,
argv=command.argv,
status=CheckStatus.FAILED,
exit_code=1,
duration_seconds=0,
required=command.required,
)
async def verify(
context: PreparationContext,
checks: list[CheckResult],
) -> VerifierResult:
nonlocal verifier_called
verifier_called = True
return await _verified(context, checks)
result = await prepare_fix(
_request(_candidate(commit), attempts=1),
workspace,
repair=_noop_repair,
verify=verify,
command_runner=runner,
)
assert result.state is PreparationState.FAILED
assert verifier_called is False
assert "quality gate" in result.stop_reason
@pytest.mark.asyncio
async def test_repository_baseline_failure_is_a_typed_blocker(
tmp_path: Path,
) -> None:
workspace, commit = _workspace(tmp_path)
async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
return CheckResult(
name=command.name,
argv=command.argv,
status=CheckStatus.FAILED,
exit_code=1,
duration_seconds=0,
output="existing failure",
required=command.required,
baseline_status=CheckStatus.FAILED,
baseline_output="existing failure",
)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=_noop_repair,
verify=_verified,
command_runner=runner,
)
assert result.state is PreparationState.BLOCKED
assert result.blocker is not None
assert result.blocker.kind is BlockerKind.REPOSITORY_BASELINE
assert result.attempts == 1
@pytest.mark.asyncio
async def test_different_candidate_and_baseline_failures_are_repairable(
tmp_path: Path,
) -> None:
workspace, commit = _workspace(tmp_path)
repairs = 0
async def repair(
_context: PreparationContext,
_feedback: list[CheckResult],
) -> RepairOutcome:
nonlocal repairs
repairs += 1
return await _noop_repair(_context, _feedback)
async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
return CheckResult(
name=command.name,
argv=command.argv,
status=CheckStatus.FAILED,
exit_code=1,
duration_seconds=0,
output="candidate-specific failure",
required=command.required,
baseline_status=CheckStatus.FAILED,
baseline_output="different pre-existing failure",
)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=repair,
verify=_verified,
command_runner=runner,
)
assert result.state is PreparationState.FAILED
assert result.blocker is None
assert result.attempts == 2
assert repairs == 2
@pytest.mark.asyncio
async def test_unavailable_required_check_is_a_typed_blocker(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
return CheckResult(
name=command.name,
argv=command.argv,
status=CheckStatus.UNAVAILABLE,
duration_seconds=0,
output="compiler unavailable",
required=command.required,
)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=_noop_repair,
verify=_verified,
command_runner=runner,
)
assert result.state is PreparationState.BLOCKED
assert result.blocker is not None
assert result.blocker.kind is BlockerKind.ENVIRONMENT
assert result.attempts == 1
@pytest.mark.asyncio
async def test_repairable_security_rejection_gets_one_retry(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
repair_feedback: list[list[str]] = []
verifier_calls = 0
async def repair(
context: PreparationContext,
_checks: list[CheckResult],
) -> RepairOutcome:
repair_feedback.append(
[
gap
for attempt in context.feedback
if attempt.verifier is not None
for gap in attempt.verifier.gaps
]
)
(context.workspace / "app.py").write_text(
"def result():\n return 'safe'\n",
encoding="utf-8",
)
return RepairOutcome(status=RepairStatus.COMPLETE, summary="Repair complete.")
async def verify(
context: PreparationContext,
checks: list[CheckResult],
) -> VerifierResult:
nonlocal verifier_calls
verifier_calls += 1
if verifier_calls == 1:
return VerifierResult(
decision=VerificationDecision.REJECTED,
summary="A sibling path remains vulnerable.",
gaps=["Harden the sibling path."],
repairable=True,
)
return await _verified(context, checks)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=repair,
verify=verify,
)
assert result.state is PreparationState.READY
assert result.attempts == 2
assert repair_feedback == [[], ["Harden the sibling path."]]
@pytest.mark.asyncio
async def test_security_blocker_stops_without_repair_retry(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
repair_calls = 0
blocker = PreparationBlocker(
kind=BlockerKind.EXTERNAL_CONFIGURATION,
summary="A production account mapping is unavailable.",
user_action="Provide the production account mapping.",
)
async def repair(
context: PreparationContext,
checks: list[CheckResult],
) -> RepairOutcome:
nonlocal repair_calls
repair_calls += 1
return await _noop_repair(context, checks)
async def verify(
_context: PreparationContext,
_checks: list[CheckResult],
) -> VerifierResult:
return VerifierResult(
decision=VerificationDecision.INCONCLUSIVE,
summary=blocker.summary,
blocker=blocker,
)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=repair,
verify=verify,
)
assert result.state is PreparationState.BLOCKED
assert result.blocker == blocker
assert repair_calls == 1
@pytest.mark.asyncio
async def test_budget_exhaustion_fails_without_verification(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
verifier_called = False
async def exhausted(
context: PreparationContext,
_checks: list[CheckResult],
) -> RepairOutcome:
(context.workspace / "app.py").write_text(
"def result():\n return 'safe'\n",
encoding="utf-8",
)
return RepairOutcome(
status=RepairStatus.BUDGET_EXHAUSTED,
summary="The repair hit its emergency turn ceiling after editing the patch.",
turns_used=40,
)
async def verify(
context: PreparationContext,
checks: list[CheckResult],
) -> VerifierResult:
nonlocal verifier_called
verifier_called = True
return await _verified(context, checks)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=exhausted,
verify=verify,
)
assert result.state is PreparationState.FAILED
assert verifier_called is False
assert result.changed_files == ["app.py"]
@pytest.mark.asyncio
async def test_prepare_fix_preserves_explicit_blocked_outcome(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
blocker = PreparationBlocker(
kind=BlockerKind.CREDENTIAL,
summary="A required credential is unavailable.",
user_action="Provide a repository-scoped test credential.",
)
async def blocked(
_context: PreparationContext,
_checks: list[CheckResult],
) -> RepairOutcome:
return RepairOutcome(
status=RepairStatus.BLOCKED,
summary=blocker.summary,
blocker=blocker,
)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=blocked,
verify=_verified,
)
assert result.state is PreparationState.BLOCKED
assert result.blocker == blocker
assert result.verifier is None
@pytest.mark.asyncio
async def test_prepare_fix_requires_verifier_security_test(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
async def verify_without_test(
_context: PreparationContext,
_checks: list[CheckResult],
) -> VerifierResult:
return VerifierResult(
decision=VerificationDecision.VERIFIED,
summary="The verifier did not execute the issue-specific test.",
security_invariant_closed=True,
reproduction_executed=False,
)
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=_noop_repair,
verify=verify_without_test,
)
assert result.state is PreparationState.FAILED
assert "did not execute a security test" in " ".join(result.gaps)
@pytest.mark.asyncio
async def test_prepare_fix_requires_closed_security_invariant(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
async def incomplete(
context: PreparationContext,
checks: list[CheckResult],
) -> VerifierResult:
verified = await _verified(context, checks)
return verified.model_copy(update={"security_invariant_closed": False})
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=_noop_repair,
verify=incomplete,
)
assert result.state is PreparationState.FAILED
assert "invariant" in " ".join(result.gaps)
@pytest.mark.asyncio
async def test_prepare_fix_requires_a_source_change(tmp_path: Path) -> None:
workspace, commit = _workspace(tmp_path)
async def no_change(
_context: PreparationContext,
_checks: list[CheckResult],
) -> RepairOutcome:
return RepairOutcome(status=RepairStatus.COMPLETE, summary="No change.")
result = await prepare_fix(
_request(_candidate(commit)),
workspace,
repair=no_change,
verify=_verified,
)
assert result.state is PreparationState.FAILED
assert result.final_file_manifest == []
assert "change repository source" in result.stop_reason
@pytest.mark.asyncio
async def test_prepare_fix_rejects_wrong_source_commit(tmp_path: Path) -> None:
workspace, _commit = _workspace(tmp_path)
result = await prepare_fix(
_request(_candidate("0" * 40)),
workspace,
repair=_noop_repair,
verify=_verified,
)
assert result.state is PreparationState.STALE
assert result.blocker is not None
assert result.blocker.kind is BlockerKind.SOURCE
@pytest.mark.asyncio
async def test_prepare_fix_does_not_apply_escaping_draft_edit(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")
candidate = _candidate(_git(workspace, "rev-parse", "HEAD")).model_copy(
update={
"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.READY
assert outside.read_text(encoding="utf-8") == "target\n"
assert "link.txt" not in result.changed_files
@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
@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', '<absent>'))",
],
)
sealed = await run_command(tmp_path, command, network_allowed=True)
assert sealed.status is CheckStatus.PASSED
assert "<absent>" 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
+1 -1
View File
@@ -264,4 +264,4 @@ def test_render_vulnerability_md_surfaces_calibration_metadata() -> None:
assert "Egress appears filtered at the network layer." in md
assert "## Confidence Rationale" in md
assert "## What Would Change This Severity" in md
assert "## Reported Candidate Checks" in md
assert "## Fix Verification" in md
-24
View File
@@ -1981,30 +1981,6 @@ def test_update_refuses_code_locations_it_cannot_use(report_state: ReportState)
assert any("start_line" in error for error in result["errors"])
def test_update_marks_preparation_stale_when_candidate_cannot_be_rebuilt(
report_state: ReportState,
) -> None:
"""A revision whose locations cannot form a candidate still supersedes the
preparation recorded for the draft it replaced."""
_seed_weak_report(report_state)
report = report_state.vulnerability_reports[0]
candidate = {"security_invariant": "Parametrize the query.", "draft_edits": []}
report["fix_candidate"] = candidate
report["fix_preparation"] = {"state": "ready", "candidate_digest": "0" * 64}
result = _do_update(
report_id="vuln-0009",
update_reason="Repointing the finding to the validated handler.",
fields={"code_locations": [{"file": "./files.py", "start_line": 4, "end_line": 9}]},
)
assert result["success"] is True
assert report["code_locations"] == [{"file": "./files.py", "start_line": 4, "end_line": 9}]
assert report["fix_preparation"]["state"] == "stale"
assert "changed after preparation" in report["fix_preparation"]["stop_reason"]
assert report["fix_candidate"] == candidate
def _seed_saved_report(report_state: ReportState) -> Path:
"""The weak report, written to disk the way a filed finding is."""
_seed_weak_report(report_state)
+3 -112
View File
@@ -5,7 +5,6 @@ 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
@@ -130,37 +129,14 @@ 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 verified candidate as an artifactChange.
candidate = _fix_candidate()
def test_write_sarif_builds_fixes_from_code_location_fix_pairs(tmp_path: Path) -> None:
# A code location carrying fix_before/fix_after must surface as a SARIF
# fix (artifactChange/replacement) so consumers can offer a one-click fix.
write_sarif(
tmp_path,
[
_finding(
remediation_steps="Use a parameterized query.",
fix_candidate=candidate,
fix_preparation={
"state": "ready",
"candidate_digest": FixCandidateV1.model_validate(candidate).digest(),
},
code_locations=[
{
"file": "app.py",
@@ -183,91 +159,6 @@ 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_emits_verified_candidate_from_preparation(tmp_path: Path) -> None:
# Preparation re-anchors the draft to the lines its block actually occupies,
# so the result carries the verified candidate and SARIF emits those lines
# rather than the stale positions the report's draft still records.
verified = _fix_candidate(
draft_edits=[
{
"file": "app.py",
"start_line": 9,
"end_line": 9,
"before": 'query = "SELECT * FROM u WHERE id=" + uid',
"after": 'query = "SELECT * FROM u WHERE id=%s"',
}
]
)
write_sarif(
tmp_path,
[
_finding(
fix_candidate=_fix_candidate(),
fix_preparation={
"state": "ready",
"candidate": verified,
"candidate_digest": FixCandidateV1.model_validate(verified).digest(),
},
)
],
)
result = _read(tmp_path)["runs"][0]["results"][0]
replacement = result["fixes"][0]["artifactChanges"][0]["replacements"][0]
assert replacement["deletedRegion"]["startLine"] == 9
assert replacement["deletedRegion"]["endLine"] == 9
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,
[
_finding(
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_omits_fixes_without_fix_pairs(tmp_path: Path) -> None:
write_sarif(tmp_path, [_finding()])
assert "fixes" not in _read(tmp_path)["runs"][0]["results"][0]