Compare commits

...

4 Commits

Author SHA1 Message Date
Alex Schapiro
38186a7e50 skills: point to the strix cloud CLI in every skill 2026-09-01 22:05:20 +00:00
Ahmed Allam
8fdf6a5c09 chore: release v1.6.0 2026-09-01 23:29:06 +03:00
alex s
46cf2f52f3 report: add update_vulnerability_report so an agent can revise a filed finding (#1210)
Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>
2026-09-01 13:07:17 -07:00
alex s
3de9471431 Link CLI wallet (#1222) 2026-09-01 16:00:32 -04:00
28 changed files with 1834 additions and 105 deletions

View File

@@ -1,6 +1,6 @@
[project]
name = "strix-agent"
version = "1.5.3"
version = "1.6.0"
description = "Open-source AI Hackers for your apps"
readme = "README.md"
license = "Apache-2.0"

View File

@@ -11,7 +11,7 @@ metadata:
APIs fail differently from web UIs: there is no rendered surface to crawl, the interesting bugs are authorization-shaped rather than injection-shaped, and the same endpoint behaves differently per token. This workflow targets those specifics with Strix's autonomous agents, using the current [OWASP API Security Top 10 (2023)](https://owasp.org/API-Security/editions/2023/en/0x11-t10/) as the coverage checklist. For the web-app equivalent, the current edition is the OWASP Top 10:2025 — see **owasp-top-10-testing**.
Install, LLM setup, full CLI flags, and the managed-cloud path are in the **penetration-testing-with-strix** skill. Read it if `strix --version` fails or the target is not an API.
Install, LLM setup, full CLI flags, and the managed-cloud path are in the **penetration-testing-with-strix** skill. Read it if `strix --version` fails or the target is not an API. For a run with no Docker and no LLM key, the same binary drives the managed platform: `strix cloud login`, then `strix cloud scans start ...` (details in **managed-pentesting-with-strix**).
## 1. Gather what the agents need

View File

@@ -11,7 +11,7 @@ metadata:
Entry point for "make my application secure" requests, where the target is not yet a single URL or repo. The job here is to pick the right test per asset, run it, and produce one ranked plan — not to run everything at maximum depth.
Install, LLM setup, all CLI flags, and the managed-cloud path live in the **penetration-testing-with-strix** skill. Read it first if `strix --version` fails.
Install, LLM setup, all CLI flags, and the managed-cloud path live in the **penetration-testing-with-strix** skill. Read it first if `strix --version` fails. For a run with no Docker and no LLM key, the same binary drives the managed platform: `strix cloud login`, then `strix cloud scans start ...` (details in **managed-pentesting-with-strix**).
Only test assets the user owns or is authorized to test. Confirm authorization before the first run, and prefer staging over production, because the agents send real exploit payloads and can change data.

View File

@@ -113,11 +113,11 @@ Gate the pipeline on the exit code (see the budget/fail-open caveat above — gi
# Option B — Managed platform (no runner infra)
No workflow file, no Docker, no LLM key. Two ways to use it:
No workflow file, no Docker, no LLM key. Three ways to use it:
1. **PR-review app (zero code):** the user installs the Strix GitHub/GitLab/Bitbucket app and enables PR reviews for the repo in the app.strix.ai dashboard. Every PR is then reviewed automatically, with findings posted as PR comments. Nothing to add to the repo. This is the lowest-effort path — recommend it first when the user just wants PR gating.
2. **API-triggered from any pipeline:** if you want to trigger from an existing pipeline (or a system without the SCM app), call the API with a token that has `pr_reviews:write` (or `scans:write`). Store the token as a CI secret; ask the user to create it at **Settings → API Access**. Example GitHub Actions step:
2. **CLI-triggered from any pipeline:** if you want to trigger from an existing pipeline (or a system without the SCM app), use the same `strix` binary with a token that has `pr_reviews:write`. Store the token as a CI secret and ask the user to create it at **Settings → API Access**. Read the repository's `provider` and `installation_id` once with `strix cloud repos list`. Example GitHub Actions step:
```yaml
- name: Strix PR review (managed)
@@ -125,12 +125,25 @@ No workflow file, no Docker, no LLM key. Two ways to use it:
env:
STRIX_API_TOKEN: ${{ secrets.STRIX_API_TOKEN }}
run: |
curl -sS --fail https://app.strix.ai/api/v1/pr-reviews/start \
-H "Authorization: Bearer $STRIX_API_TOKEN" \
-H "Content-Type: application/json" \
-d "{\"repository_full_name\":\"${{ github.repository }}\",\"pr_number\":${{ github.event.pull_request.number }}}"
curl -sSL https://strix.ai/install | bash
strix cloud pr-reviews start \
--provider github \
--installation-id "${{ vars.STRIX_INSTALLATION_ID }}" \
--repository-full-name "${{ github.repository }}" \
--pr-number "${{ github.event.pull_request.number }}"
```
To gate the build on results, poll the PR review / scan status and fail on unresolved criticals/highs. Full endpoints (PR reviews, scans, SARIF export, schedules for scheduled deep scans) are in the **managed-pentesting-with-strix** skill.
Output is JSON when stdout is not a terminal, and there are no prompts without a TTY. To gate the build on results, poll `strix cloud pr-reviews get <id> --json` and fail on unresolved criticals or highs. The raw REST endpoint (`POST /api/v1/pr-reviews/start`) works too when the pipeline cannot install the CLI.
3. **Source upload from a pipeline without an SCM app:** upload the checked-out tree as a cloud code review (`scans:write` and `uploads:write`). The two-step digest handoff keeps a human in control of what leaves the runner:
```bash
strix cloud scans start --source . --dry-run --show-files --json # review, capture source.archive_sha256
strix cloud scans start --source . --approve-sha256 "$SOURCE_SHA256" --wait
```
Exit codes: `0` success, `4` auth or plan limit, `5` payment required. Non-Enterprise scans consume credits.
Full CLI coverage (PR reviews, scans, SARIF export, schedules) is in the **managed-pentesting-with-strix** skill.
Recommend Option B for most teams (no maintenance, central dashboard); use Option A when scans must stay entirely within your own infrastructure.

View File

@@ -11,7 +11,7 @@ metadata:
White-box security review with Strix: the agents read the source to build a model of routes, sinks, and authorization checks, then attempt real exploitation. Findings come with a proof-of-concept, so the output is a short list of proven issues rather than the hundreds of "potential" hits a pattern-matching scanner produces.
Install, LLM setup, all flags, and the managed-cloud path are in the **penetration-testing-with-strix** skill.
Install, LLM setup, all flags, and the managed-cloud path are in the **penetration-testing-with-strix** skill. For a run with no Docker and no LLM key, the same binary drives the managed platform: `strix cloud login`, then `strix cloud scans start ...` (details in **managed-pentesting-with-strix**).
## Run it

View File

@@ -18,7 +18,7 @@ Get the findings from wherever the scan ran:
- **OSS CLI** — artifacts in `strix_runs/<run-name>/`:
- `vulnerabilities/*.md` — one finding per file: description, severity, PoC steps or script, affected code locations, remediation guidance.
- `vulnerabilities.json` — the same findings as JSON (ids, severity, CWE/CVE, `code_locations` with `fix_before`/`fix_after` suggestions when available).
- **Cloud (app.strix.ai)** — fetch the scan's `vulnerabilities[]` via `GET /api/v1/scans/{scanId}` (or `GET /api/v1/vulnerabilities` org-wide). Each carries `severity, cwe, endpoint, method, impact, technical_analysis, poc_description, poc_script_code` and, for code findings, `code_file`/`code_diff`/`code_before`/`code_after`. See the **managed-pentesting-with-strix** skill for auth.
- **Cloud (app.strix.ai)** — pull findings with the CLI: `strix cloud vulns list --scan-id <scan-id> --json` (or `strix cloud scans get <scan-id> --json | jq '.vulnerabilities'`, or `strix cloud vulns list --severity critical` org-wide). Each finding carries `severity, cwe, endpoint, method, impact, technical_analysis, poc_description, poc_script_code` and, for code findings, `code_file`/`code_diff`/`code_before`/`code_after`. After a fix is verified, mark it with `strix cloud vulns update <id> --status fixed`. See the **managed-pentesting-with-strix** skill for `strix cloud login` and scopes.
Order work by severity: critical → high → medium → low. Every Strix finding was validated with a working proof-of-concept, so do not dismiss findings as false positives without re-testing the PoC yourself.

View File

@@ -96,7 +96,7 @@ Non-Enterprise scans consume org credits. Enterprise engagements are plan-includ
strix cloud credits
```
When the balance is too low, buy credits with `strix cloud billing topup` (`billing:write`, admin token). The server answers the first request with **HTTP 402 and a machine-payment challenge** (Stripe Machine Payments Protocol). The CLI pays the challenge with the `mppx` client when Node.js is available — the user approves the spend in their agent wallet, for example the [Link Agent Wallet](https://link.com/agents). The response returns the receipt (`credits_granted`, `duplicate`, `reference`) and the new balance.
When the balance is too low, buy credits with `strix cloud billing topup` (`billing:write`, admin token). The server answers the first request with **HTTP 402 and a machine-payment challenge** (Stripe Machine Payments Protocol). The CLI pays the challenge with the Stripe Link wallet client when Node.js is available — the user approves the spend in the [Link app](https://link.com/agents). The response returns the receipt (`credits_granted`, `duplicate`, `reference`) and the new balance.
A default-tier source-only code review currently starts at 60 credits. Source uploads are not free: they launch an ordinary `code_review` and use the same deterministic scope estimator. The service checks the full balance before launch, reserves credits atomically only after validation succeeds, and does not create or charge a rejected scan. Retests and Enterprise scans are exempt.
@@ -105,7 +105,7 @@ strix cloud billing topup --credits 20 --yes # explicit approval; skips the TT
strix cloud billing topup --credits 20 --no-pay # print the 402 challenge without paying
```
The default payment path is the Stripe agent wallet. Tell the user to set it up one time at [link.com/agents](https://link.com/agents). After setup, the user approves each payment in the Link app, and no keys or variables are necessary.
The default payment path is the Stripe Link wallet. When no wallet is connected, an interactive `strix cloud billing topup` starts the Link sign-in for the user and prints the verification link. The user approves the connection one time in the Link app, and then approves each payment there. No keys or variables are necessary. In a non-interactive process, the command stops and tells the user to connect the wallet at [link.com/agents](https://link.com/agents) or to use the hosted checkout link.
In a non-interactive agent or CI process, payment never proceeds unless the command includes `--yes`. Show the challenge or estimated spend to the user and obtain approval before adding it. `--no-pay` always stops after printing the challenge.

View File

@@ -13,7 +13,7 @@ The OWASP Top 10 is a taxonomy of risk categories, not a test suite — "OWASP T
**Use the current edition: [OWASP Top 10:2025](https://owasp.org/Top10/)** (8th installment, superseding 2021). Ask the user before targeting an older edition — some compliance checklists still reference 2021, and a report labelled with the wrong edition is misleading. Key differences from 2021: **SSRF is folded into A01**, **A03 Software Supply Chain Failures** expands the old "Vulnerable and Outdated Components", and **A10 Mishandling of Exceptional Conditions** is new; A02 Security Misconfiguration moved 5→2.
Install, LLM setup, and the managed-cloud alternative: **penetration-testing-with-strix**.
Install, LLM setup, and the managed-cloud alternative: **penetration-testing-with-strix**. For a run with no Docker and no LLM key, the same binary drives the managed platform: `strix cloud login`, then `strix cloud scans start ...` (details in **managed-pentesting-with-strix**).
## What is and is not testable by an agent

View File

@@ -11,7 +11,7 @@ metadata:
Black-box (and optionally source-assisted) penetration testing of a running web app with Strix's autonomous agents. Every reported finding is validated with a working exploit, so there are no signature-based false positives to triage.
Install, LLM setup, all CLI flags, and the managed-cloud alternative are covered in the **penetration-testing-with-strix** skill — read it if the target is not a running web app, or if `strix --version` fails. This skill is the web-app-specific workflow.
Install, LLM setup, all CLI flags, and the managed-cloud alternative are covered in the **penetration-testing-with-strix** skill — read it if the target is not a running web app, or if `strix --version` fails. For a run with no Docker and no LLM key, the same binary drives the managed platform: `strix cloud login`, then `strix cloud scans start ...` (details in **managed-pentesting-with-strix**). This skill is the web-app-specific workflow.
## 1. Confirm authorization and scope

View File

@@ -52,6 +52,7 @@ from strix.tools.reporting.tool import (
create_vulnerability_report,
get_report,
list_reports,
update_vulnerability_report,
)
from strix.tools.respond.tool import respond_to_user
from strix.tools.thinking.tool import think
@@ -580,6 +581,7 @@ _BASE_TOOLS: tuple[Tool, ...] = (
web_search,
create_vulnerability_report,
create_dependency_report,
update_vulnerability_report,
list_reports,
get_report,
list_requests,

View File

@@ -239,7 +239,8 @@ VALIDATION REQUIREMENTS:
- 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
- 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
- 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.
- 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.
- REVIEWING FILED FINDINGS (orchestrator/root agent): use list_reports to see every vulnerability filed so far in this scan (by any agent, root or child) — metadata-first with per-severity counts — and get_report to read one finding in full by its id. These are read-only orchestration tools: the root agent uses them to track coverage, avoid dispatching work on already-covered ground, assemble the finish_scan executive summary, and reason about attack-chaining across confirmed findings. Leaf/specialist agents should NOT call them — just do your assigned testing and file findings. Each entry shows which agent filed it (agent_name), and your own entries are flagged by_you. list_notes/get_note do the same for notes.
STATE & COORDINATION TOOLS (when and how):

View File

@@ -105,14 +105,15 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
report_state.set_scan_config(scan_config)
report_state.save_run_data()
def display_vulnerability(report: dict[str, Any]) -> None:
def display_vulnerability(report: dict[str, Any], *, updated: bool = False) -> None:
report_id = report.get("id", "unknown")
vuln_text = format_vulnerability_report(report)
suffix = " (updated)" if updated else ""
vuln_panel = Panel(
vuln_text,
title=f"[bold red]{report_id.upper()}",
title=f"[bold red]{report_id.upper()}{suffix}",
title_align="left",
border_style="red",
padding=(1, 2),
@@ -122,6 +123,9 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
console.print()
report_state.vulnerability_found_callback = display_vulnerability
report_state.vulnerability_updated_callback = lambda report: display_vulnerability(
report, updated=True
)
def cleanup_on_exit() -> None:
report_state.cleanup()

View File

@@ -9,6 +9,8 @@ import shutil
import subprocess
import sys
import tempfile
import webbrowser
from contextlib import suppress
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
@@ -30,6 +32,23 @@ _MAX_WALLET_DETAIL_CHARS = 2_000
# platform. This version is also old enough to remain installable in npm
# environments that apply a short package-publication safety window.
_MPPX_PACKAGE = "mppx@0.8.17"
# Stripe's own wallet client. It runs the complete challenge flow: it creates a
# spend request, waits for the person to approve it in the Link app, and retries
# the payment with the approved credential.
_LINK_CLI_PACKAGE = "@stripe/link-cli@0.13.1"
_LINK_CLI_CLIENT_NAME = "Strix CLI"
_LINK_LOGIN_TIMEOUT_S = 300
# Poll every 2 seconds while the person approves the spend request in the Link
# app. 150 attempts give the person 5 minutes.
_LINK_APPROVAL_POLL_INTERVAL_S = 2
_LINK_APPROVAL_MAX_ATTEMPTS = 150
# Bound every wallet subprocess so a stalled npm download or wallet request
# cannot block the top-up command forever. The poll step gets the full
# approval window plus this margin.
_WALLET_STEP_TIMEOUT_S = 300
_LINK_APPROVAL_TIMEOUT_S = (
_LINK_APPROVAL_POLL_INTERVAL_S * _LINK_APPROVAL_MAX_ATTEMPTS + _WALLET_STEP_TIMEOUT_S
)
_NPM_REGISTRY = "https://registry.npmjs.org"
_WALLET_ENV_NAMES = frozenset(
{
@@ -58,6 +77,8 @@ _WALLET_ENV_NAMES = frozenset(
"TMPDIR",
"USERPROFILE",
"XDG_CONFIG_HOME",
"XDG_DATA_HOME",
"XDG_STATE_HOME",
"all_proxy",
"http_proxy",
"https_proxy",
@@ -74,7 +95,7 @@ class _WalletClientResult:
upstream_responses: tuple[WalletUpstreamResponse, ...]
def run_topup( # noqa: PLR0911, PLR0912
def run_topup( # noqa: PLR0911, PLR0912, PLR0915
console: Console,
args: argparse.Namespace,
body: dict[str, Any],
@@ -137,24 +158,25 @@ def run_topup( # noqa: PLR0911, PLR0912
payment_method = getattr(args, "payment_method", None) or os.environ.get(
"MPPX_STRIPE_PAYMENT_METHOD"
)
if not payment_method and (
not as_json
and not os.environ.get("MPPX_ACCOUNT")
and not os.environ.get("MPPX_STRIPE_SECRET_KEY")
):
console.print(
"[dim]Tip: payments need a wallet. Set up a Stripe agent wallet at "
"https://link.com/agents, and the user approves each payment in the Link app. "
"If the user does not want a wallet, run "
"`strix cloud billing subscribe --plan strix_top_up` for a hosted checkout link.[/]"
)
use_link_wallet = payment_method is None and not _mppx_wallet_configured()
if use_link_wallet:
setup_error = _prepare_link_wallet(console, npx, as_json=as_json)
if setup_error is not None:
emit(
console,
{"error": setup_error, "challenge": challenge},
as_json=as_json,
)
return http.EXIT_PAYMENT
try:
wallet_result = _run_wallet_client(
console,
npx,
args,
body,
token=token,
payment_method=payment_method,
use_link_wallet=use_link_wallet,
capture_output=as_json,
)
except KeyboardInterrupt:
@@ -185,19 +207,20 @@ def run_topup( # noqa: PLR0911, PLR0912
result = wallet_result.process
confirmed_receipt = _confirmed_topup_receipt(wallet_result.upstream_responses)
if confirmed_receipt is not None:
if as_json:
emit(console, confirmed_receipt, as_json=True)
emit(console, confirmed_receipt, as_json=as_json)
return http.EXIT_OK
stdout = str(getattr(result, "stdout", "") or "").strip()
stderr = str(getattr(result, "stderr", "") or "").strip()
if not as_json:
console.print(
"[yellow]The wallet exited without a confirmed receipt. The payment outcome is "
"unknown; run `strix cloud billing credits` before retrying.[/]"
)
detail = _wallet_detail(stderr or stdout or "")
if detail:
console.print(f"[dim]Wallet output: {detail}[/]")
return http.EXIT_PAYMENT
stdout = str(getattr(result, "stdout", "") or "").strip()
stderr = str(getattr(result, "stderr", "") or "").strip()
if result.returncode == 0:
try:
receipt = json.loads(stdout)
@@ -262,15 +285,17 @@ def run_topup( # noqa: PLR0911, PLR0912
def _run_wallet_client(
console: Console,
npx: str,
args: argparse.Namespace,
body: dict[str, Any],
*,
token: str | None,
payment_method: str | None,
use_link_wallet: bool,
capture_output: bool,
) -> _WalletClientResult:
"""Run mppx through the loopback bridge without exposing the API token to it."""
"""Run the wallet through the loopback bridge without exposing the API token."""
upstream_url = f"{http.app_url()}/api/v1/billing/topup"
body_json = json.dumps(body)
wallet_env = _wallet_environment()
@@ -281,6 +306,7 @@ def _run_wallet_client(
global_config = wallet_root / "global.npmrc"
user_config.touch(mode=0o600)
global_config.touch(mode=0o600)
npx_prefix = _npx_prefix(npx, wallet_root)
with wallet_payment_bridge(
upstream_url=upstream_url,
api_token=http.api_token(token),
@@ -289,39 +315,356 @@ def _run_wallet_client(
timeout=getattr(args, "timeout", None),
response_observer=upstream_responses.append,
) as wallet_url:
command = [
npx,
"--yes",
f"--registry={_NPM_REGISTRY}",
"--ignore-scripts",
f"--userconfig={user_config}",
f"--globalconfig={global_config}",
f"--cache={wallet_root / 'npm-cache'}",
_MPPX_PACKAGE,
wallet_url,
"--fail",
"-J",
body_json,
]
if payment_method:
command += ["-M", f"paymentMethod={payment_method}"]
process = subprocess.run( # noqa: S603
command,
check=False,
capture_output=capture_output,
text=True,
env=wallet_env,
cwd=wallet_root,
)
if use_link_wallet:
process = _run_link_wallet_flow(
console,
npx_prefix,
wallet_url,
body,
body_json,
wallet_env,
wallet_root,
quiet=capture_output,
)
else:
command = [
*npx_prefix,
_MPPX_PACKAGE,
wallet_url,
"--fail",
"-J",
body_json,
]
if payment_method:
command += ["-M", f"paymentMethod={payment_method}"]
try:
process = subprocess.run( # noqa: S603
command,
check=False,
capture_output=capture_output,
text=True,
env=wallet_env,
cwd=wallet_root,
timeout=_LINK_APPROVAL_TIMEOUT_S,
)
except subprocess.TimeoutExpired as timeout_error:
process = subprocess.CompletedProcess(
args=command,
returncode=1,
stdout=_decoded_stream(timeout_error.stdout),
stderr=(
"The wallet step did not complete within "
f"{_LINK_APPROVAL_TIMEOUT_S} seconds."
),
)
return _WalletClientResult(process=process, upstream_responses=tuple(upstream_responses))
def _run_link_wallet_flow(
console: Console,
npx_prefix: list[str],
wallet_url: str,
body: dict[str, Any],
body_json: str,
wallet_env: dict[str, str],
wallet_root: Path,
*,
quiet: bool,
) -> subprocess.CompletedProcess[str]:
"""Create the spend request, wait for approval in the Link app, then pay."""
def run_step(
arguments: list[str],
progress_message: str,
timeout: int = _WALLET_STEP_TIMEOUT_S,
) -> subprocess.CompletedProcess[str]:
command = [*npx_prefix, _LINK_CLI_PACKAGE, *arguments]
def run() -> subprocess.CompletedProcess[str]:
try:
return subprocess.run( # noqa: S603
command,
check=False,
capture_output=True,
text=True,
env=wallet_env,
cwd=wallet_root,
timeout=timeout,
)
except subprocess.TimeoutExpired as timeout_error:
return subprocess.CompletedProcess(
args=command,
returncode=1,
stdout=_decoded_stream(timeout_error.stdout),
stderr=f"The wallet step did not complete within {timeout} seconds.",
)
if quiet:
return run()
with console.status(progress_message):
return run()
created = run_step(
[
"mpp",
"pay",
wallet_url,
"--method",
"POST",
"--data",
body_json,
"--context",
_payment_context(body),
"--format",
"json",
],
"Starting the Stripe Link wallet…",
)
spend_request = _pending_spend_request(created.stdout)
if spend_request is None:
return created
request_id, approval_url = spend_request
if not quiet:
console.print(f"[yellow]Approve the payment in the Link app:[/] {approval_url}")
if sys.stdin.isatty() and sys.stdout.isatty() and approval_url.startswith("https://"):
with suppress(Exception):
webbrowser.open(approval_url)
polled = run_step(
[
"spend-request",
"retrieve",
request_id,
"--interval",
str(_LINK_APPROVAL_POLL_INTERVAL_S),
"--max-attempts",
str(_LINK_APPROVAL_MAX_ATTEMPTS),
"--format",
"jsonl",
],
"Waiting for the approval in the Link app…",
timeout=_LINK_APPROVAL_TIMEOUT_S,
)
if _final_spend_request_status(polled.stdout) != "approved":
return polled
return run_step(
[
"mpp",
"pay",
wallet_url,
"--spend-request-id",
request_id,
"--method",
"POST",
"--data",
body_json,
"--format",
"json",
],
"Completing the payment…",
)
def _decoded_stream(stream: str | bytes | None) -> str:
"""Return captured subprocess output as text."""
if stream is None:
return ""
if isinstance(stream, bytes):
return stream.decode(errors="replace")
return stream
def _embedded_json_documents(text: str) -> list[Any]:
"""Extract JSON documents from wallet output that can contain other text."""
documents: list[Any] = []
decoder = json.JSONDecoder()
position = 0
while position < len(text):
start_candidates = [
index for index in (text.find("[", position), text.find("{", position)) if index != -1
]
if not start_candidates:
break
start = min(start_candidates)
try:
document, end = decoder.raw_decode(text, start)
except ValueError:
position = start + 1
continue
documents.append(document)
position = end
return documents
def _spend_request_records(stdout: str) -> list[dict[str, Any]]:
"""Parse spend-request records from JSON or JSON-lines wallet output."""
records: list[dict[str, Any]] = []
for candidate in _embedded_json_documents((stdout or "").strip()):
items = candidate if isinstance(candidate, list) else [candidate]
for item in items:
if not isinstance(item, dict):
continue
record = cast("dict[str, Any]", item)
data = record.get("data")
if isinstance(data, dict):
record = cast("dict[str, Any]", data)
records.append(record)
return records
def _pending_spend_request(stdout: str) -> tuple[str, str] | None:
"""Find a spend request that waits for approval in the Link app."""
for record in _spend_request_records(stdout):
request_id = record.get("id")
approval_url = record.get("approval_url")
if (
record.get("status") == "pending_approval"
and isinstance(request_id, str)
and request_id
and isinstance(approval_url, str)
):
return request_id, approval_url
return None
def _final_spend_request_status(stdout: str) -> str | None:
"""Return the last reported status from the approval poll output."""
status: str | None = None
for record in _spend_request_records(stdout):
value = record.get("status")
if isinstance(value, str):
status = value
return status
def _npx_prefix(npx: str, wallet_root: Path) -> list[str]:
"""Install the wallet client from a fixed registry without lifecycle scripts."""
return [
npx,
"--yes",
f"--registry={_NPM_REGISTRY}",
"--ignore-scripts",
f"--userconfig={wallet_root / 'user.npmrc'}",
f"--globalconfig={wallet_root / 'global.npmrc'}",
f"--cache={_wallet_npm_cache()}",
]
def _wallet_npm_cache() -> Path:
"""Keep one private npm cache so the pinned wallet client installs once."""
cache = Path.home() / ".strix" / "wallet-npm-cache"
cache.mkdir(mode=0o700, parents=True, exist_ok=True)
return cache
def _payment_context(body: dict[str, Any]) -> str:
"""Describe the purchase for the person who approves it in the Link app."""
credits_requested = body.get("credits")
return (
f"Strix scan credits. The Strix command line interface asks to buy "
f"{credits_requested} scan credit(s) for the selected Strix workspace on "
"app.strix.ai. Strix spends the credits on managed penetration test scans "
"that the user starts."
)
def _mppx_wallet_configured() -> bool:
"""Report whether the person already configured the mppx wallet client."""
return bool(os.environ.get("MPPX_ACCOUNT") or os.environ.get("MPPX_STRIPE_SECRET_KEY"))
def _run_link_cli(
npx: str,
arguments: list[str],
*,
capture_output: bool,
timeout: float | None = None,
) -> subprocess.CompletedProcess[str]:
"""Run one Stripe Link wallet command in an isolated npm environment."""
with tempfile.TemporaryDirectory(prefix="strix-wallet-") as wallet_cwd:
wallet_root = Path(wallet_cwd)
(wallet_root / "user.npmrc").touch(mode=0o600)
(wallet_root / "global.npmrc").touch(mode=0o600)
return subprocess.run( # noqa: S603
[*_npx_prefix(npx, wallet_root), _LINK_CLI_PACKAGE, *arguments],
check=False,
capture_output=capture_output,
text=True,
env=_wallet_environment(),
cwd=wallet_root,
timeout=timeout,
)
def _link_wallet_authenticated(npx: str) -> bool:
"""Report whether a Link wallet is already connected to this machine."""
try:
result = _run_link_cli(
npx,
["auth", "status", "--format", "json"],
capture_output=True,
timeout=_LINK_LOGIN_TIMEOUT_S,
)
except (OSError, subprocess.SubprocessError):
return False
try:
payload = json.loads(result.stdout or "null")
except (TypeError, ValueError):
return False
if isinstance(payload, list):
payload = payload[0] if payload else None
return bool(isinstance(payload, dict) and payload.get("authenticated"))
def _prepare_link_wallet(console: Console, npx: str, *, as_json: bool) -> str | None:
"""Connect a Link wallet when none is present. Return an error message on failure."""
if _link_wallet_authenticated(npx):
return None
manual_setup = (
"Payment needs a Stripe Link wallet. Run `strix cloud billing topup` in an "
"interactive terminal to connect one, or set up the wallet at "
"https://link.com/agents. For a browser checkout instead, run "
"`strix cloud billing subscribe --plan strix_top_up`."
)
if as_json or not (sys.stdin.isatty() and sys.stdout.isatty()):
return manual_setup
console.print(
"[yellow]No Stripe Link wallet is connected.[/] Strix starts the Link sign-in now. "
"Approve the connection in the Link app, then Strix continues the payment. "
"The user approves every payment in the Link app."
)
try:
_run_link_cli(
npx,
[
"auth",
"login",
"--client-name",
_LINK_CLI_CLIENT_NAME,
"--interval",
"3",
"--timeout",
str(_LINK_LOGIN_TIMEOUT_S),
],
capture_output=False,
timeout=_LINK_LOGIN_TIMEOUT_S + 30,
)
except (OSError, subprocess.SubprocessError):
return manual_setup
if _link_wallet_authenticated(npx):
return None
return manual_setup
def _wallet_environment() -> dict[str, str]:
"""Pass only platform essentials and explicit wallet variables to npm/mppx."""
environment = {
name: value
for name, value in os.environ.items()
if name in _WALLET_ENV_NAMES or name.startswith("MPPX_")
if name in _WALLET_ENV_NAMES or name.startswith(("LINK_", "MPPX_"))
}
for name in ("NO_PROXY", "no_proxy"):
entries = [entry.strip() for entry in environment.get(name, "").split(",") if entry.strip()]

View File

@@ -2,8 +2,8 @@
The ``mppx`` CLI accepts custom HTTP headers through ``-H`` only. Passing a
Strix API token that way exposes it to process-listing tools. This module keeps
the token in the Strix process and injects it while forwarding the wallet's two
requests (challenge and paid retry) to the fixed billing endpoint.
the token in the Strix process and injects it while forwarding the wallet's few
requests (challenge probes and the paid retry) to the fixed billing endpoint.
"""
from __future__ import annotations
@@ -25,7 +25,7 @@ if TYPE_CHECKING:
_DEFAULT_REQUEST_TIMEOUT_S = 120.0
_MAX_REQUEST_BODY_BYTES = 64 * 1024
_MAX_UPSTREAM_RESPONSE_BYTES = 1024 * 1024
_MAX_WALLET_REQUESTS = 2
_MAX_WALLET_REQUESTS = 3
_HOP_BY_HOP_HEADERS = frozenset(
{
"connection",
@@ -54,7 +54,7 @@ class _BridgeState:
lock: threading.Lock = field(default_factory=threading.Lock)
def claim_request(self) -> bool:
"""Allow only the challenge request and its paid retry."""
"""Allow only the challenge probes and the one paid retry."""
with self.lock:
if self.request_count >= _MAX_WALLET_REQUESTS:
return False

View File

@@ -678,8 +678,9 @@ def _build_parser(group: str, verb_label: str, cmd: Cmd) -> argparse.ArgumentPar
default=None,
metavar="PM_ID",
help=(
"Stripe payment method for the card payment. "
"Defaults to MPPX_STRIPE_PAYMENT_METHOD."
"Pay with the mppx wallet client and this Stripe payment method "
"instead of the Stripe Link wallet. Defaults to "
"MPPX_STRIPE_PAYMENT_METHOD."
),
)
if cmd.path == "/scans" and cmd.method == "POST":

View File

@@ -194,3 +194,28 @@ func TestVulnerabilityReportRendersCalibrationFields(t *testing.T) {
"Fix Verification", "bypass review reasoned only",
)
}
func TestVulnerabilityReportUpdateRendersReportAndReason(t *testing.T) {
out := ansi.Strip(Tool(tool("update_vulnerability_report",
map[string]any{
"report_id": "vuln-0009",
"update_reason": "built a working unauthenticated file write against the endpoint",
"poc_script_code": "curl -X PATCH https://target/files/uuid",
},
map[string]any{
"success": true,
"action": "updated",
"report_id": "vuln-0009",
"severity": "critical",
"cvss_score": 9.3,
"updated_fields": []any{"poc_script_code"},
},
"completed")))
requireContains(t, out,
"Vulnerability Report Updated",
"vuln-0009",
"built a working unauthenticated file write",
"CRITICAL",
"9.3",
)
}

View File

@@ -84,6 +84,8 @@ func Tool(data map[string]any) string {
return renderViewImage(args, result)
case "create_vulnerability_report":
return renderVulnerabilityReport(args, result)
case "update_vulnerability_report":
return renderVulnerabilityReportUpdate(args, result)
case "create_dependency_report":
return renderDependencyReport(args, result)
case "list_reports":

View File

@@ -12,15 +12,27 @@ import (
// ---------------------------------------------------------------------------
func renderVulnerabilityReport(args map[string]any, result any) string {
return renderReport(args, result, "Vulnerability Report", "Creating report...")
}
// A revision names the report it changes and carries only the fields it
// replaces, so it renders the same sections with the ones it left alone absent.
func renderVulnerabilityReportUpdate(args map[string]any, result any) string {
return renderReport(args, result, "Vulnerability Report Updated", "Updating report...")
}
func renderReport(args map[string]any, result any, heading, pending string) string {
resultMap, _ := result.(map[string]any)
var b strings.Builder
b.WriteString("🐞 " + Bold(ReportHdr).Render("Vulnerability Report"))
b.WriteString("🐞 " + Bold(ReportHdr).Render(heading))
field := func(label, value string) {
if value != "" {
b.WriteString("\n\n" + Bold(Field).Render(label+": ") + value)
}
}
reportID := StringValue(args["report_id"])
field("Report", reportID)
title := StringValue(args["title"])
field("Title", title)
@@ -59,6 +71,7 @@ func renderVulnerabilityReport(args map[string]any, result any) string {
}
}
section("Reason", StringValue(args["update_reason"]))
section("Description", StringValue(args["description"]))
section("Impact", StringValue(args["impact"]))
section("Technical Analysis", StringValue(args["technical_analysis"]))
@@ -76,8 +89,8 @@ func renderVulnerabilityReport(args map[string]any, result any) string {
// was verified belongs next to it rather than in the artifact alone.
section("Fix Verification", StringValue(args["fix_verification"]))
if title == "" {
b.WriteString("\n " + Dim().Render("Creating report..."))
if title == "" && reportID == "" {
b.WriteString("\n " + Dim().Render(pending))
}
return "\n\n" + b.String() + "\n\n"
}

View File

@@ -102,6 +102,9 @@ class GoTuiRuntime:
self.report_state.vulnerability_found_callback = lambda _report: (
self.controller.notify_changed()
)
self.report_state.vulnerability_updated_callback = lambda _report: (
self.controller.notify_changed()
)
self.controller.notify_changed()
async def start_from_setup(self, verify: bool = True) -> None:

View File

@@ -116,7 +116,7 @@ const CATEGORY_TOOLS: Record<ToolCategory, readonly string[]> = {
filesystem: ["apply_patch", "view_image", "str_replace_editor", "list_files", "search_files"],
// Caido proxy tools (legacy: send_request)
proxy: ["list_requests", "view_request", "repeat_request", "list_sitemap", "view_sitemap_entry", "scope_rules", "send_request"],
reporting: ["create_vulnerability_report", "list_reports", "get_report"],
reporting: ["create_vulnerability_report", "update_vulnerability_report", "list_reports", "get_report"],
thinking: ["think"],
agents: ["create_agent", "agent_finish", "send_message_to_agent", "wait_for_agents", "view_agent_graph", "stop_agent"],
search: ["web_search"],

View File

@@ -44,6 +44,56 @@ def _strix_version() -> str | None:
return None
# Content a revision may replace. The identity of the finding (id, timestamp,
# finding_class) and its original author stay put. dependency_metadata is
# replaced whole, so a caller carries the package identity over itself.
UPDATABLE_REPORT_FIELDS = frozenset(
{
"title",
"dependency_metadata",
"severity",
"description",
"impact",
"target",
"technical_analysis",
"poc_description",
"poc_script_code",
"remediation_steps",
"evidence",
"assumptions",
"counterevidence",
"confidence",
"confidence_rationale",
"severity_change_conditions",
"fix_effort",
"cvss",
"cvss_breakdown",
"endpoint",
"method",
"cve",
"cwe",
"code_locations",
"fix_verification",
"fix_pr_body",
}
)
_LOWERCASE_REPORT_FIELDS = frozenset({"severity", "confidence", "fix_effort"})
# Fields that only describe another field. A revision may raise the rating or
# replace the locations without restating the reasoning behind the old one, and
# that leftover reasoning then contradicts the finding it annotates
# ("confidence: high" beside a rationale calling the evidence unconfirmed). When
# the field they describe changes and the update carries no replacement, they
# are dropped rather than kept.
_DEPENDENT_REPORT_FIELDS: dict[str, tuple[str, ...]] = {
"confidence": ("confidence_rationale",),
"severity": ("severity_change_conditions",),
"cvss": ("cvss_breakdown",),
"code_locations": ("fix_verification",),
}
def _clean_title(title: str) -> str:
"""Return a single-line finding title.
@@ -169,6 +219,7 @@ class ReportState:
self.caido_url: str | None = None
self.vulnerability_found_callback: Callable[[dict[str, Any]], None] | None = None
self.vulnerability_updated_callback: Callable[[dict[str, Any]], None] | None = None
self._sarif_repo_ctx: dict[str, Any] | None = None
self._sarif_repo_ctx_ready: bool = False
@@ -236,6 +287,12 @@ class ReportState:
)
self.vulnerability_reports = [r for r in data if isinstance(r, dict)]
for r in self.vulnerability_reports:
# A finding written before the class was persisted still carries the
# metadata of its class, so name the class it always had.
if not r.get("finding_class"):
r["finding_class"] = (
"dependency_cve" if r.get("dependency_metadata") else "dynamic"
)
title = r.get("title")
stale_md = False
if isinstance(title, str):
@@ -357,6 +414,100 @@ class ReportState:
self.save_run_data()
return report_id
def update_vulnerability_report(
self,
report_id: str,
fields: dict[str, Any],
*,
update_reason: str | None = None,
updated_by_agent_id: str | None = None,
updated_by_agent_name: str | None = None,
) -> dict[str, Any] | None:
"""Apply a revision to an existing report, keeping its id.
A field that only describes a field this update replaces is dropped when
the update carries no replacement for it, so the revised report cannot
state a new rating beside the superseded reasoning for the old one.
Returns the revised report, or ``None`` when the id is unknown or when
nothing in ``fields`` changes it.
"""
report = next((r for r in self.vulnerability_reports if r.get("id") == report_id), None)
if report is None:
logger.warning("cannot update unknown vulnerability report %s", report_id)
return None
changed: dict[str, Any] = {}
for key, raw_value in fields.items():
if key not in UPDATABLE_REPORT_FIELDS or raw_value is None:
continue
value = raw_value
if isinstance(value, str):
value = _clean_title(value) if key == "title" else value.strip()
if key in _LOWERCASE_REPORT_FIELDS:
value = value.lower()
if not value:
continue
if report.get(key) == value:
continue
changed[key] = value
superseded = {
dependent
for primary, dependents in _DEPENDENT_REPORT_FIELDS.items()
if primary in changed
for dependent in dependents
if dependent not in changed and report.get(dependent) not in (None, "", [], {})
}
if not changed and not superseded:
logger.info("update for %s carried no new content; keeping it as is", report_id)
return None
entry: dict[str, Any] = {
"timestamp": datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S UTC"),
"fields": sorted(changed),
}
if superseded:
entry["dropped_fields"] = sorted(superseded)
if update_reason and update_reason.strip():
entry["reason"] = update_reason.strip()[:500]
if updated_by_agent_id:
entry["agent_id"] = updated_by_agent_id
if updated_by_agent_name:
entry["agent_name"] = updated_by_agent_name
for key in ("severity", "cvss", "confidence"):
if key in changed and report.get(key) is not None:
entry[f"previous_{key}"] = report[key]
raw_history = report.get("update_history")
history: list[dict[str, Any]] = (
[e for e in raw_history if isinstance(e, dict)] if isinstance(raw_history, list) else []
)
history.append(entry)
report.update(changed)
for dependent in superseded:
report.pop(dependent, None)
report["update_history"] = history
report["updated_at"] = entry["timestamp"]
# The markdown on disk still shows the superseded evidence, so let the
# writer re-render it.
self._saved_vuln_ids.discard(report_id)
logger.info(
"Updated vulnerability report %s (%s)",
report_id,
", ".join(entry["fields"]) or "no field replaced",
)
if self.vulnerability_updated_callback:
self.vulnerability_updated_callback(report)
self.save_run_data()
return report
def get_existing_vulnerabilities(self) -> list[dict[str, Any]]:
return list(self.vulnerability_reports)

View File

@@ -356,4 +356,41 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
lines.append(str(report["assumptions"]))
lines.append("")
lines.extend(render_update_history(report.get("update_history")))
return "\n".join(lines)
def render_update_history(history: Any) -> list[str]:
"""Render the audit trail of every revision a report has received."""
if not isinstance(history, list):
return []
entries: list[dict[str, Any]] = [
cast("dict[str, Any]", e) for e in history if isinstance(e, dict)
]
if not entries:
return []
lines = ["## Update History\n"]
for entry in entries:
author = str(entry.get("agent_name") or entry.get("agent_id") or "an agent")
raw_fields = entry.get("fields")
fields: list[Any] = raw_fields if isinstance(raw_fields, list) else []
changed = ", ".join(str(field) for field in fields)
timestamp = str(entry.get("timestamp") or "unknown")
lines.append(f"**{timestamp}** — {author} updated: {changed}")
raw_dropped = entry.get("dropped_fields")
if isinstance(raw_dropped, list) and raw_dropped:
dropped = ", ".join(str(field) for field in raw_dropped)
lines.append(f" Dropped as superseded: {dropped}")
for key, label in (
("previous_severity", "severity"),
("previous_cvss", "CVSS"),
("previous_confidence", "confidence"),
):
if entry.get(key) is not None:
lines.append(f" Previous {label}: {entry[key]}")
if entry.get("reason"):
lines.append(f" Reason: {entry['reason']}")
lines.append("")
return lines

View File

@@ -12,13 +12,17 @@ import json
import logging
import re
from pathlib import PurePosixPath
from typing import Any
from typing import TYPE_CHECKING, Any
from agents import RunContextWrapper, function_tool
from strix.tools.nullish import clean_optional
if TYPE_CHECKING:
from strix.report.state import ReportState
logger = logging.getLogger(__name__)
@@ -257,6 +261,329 @@ def _validate_fix_verification(
]
def _finding_class_of(report: dict[str, Any]) -> str:
"""Resolve the class of a stored finding.
A finding filed before ``finding_class`` was persisted still carries the
metadata of its class. A record with dependency metadata is a dependency
finding even when the field is absent, so read the metadata before falling
back to dynamic.
"""
declared = str(report.get("finding_class") or "").lower()
if declared:
return declared
if report.get("dependency_metadata"):
return "dependency_cve"
return "dynamic"
_UPDATE_TEXT_FIELDS = (
"title",
"description",
"impact",
"target",
"technical_analysis",
"poc_description",
"poc_script_code",
"remediation_steps",
"evidence",
"assumptions",
"counterevidence",
"confidence_rationale",
"severity_change_conditions",
"endpoint",
"method",
"fix_verification",
"fix_pr_body",
"contextual_cvss_reasoning",
)
def _collect_update_changes( # noqa: PLR0912
fields: dict[str, Any],
) -> tuple[dict[str, Any], list[str]]:
"""Validate the fields a revision replaces and return them with any errors."""
errors: list[str] = []
changes: dict[str, Any] = {}
for name in _UPDATE_TEXT_FIELDS:
value = clean_optional(fields.get(name))
if value is not None:
changes[name] = value
confidence = clean_optional(fields.get("confidence"))
if confidence is not None:
confidence = confidence.lower()
if confidence not in _VALID_CONFIDENCE:
errors.append(
f"Invalid confidence: {confidence!r}. Must be one of: {sorted(_VALID_CONFIDENCE)}"
)
else:
changes["confidence"] = confidence
fix_effort = clean_optional(fields.get("fix_effort"))
if fix_effort is not None:
fix_effort = fix_effort.lower()
if fix_effort not in _VALID_FIX_EFFORT:
errors.append(
f"Invalid fix_effort: {fix_effort!r}. Must be one of: {sorted(_VALID_FIX_EFFORT)}"
)
else:
changes["fix_effort"] = fix_effort
breakdown = fields.get("cvss_breakdown")
if breakdown is not None:
breakdown_errors = _validate_cvss_breakdown(breakdown)
errors.extend(breakdown_errors)
if not breakdown_errors:
try:
cvss_score, severity, _vector = _calculate_cvss(breakdown)
except ValueError as exc:
errors.append(str(exc))
else:
# The rating belongs to the vector, so a revised vector carries
# its own score and severity rather than leaving the old ones.
changes["cvss_breakdown"] = breakdown
changes["cvss"] = cvss_score
changes["severity"] = severity
raw_locations = fields.get("code_locations")
locations = _normalize_code_locations(raw_locations)
if locations:
errors.extend(_validate_code_locations(locations))
errors.extend(_validate_fix_verification(locations, changes.get("fix_verification")))
changes["code_locations"] = locations
elif raw_locations:
errors.append(
"code_locations were dropped as unusable - every location needs a relative "
"'file' and an integer 'start_line'"
)
cve, cwe, identifier_errors = _validate_identifiers(
clean_optional(fields.get("cve")), clean_optional(fields.get("cwe"))
)
errors.extend(identifier_errors)
if cve:
changes["cve"] = cve
if cwe:
changes["cwe"] = cwe
return changes, errors
# Evidence that only a dynamic finding carries. A dependency finding describes a
# package, not a request against an endpoint.
_DYNAMIC_ONLY_UPDATE_FIELDS = (
"endpoint",
"method",
"poc_description",
"poc_script_code",
)
# A dependency finding is rated in the context of the codebase that pins it, and
# that rating is only shown with the reasoning behind it.
_DEPENDENCY_ONLY_UPDATE_FIELDS = ("contextual_cvss_reasoning",)
def _reject_cross_class_revision(
report_id: str,
matched_class: str,
offending: list[str],
) -> dict[str, Any]:
logger.info(
"Revision of %s carries fields (%s) a %s finding does not hold; rejecting",
report_id,
", ".join(offending),
matched_class,
)
return {
"success": False,
"error": (
f"Report '{report_id}' is a {matched_class} finding, so it cannot carry "
f"{', '.join(offending)}. File your proof as its own vulnerability report "
"instead of writing it onto this one."
),
"report_id": report_id,
"finding_class": matched_class,
"rejected_fields": offending,
}
def _rate_dependency_revision(
report_id: str,
matched: dict[str, Any],
changes: dict[str, Any],
) -> dict[str, Any] | None:
"""Turn a replacement ``cvss_breakdown`` into the contextual rating of a dependency.
A dependency record keeps its rating as ``cvss``/``severity`` plus the
contextual breakdown, vector and reasoning inside ``dependency_metadata``.
The package identity in that metadata is copied over untouched. A new
breakdown needs its own reasoning. The reasoning alone can be corrected
when the record already carries the breakdown it explains.
"""
breakdown = changes.pop("cvss_breakdown", None)
reasoning = changes.pop("contextual_cvss_reasoning", None)
if breakdown is None and reasoning is None:
return None
metadata = dict(matched.get("dependency_metadata") or {})
if breakdown is None and not metadata.get("contextual_cvss_breakdown"):
return {
"success": False,
"error": "Validation failed",
"errors": [
"cvss_breakdown is required: this dependency finding carries no "
"contextual rating yet, so contextual_cvss_reasoning has nothing to explain"
],
"report_id": report_id,
}
if reasoning is None:
return {
"success": False,
"error": "Validation failed",
"errors": [
"contextual_cvss_reasoning is required: a dependency finding is re-rated "
"with the cvss_breakdown observed in this codebase together with the "
"reasoning a reader can check"
],
"report_id": report_id,
}
if breakdown is not None:
score, _severity, vector = _calculate_cvss(breakdown)
metadata["contextual_cvss_breakdown"] = breakdown
metadata["contextual_cvss_score"] = score
metadata["contextual_cvss_vector"] = vector
metadata["contextual_cvss_reasoning"] = reasoning[:_MAX_CONTEXTUAL_REASONING_CHARS]
changes["dependency_metadata"] = metadata
return None
def _fit_revision_to_class(
report_state: ReportState,
report_id: str,
changes: dict[str, Any],
) -> dict[str, Any] | None:
"""Keep a revision inside the class of the finding it names.
A finding keeps its class and the metadata that belongs to it. Writing an
exploit onto a dependency record would leave it carrying a package pin next
to a request against an endpoint, so the proof belongs in its own dynamic
finding instead. A dependency finding is still re-rated, through the
contextual CVSS it was filed with.
"""
matched = next(
(r for r in report_state.get_existing_vulnerabilities() if r.get("id") == report_id),
None,
)
if matched is None:
return None
matched_class = _finding_class_of(matched)
foreign = (
_DEPENDENCY_ONLY_UPDATE_FIELDS
if matched_class == "dynamic"
else _DYNAMIC_ONLY_UPDATE_FIELDS
)
offending = [name for name in foreign if name in changes]
if offending:
return _reject_cross_class_revision(report_id, matched_class, offending)
if matched_class == "dynamic":
return None
return _rate_dependency_revision(report_id, matched, changes)
def _read_revision(
report_id: str, update_reason: str, fields: dict[str, Any]
) -> tuple[dict[str, Any], dict[str, Any] | None]:
"""Return the changes a revision asks for, or the reason it cannot be acted on."""
if not report_id or not str(update_reason or "").strip():
missing = "report_id" if not report_id else "update_reason"
return {}, {
"success": False,
"error": (
f"{missing} cannot be empty - name the report you are revising and state "
"what you learned that it does not yet carry"
),
}
changes, errors = _collect_update_changes(fields)
if errors:
return {}, {"success": False, "error": "Validation failed", "errors": errors}
if not changes:
return {}, {
"success": False,
"error": "No fields to update - pass at least one field you want to replace",
}
return changes, None
def _do_update(
*,
report_id: str,
update_reason: str,
fields: dict[str, Any],
agent_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
"""Apply an agent's own revision to a report it can name.
Editing a finding is its own operation and the only way a filed finding
changes. Deduplication never reaches this path: it only decides whether a
new candidate is a finding already on file.
"""
report_id = (report_id or "").strip()
changes, rejection = _read_revision(report_id, update_reason, fields)
if rejection is not None:
return rejection
from strix.report.state import get_global_report_state
report_state = get_global_report_state()
if report_state is None:
return {
"success": False,
"error": "Report state unavailable - no reports have been filed yet",
}
class_error = _fit_revision_to_class(report_state, report_id, changes)
if class_error is not None:
return class_error
updated = report_state.update_vulnerability_report(
report_id,
changes,
update_reason=update_reason,
updated_by_agent_id=agent_id,
updated_by_agent_name=agent_name,
)
if updated is None:
known = [r.get("id") for r in report_state.get_existing_vulnerabilities()]
if report_id not in known:
error = f"Report with id '{report_id}' not found"
else:
error = f"Report '{report_id}' already says this - nothing in your update changes it"
return {"success": False, "error": error, "report_id": report_id}
logger.info(
"Vulnerability report %s revised by its author: severity=%s cvss=%s fields=%s",
report_id,
updated.get("severity"),
updated.get("cvss"),
", ".join(sorted(changes)),
)
return {
"success": True,
"action": "updated",
"message": f"Report '{report_id}' now carries your revision. Do not file it again.",
"report_id": report_id,
"updated_fields": sorted(changes),
"severity": updated.get("severity"),
"cvss_score": updated.get("cvss"),
}
async def _do_create(
*,
title: str,
@@ -359,9 +686,37 @@ async def _do_create(
"endpoint": endpoint,
"method": method,
}
report_fields: dict[str, Any] = {
"title": title,
"description": description,
"severity": severity,
"impact": impact,
"target": target,
"technical_analysis": technical_analysis,
"poc_description": poc_description,
"poc_script_code": poc_script_code,
"remediation_steps": remediation_steps,
"evidence": evidence,
"assumptions": assumptions,
"counterevidence": counterevidence,
"confidence": confidence,
"confidence_rationale": confidence_rationale,
"severity_change_conditions": severity_change_conditions,
"fix_effort": fix_effort,
"cvss": cvss_score,
"cvss_breakdown": cvss_breakdown,
"endpoint": endpoint,
"method": method,
"cve": cve,
"cwe": cwe,
"code_locations": parsed_locations,
"fix_verification": fix_verification,
"fix_pr_body": fix_pr_body,
}
dedupe = await check_duplicate(candidate, existing)
if dedupe.get("is_duplicate"):
duplicate_id = dedupe.get("duplicate_id", "")
duplicate_id = str(dedupe.get("duplicate_id") or "")
duplicate_title = next(
(r.get("title", "Unknown") for r in existing if r.get("id") == duplicate_id),
"",
@@ -379,31 +734,7 @@ async def _do_create(
}
report_id = report_state.add_vulnerability_report(
title=title,
description=description,
severity=severity,
impact=impact,
target=target,
technical_analysis=technical_analysis,
poc_description=poc_description,
poc_script_code=poc_script_code,
remediation_steps=remediation_steps,
evidence=evidence,
assumptions=assumptions,
counterevidence=counterevidence,
confidence=confidence,
confidence_rationale=confidence_rationale,
severity_change_conditions=severity_change_conditions,
fix_effort=fix_effort,
cvss=cvss_score,
cvss_breakdown=cvss_breakdown,
endpoint=endpoint,
method=method,
cve=cve,
cwe=cwe,
code_locations=parsed_locations,
fix_verification=fix_verification,
fix_pr_body=fix_pr_body,
**report_fields,
agent_id=agent_id if isinstance(agent_id, str) else None,
agent_name=agent_name if isinstance(agent_name, str) else None,
)
@@ -512,7 +843,9 @@ async def create_vulnerability_report(
Automatic LLM-based **deduplication** rejects reports that describe
the same root cause on the same asset as an existing report. If you
get a ``duplicate_of`` response, do NOT retry — move on to other
areas.
areas. When you have learned something a filed finding does not yet
carry, revise that finding with ``update_vulnerability_report``
instead of filing this report again.
**Counterevidence pass (required before filing)**: actively build the
strongest case that this finding is NOT exploitable, or less severe
@@ -886,6 +1219,147 @@ async def create_vulnerability_report(
return json.dumps(result, ensure_ascii=False, default=str)
@function_tool(timeout=60, strict_mode=False)
async def update_vulnerability_report(
ctx: RunContextWrapper,
report_id: str,
update_reason: str,
title: str | None = None,
description: str | None = None,
impact: str | None = None,
target: str | None = None,
technical_analysis: str | None = None,
poc_description: str | None = None,
poc_script_code: str | None = None,
remediation_steps: str | None = None,
evidence: str | None = None,
assumptions: str | None = None,
counterevidence: str | None = None,
confidence: str | None = None,
confidence_rationale: str | None = None,
severity_change_conditions: str | None = None,
fix_effort: str | None = None,
cvss_breakdown: dict[str, str] | None = None,
endpoint: str | None = None,
method: str | None = None,
cve: str | None = None,
cwe: str | None = None,
code_locations: list[dict[str, Any]] | None = None,
fix_verification: str | None = None,
fix_pr_body: str | None = None,
contextual_cvss_reasoning: str | None = None,
) -> str:
"""Revise a vulnerability report that is already filed, keeping its id.
Use this when you learn something a filed finding does not yet carry:
- You built the working exploit after filing the finding on static
evidence, so the PoC and the confidence change.
- You chained the finding with another one and the real impact is
higher, so the impact narrative and the CVSS vector change.
- Further testing narrowed or weakened the finding, so the severity
must come down.
- Counterevidence, remediation, or a code location was wrong or
incomplete.
This is not deduplication. You do not need a duplicate verdict to
revise your own finding, and you must not file a second report for a
finding you can revise. Call ``list_reports`` or ``get_report`` first
to find the id and read what the report already says.
Pass only the fields you want to replace. Every other field stays as
it is. Reporting rules of ``create_vulnerability_report`` apply to
every field you pass, including the markdown and tone rules.
Notes on specific fields:
- ``cvss_breakdown`` replaces the whole vector. The score and the
severity are recalculated from it, so pass all 8 metrics. On a
dependency finding it replaces the contextual rating and needs
``contextual_cvss_reasoning`` with it. Pass the reasoning alone to
correct only the explanation of the rating already on file.
- A dependency finding never carries ``endpoint``, ``method`` or a PoC.
File a proven exploit of the package as its own report.
- A field that only explains another field is dropped when the field
it explains changes and you pass no replacement. Pass
``confidence_rationale`` with a new ``confidence``, and
``severity_change_conditions`` with a new ``cvss_breakdown``.
- ``code_locations`` replaces the whole list. A location carrying
``fix_after`` needs ``fix_verification``.
The report keeps its id, its original author, and its filing time. The
revision is recorded in the report as update history, so state the
reason plainly.
Args:
report_id: Id of the report to revise (format ``vuln-NNNN``).
update_reason: What you learned that the report does not yet
carry, in one or two sentences.
title: Replacement title.
description: Replacement overview.
impact: Replacement impact narrative.
target: Replacement affected asset.
technical_analysis: Replacement technical details.
poc_description: Replacement PoC steps (no code).
poc_script_code: Replacement exploit script or payload.
remediation_steps: Replacement remediation prose (no code).
evidence: Replacement evidence.
assumptions: Replacement exploitability prerequisites.
counterevidence: Replacement case against the finding.
confidence: ``high`` / ``medium`` / ``low``.
confidence_rationale: The gap behind a confidence below ``high``.
severity_change_conditions: What would move the severity now.
fix_effort: ``trivial`` / ``low`` / ``medium`` / ``high``.
cvss_breakdown: All 8 CVSS metrics. Replaces the score and the
severity too.
endpoint: Replacement endpoint.
method: Replacement HTTP method.
cve: Replacement CVE id.
cwe: Replacement CWE id.
code_locations: Replacement code locations.
fix_verification: Verification statement for an applyable fix.
fix_pr_body: Replacement fix PR body.
contextual_cvss_reasoning: Dependency findings only. What you
observed in this codebase that justifies the contextual
``cvss_breakdown``.
"""
agent_id, agent_name = _caller_identity(ctx)
result = await asyncio.to_thread(
_do_update,
report_id=report_id,
update_reason=update_reason,
fields={
"title": title,
"description": description,
"impact": impact,
"target": target,
"technical_analysis": technical_analysis,
"poc_description": poc_description,
"poc_script_code": poc_script_code,
"remediation_steps": remediation_steps,
"evidence": evidence,
"assumptions": assumptions,
"counterevidence": counterevidence,
"confidence": confidence,
"confidence_rationale": confidence_rationale,
"severity_change_conditions": severity_change_conditions,
"fix_effort": fix_effort,
"cvss_breakdown": cvss_breakdown,
"endpoint": endpoint,
"method": method,
"cve": cve,
"cwe": cwe,
"code_locations": code_locations,
"fix_verification": fix_verification,
"fix_pr_body": fix_pr_body,
"contextual_cvss_reasoning": contextual_cvss_reasoning,
},
agent_id=agent_id,
agent_name=agent_name,
)
return json.dumps(result, ensure_ascii=False, default=str)
_DEP_SEVERITY_FROM_CVSS = {
(9.0, 10.0): "critical",
(7.0, 9.0): "high",

View File

@@ -116,7 +116,7 @@ def test_bridge_forwards_only_the_approved_request_and_protected_headers(
assert len(captured) == 1
def test_bridge_allows_only_two_valid_wallet_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
def test_bridge_limits_valid_wallet_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
calls = 0
def fake_request(*_args: Any, **_kwargs: Any) -> _StreamingResponse:
@@ -132,8 +132,9 @@ def test_bridge_allows_only_two_valid_wallet_attempts(monkeypatch: pytest.Monkey
) as wallet_url:
assert _post(wallet_url, b"{}") == b"{}"
assert _post(wallet_url, b"{}") == b"{}"
with pytest.raises(urllib.error.HTTPError) as third_request:
assert _post(wallet_url, b"{}") == b"{}"
with pytest.raises(urllib.error.HTTPError) as extra_request:
_post(wallet_url, b"{}")
assert third_request.value.code == 429
assert calls == 2
assert extra_request.value.code == 429
assert calls == payment_proxy._MAX_WALLET_REQUESTS

118
tests/test_cloud_wallet.py Normal file
View File

@@ -0,0 +1,118 @@
"""Tests for the Stripe Link wallet setup path of `strix cloud billing topup`."""
from __future__ import annotations
import subprocess
from typing import TYPE_CHECKING, Any
from rich.console import Console
from strix.interface.cloud import billing
if TYPE_CHECKING:
import pytest
_MIN_LINK_CONTEXT_CHARS = 100
def _completed(stdout: str) -> subprocess.CompletedProcess[str]:
return subprocess.CompletedProcess(args=["link-cli"], returncode=0, stdout=stdout, stderr="")
def test_payment_context_is_long_enough_for_link_approval() -> None:
context = billing._payment_context({"credits": 5})
assert len(context) >= _MIN_LINK_CONTEXT_CHARS
assert "5" in context
def test_mppx_wallet_configured_follows_environment(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("MPPX_ACCOUNT", raising=False)
monkeypatch.delenv("MPPX_STRIPE_SECRET_KEY", raising=False)
assert billing._mppx_wallet_configured() is False
monkeypatch.setenv("MPPX_ACCOUNT", "agent")
assert billing._mppx_wallet_configured() is True
def test_link_wallet_authenticated_reads_status_list(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
billing,
"_run_link_cli",
lambda *_args, **_kwargs: _completed('[{"authenticated": true}]'),
)
assert billing._link_wallet_authenticated("npx") is True
def test_link_wallet_authenticated_handles_unusable_output(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(billing, "_run_link_cli", lambda *_args, **_kwargs: _completed("not json"))
assert billing._link_wallet_authenticated("npx") is False
def test_link_wallet_authenticated_handles_launch_failure(
monkeypatch: pytest.MonkeyPatch,
) -> None:
def explode(*_args: Any, **_kwargs: Any) -> subprocess.CompletedProcess[str]:
raise OSError
monkeypatch.setattr(billing, "_run_link_cli", explode)
assert billing._link_wallet_authenticated("npx") is False
def test_pending_spend_request_reads_the_created_record() -> None:
stdout = (
'[{"id": "lsrq_123", "status": "pending_approval", '
'"approval_url": "https://app.link.com/activity/approve/lsrq_123"}]'
)
assert billing._pending_spend_request(stdout) == (
"lsrq_123",
"https://app.link.com/activity/approve/lsrq_123",
)
assert billing._pending_spend_request('[{"id": "lsrq_1", "status": "approved"}]') is None
assert billing._pending_spend_request("not json") is None
def test_pending_spend_request_tolerates_banner_text_around_pretty_json() -> None:
stdout = (
"Update available for @stripe/link-cli: 0.13.1 -> 0.16.0\n"
"[\n {\n"
' "id": "lsrq_9",\n'
' "status": "pending_approval",\n'
' "approval_url": "https://app.link.com/activity/approve/lsrq_9"\n'
" }\n]"
)
assert billing._pending_spend_request(stdout) == (
"lsrq_9",
"https://app.link.com/activity/approve/lsrq_9",
)
def test_final_spend_request_status_reads_the_last_poll_line() -> None:
stdout = '{"status": "pending_approval"}\n{"status": "approved"}\n'
assert billing._final_spend_request_status(stdout) == "approved"
assert billing._final_spend_request_status("") is None
def test_final_spend_request_status_unwraps_chunk_envelopes() -> None:
stdout = (
'{"type":"chunk","data":{"id":"lsrq_9","status":"pending_approval"}}\n'
'{"type":"chunk","data":{"id":"lsrq_9","status":"approved"}}\n'
'{"type":"done","ok":true,"meta":{"command":"spend-request retrieve"}}\n'
)
assert billing._final_spend_request_status(stdout) == "approved"
def test_prepare_link_wallet_skips_login_when_connected(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(billing, "_link_wallet_authenticated", lambda _npx: True)
assert billing._prepare_link_wallet(Console(), "npx", as_json=True) is None
def test_prepare_link_wallet_explains_setup_without_a_terminal(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(billing, "_link_wallet_authenticated", lambda _npx: False)
message = billing._prepare_link_wallet(Console(), "npx", as_json=True)
assert message is not None
assert "https://link.com/agents" in message

View File

@@ -73,6 +73,41 @@ def test_hydrate_from_run_dir_strips_control_chars_from_title(
assert md_path.read_text(encoding="utf-8").startswith("# XSS in search form\n")
def test_hydrate_names_the_class_a_legacy_record_always_had(
report_state: ReportState,
) -> None:
# A run started before the class was persisted still holds the package metadata
# of a dependency finding, and resume must not read it as a dynamic one.
(report_state.get_run_dir() / "vulnerabilities.json").write_text(
json.dumps(
[
{
"id": "vuln-0001",
"title": "Directus 11.5.1 is affected by CVE-2025-55746",
"severity": "medium",
"timestamp": "2026-01-01 00:00:00 UTC",
"dependency_metadata": {
"package_name": "directus",
"installed_version": "11.5.1",
},
},
{
"id": "vuln-0002",
"title": "Reflected XSS in search",
"severity": "medium",
"timestamp": "2026-01-01 00:00:00 UTC",
},
]
),
encoding="utf-8",
)
report_state.hydrate_from_run_dir()
assert report_state.vulnerability_reports[0]["finding_class"] == "dependency_cve"
assert report_state.vulnerability_reports[1]["finding_class"] == "dynamic"
def _seed(state: ReportState) -> None:
state.add_vulnerability_report(
title="Reflected XSS in search",

View File

@@ -16,8 +16,10 @@ from strix.tools.finish.tool import finish_scan
from strix.tools.reporting.tool import (
_do_create,
_do_create_dependency,
_do_update,
create_dependency_report,
create_vulnerability_report,
update_vulnerability_report,
)
@@ -1251,3 +1253,507 @@ async def test_dependency_report_rejects_contextual_breakdown_without_reasoning(
assert result["success"] is False
assert any("contextual_cvss_reasoning is required" in error for error in result["errors"])
assert report_state.vulnerability_reports == []
_CONFIRMED_KWARGS: dict[str, Any] = {
"title": "Unauthenticated file write on /files/{id}",
"description": "A multipart PATCH writes attacker content before the permission check.",
"impact": "Any anonymous user overwrites stored files and serves attacker content.",
"target": "https://cms.example.com",
"technical_analysis": "disk.write runs before the authorization guard.",
"poc_description": "1. PATCH /files/<uuid> with a multipart body as an anonymous user.",
"poc_script_code": "PATCH /files/2f1c HTTP/1.1\n\n--x\nowned\n--x--",
"remediation_steps": "Authorize before the write.",
"evidence": "The stored file returns the injected payload after the 403 response.",
"assumptions": "Assumes the uuid of one existing file is known.",
"counterevidence": "The endpoint answers 403, yet the write already landed.",
"confidence": "HIGH",
"confidence_rationale": "The write was observed end to end against the live host.",
"severity_change_conditions": "A guard before disk.write would remove the impact.",
"fix_effort": "MEDIUM",
"cvss_breakdown": _CVSS,
"endpoint": "/files/{id}",
"method": "PATCH",
"cve": "CVE-2025-55746",
"cwe": "CWE-863",
"code_locations": None,
}
def _seed_weak_report(report_state: ReportState) -> None:
"""A version-based, unproven entry for the same issue, as an earlier agent files it."""
report_state.vulnerability_reports.append(
{
"id": "vuln-0009",
"title": "Directus 11.5.1 exposed on public host (in scope for CVE-2025-55746)",
"severity": "medium",
"timestamp": "2026-01-01 00:00:00 UTC",
"description": "The banner reports a version affected by CVE-2025-55746.",
"target": "https://cms.example.com",
"confidence": "low",
"evidence": "The version banner only.",
"cvss": 5.3,
"finding_class": "dynamic",
"agent_id": "aaaa1111",
}
)
report_state._saved_vuln_ids.add("vuln-0009")
async def test_duplicate_verdict_rejects_without_touching_the_existing_report(
report_state: ReportState, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Deduplication only answers identity. A duplicate is rejected and points at the
finding it matched; revising that finding is a separate, explicit operation."""
_seed_weak_report(report_state)
async def fake_check_duplicate(
_candidate: dict[str, Any], _existing: list[dict[str, Any]]
) -> dict[str, Any]:
return {
"is_duplicate": True,
"duplicate_id": "vuln-0009",
"confidence": 0.9,
"reason": "Same root cause on the same endpoint.",
}
monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate)
result = await _do_create(**_CONFIRMED_KWARGS, agent_id="834f79fb", agent_name="Validation")
assert result["success"] is False
assert result["duplicate_of"] == "vuln-0009"
assert "action" not in result
assert len(report_state.vulnerability_reports) == 1
report = report_state.vulnerability_reports[0]
assert report["severity"] == "medium", "a duplicate verdict never edits the matched finding"
assert "poc_script_code" not in report
assert "update_history" not in report
def test_update_vulnerability_report_records_chained_impact(report_state: ReportState) -> None:
"""Attack chaining raises the impact of a finding already on file."""
_seed_weak_report(report_state)
updated = report_state.update_vulnerability_report(
"vuln-0009",
{
"severity": "CRITICAL",
"cvss": 9.8,
"impact": "The overwritten file loads in an admin session and takes over the account.",
"id": "vuln-9999",
"finding_class": "static",
},
update_reason="A chained admin takeover follows the file write.",
)
assert updated is not None
assert updated["id"] == "vuln-0009", "identity fields are not updatable"
assert updated["finding_class"] == "dynamic"
assert updated["severity"] == "critical"
assert updated["updated_at"]
assert report_state.update_vulnerability_report("vuln-0404", {"severity": "high"}) is None
def test_update_vulnerability_report_ignores_identical_content(report_state: ReportState) -> None:
_seed_weak_report(report_state)
assert report_state.update_vulnerability_report("vuln-0009", {"severity": "medium"}) is None
assert "update_history" not in report_state.vulnerability_reports[0]
def test_update_drops_reasoning_left_behind_by_the_field_it_describes(
report_state: ReportState,
) -> None:
"""A rating the update replaces must not keep the rationale for the old one."""
_seed_weak_report(report_state)
report = report_state.vulnerability_reports[0]
report["confidence_rationale"] = "Nothing was executed; the version banner is the only signal."
report["cvss_breakdown"] = {"attack_vector": "network", "user_interaction": "required"}
report["severity_change_conditions"] = "Confirming the write would raise this."
updated = report_state.update_vulnerability_report(
"vuln-0009",
{
"confidence": "high",
"severity": "critical",
"cvss": 9.8,
"severity_change_conditions": "A guard before the write would remove the impact.",
},
)
assert updated is not None
assert "confidence_rationale" not in updated, "the superseded rationale must not survive"
assert "cvss_breakdown" not in updated
assert updated["severity_change_conditions"].startswith("A guard"), (
"a replacement the update supplies is kept, not dropped"
)
assert updated["update_history"][0]["dropped_fields"] == [
"confidence_rationale",
"cvss_breakdown",
]
run_dir = report_state._run_dir
assert run_dir is not None
markdown = (run_dir / "vulnerabilities" / "vuln-0009.md").read_text(encoding="utf-8")
assert "version banner is the only signal" not in markdown
assert "Dropped as superseded: confidence_rationale, cvss_breakdown" in markdown
def test_agent_revises_its_own_report_without_a_duplicate_verdict(
report_state: ReportState,
) -> None:
"""Editing a finding is its own operation: no dedupe verdict is involved."""
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="An unauthenticated PATCH wrote the file, so the finding is confirmed.",
fields={
"poc_script_code": "PATCH /files/2f1c HTTP/1.1",
"confidence": "HIGH",
"confidence_rationale": "The write was replayed twice.",
"cvss_breakdown": _CVSS,
"severity_change_conditions": "A guard before the write would remove the impact.",
},
agent_id="834f79fb",
agent_name="Directus CVE-2025-55746 Validation Agent",
)
assert result["success"] is True
assert result["action"] == "updated"
assert result["severity"] == "critical"
assert result["cvss_score"] == pytest.approx(9.8)
assert "cvss" in result["updated_fields"], "a new vector carries its own score"
assert len(report_state.vulnerability_reports) == 1
report = report_state.vulnerability_reports[0]
assert report["id"] == "vuln-0009"
assert report["confidence"] == "high"
assert report["agent_id"] == "aaaa1111", "the original reporter stays on the finding"
history = report["update_history"]
assert history[0]["agent_name"] == "Directus CVE-2025-55746 Validation Agent"
assert history[0]["reason"].startswith("An unauthenticated PATCH")
run_dir = report_state._run_dir
assert run_dir is not None
markdown = (run_dir / "vulnerabilities" / "vuln-0009.md").read_text(encoding="utf-8")
assert "PATCH /files/2f1c" in markdown
@pytest.mark.parametrize(
("report_id", "update_reason", "fields", "expected"),
[
(" ", "reason", {"impact": "x"}, "report_id cannot be empty"),
("vuln-0009", " ", {"impact": "x"}, "update_reason cannot be empty"),
("vuln-0009", "reason", {}, "No fields to update"),
("vuln-0404", "reason", {"impact": "x"}, "not found"),
],
)
def test_update_rejects_a_call_it_cannot_act_on(
report_state: ReportState,
report_id: str,
update_reason: str,
fields: dict[str, Any],
expected: str,
) -> None:
_seed_weak_report(report_state)
result = _do_update(report_id=report_id, update_reason=update_reason, fields=fields)
assert result["success"] is False
assert expected in result["error"]
assert "update_history" not in report_state.vulnerability_reports[0]
def test_update_reports_every_invalid_field_at_once(report_state: ReportState) -> None:
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Raising the rating.",
fields={
"confidence": "very high",
"fix_effort": "weeks",
"cvss_breakdown": {**_CVSS, "attack_vector": "X"},
"cve": "CVE-BAD",
},
)
assert result["success"] is False
joined = " ".join(result["errors"])
assert "confidence" in joined
assert "fix_effort" in joined
assert "attack_vector" in joined
assert "CVE" in joined
assert report_state.vulnerability_reports[0]["confidence"] == "low", "nothing was applied"
def test_update_wants_verification_for_a_fix_it_would_apply(
report_state: ReportState,
) -> None:
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Adding the file the write lands in.",
fields={
"code_locations": [
{
"file": "api/src/controllers/files.ts",
"start_line": 42,
"fix_before": "await storage.write(id, body)",
"fix_after": "await assertPermission(req); await storage.write(id, body)",
}
]
},
)
assert result["success"] is False
assert any("fix_verification" in error for error in result["errors"])
def test_update_says_so_when_the_report_already_carries_it(
report_state: ReportState,
) -> None:
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Restating the severity.",
fields={"confidence": "low"},
)
assert result["success"] is False
assert "already says this" in result["error"]
assert result["report_id"] == "vuln-0009"
def test_update_tool_asks_for_the_report_and_the_reason() -> None:
schema = update_vulnerability_report.params_json_schema
assert set(schema["required"]) >= {"report_id", "update_reason"}
assert "cvss_breakdown" in schema["properties"]
assert "id" not in schema["properties"], "identity fields are not editable"
description = update_vulnerability_report.description
assert "not deduplication" in description
def test_update_keeps_an_exploit_out_of_a_dependency_finding(report_state: ReportState) -> None:
"""A dependency record is rated from its advisory, so a revision must not write a
PoC and a dynamic rating onto it. The proof belongs in its own finding."""
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report["finding_class"] = "dependency_cve"
dependency_report["dependency_metadata"] = {
"package_name": "directus",
"installed_version": "11.5.1",
}
result = _do_update(
report_id="vuln-0009",
update_reason="An unauthenticated PATCH wrote the file.",
fields={
"poc_script_code": "PATCH /files/2f1c HTTP/1.1",
"endpoint": "/files/{uuid}",
"cvss_breakdown": _CVSS,
},
)
assert result["success"] is False
assert "dependency_cve" in result["error"]
assert set(result["rejected_fields"]) == {"endpoint", "poc_script_code"}
assert dependency_report["severity"] == "medium"
assert "poc_script_code" not in dependency_report
assert "update_history" not in dependency_report
def _seed_dependency_report(report_state: ReportState) -> dict[str, Any]:
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report["finding_class"] = "dependency_cve"
dependency_report["dependency_metadata"] = {
"package_name": "directus",
"installed_version": "11.5.1",
"manifest_path": "package-lock.json",
"advisory_cvss": 9.8,
"contextual_cvss_breakdown": {**_CVSS, "confidentiality": "L"},
"contextual_cvss_score": 5.3,
"contextual_cvss_vector": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:L/I:N/A:N",
"contextual_cvss_reasoning": "The vulnerable API is imported but never called.",
}
return dependency_report
def test_update_re_rates_a_dependency_finding_through_its_contextual_cvss(
report_state: ReportState,
) -> None:
"""A dependency finding is rated in the context of the codebase. A revised
breakdown replaces that contextual rating, with the reasoning a reader can
check, and leaves the package identity alone."""
dependency_report = _seed_dependency_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="A call path from the upload handler to the vulnerable API was found.",
fields={
"cvss_breakdown": _CVSS,
"contextual_cvss_reasoning": (
"routes/upload.ts:88 reaches the affected parser with user input."
),
},
)
assert result["success"] is True
assert result["severity"] == "critical"
assert dependency_report["severity"] == "critical"
assert dependency_report["cvss"] == 9.8
assert "cvss_breakdown" not in dependency_report
assert "contextual_cvss_reasoning" not in dependency_report
metadata = dependency_report["dependency_metadata"]
assert metadata["package_name"] == "directus"
assert metadata["installed_version"] == "11.5.1"
assert metadata["manifest_path"] == "package-lock.json"
assert metadata["advisory_cvss"] == 9.8
assert metadata["contextual_cvss_breakdown"] == _CVSS
assert metadata["contextual_cvss_score"] == 9.8
assert metadata["contextual_cvss_vector"] == "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"
assert metadata["contextual_cvss_reasoning"].startswith("routes/upload.ts:88")
assert dependency_report["finding_class"] == "dependency_cve"
history = dependency_report["update_history"]
assert history[-1]["previous_severity"] == "medium"
assert set(history[-1]["fields"]) == {"cvss", "dependency_metadata", "severity"}
def test_update_wants_the_reasoning_behind_a_dependency_re_rating(
report_state: ReportState,
) -> None:
dependency_report = _seed_dependency_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="The parser is reachable.",
fields={"cvss_breakdown": _CVSS},
)
assert result["success"] is False
assert any("contextual_cvss_reasoning" in error for error in result["errors"])
assert dependency_report["severity"] == "medium"
assert dependency_report["dependency_metadata"]["contextual_cvss_score"] == 5.3
assert "update_history" not in dependency_report
def test_update_corrects_the_reasoning_behind_a_dependency_rating_alone(
report_state: ReportState,
) -> None:
"""The rating on file stays; only its explanation is replaced."""
dependency_report = _seed_dependency_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="The reasoning named the wrong module.",
fields={"contextual_cvss_reasoning": "lib/parser.ts imports it; no call site reaches it."},
)
assert result["success"] is True
assert result["updated_fields"] == ["dependency_metadata"]
assert dependency_report["severity"] == "medium"
assert dependency_report["cvss"] == 5.3
metadata = dependency_report["dependency_metadata"]
assert metadata["contextual_cvss_breakdown"] == {**_CVSS, "confidentiality": "L"}
assert metadata["contextual_cvss_score"] == 5.3
assert metadata["contextual_cvss_reasoning"].startswith("lib/parser.ts")
assert metadata["package_name"] == "directus"
def test_update_wants_a_rating_before_reasoning_about_one(
report_state: ReportState,
) -> None:
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report["finding_class"] = "dependency_cve"
dependency_report["dependency_metadata"] = {
"package_name": "directus",
"installed_version": "11.5.1",
}
result = _do_update(
report_id="vuln-0009",
update_reason="Explaining the rating.",
fields={"contextual_cvss_reasoning": "Reachable."},
)
assert result["success"] is False
assert any("cvss_breakdown is required" in error for error in result["errors"])
assert "contextual_cvss_reasoning" not in dependency_report["dependency_metadata"]
def test_update_keeps_contextual_reasoning_off_a_dynamic_finding(
report_state: ReportState,
) -> None:
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Re-rating.",
fields={"cvss_breakdown": _CVSS, "contextual_cvss_reasoning": "Reachable."},
)
assert result["success"] is False
assert result["rejected_fields"] == ["contextual_cvss_reasoning"]
assert report_state.vulnerability_reports[0]["severity"] == "medium"
def test_update_reads_a_legacy_dependency_record_by_its_metadata(
report_state: ReportState,
) -> None:
"""A dependency finding filed before finding_class was persisted still carries
package metadata, so its class is read from that, not defaulted to dynamic."""
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report.pop("finding_class", None)
dependency_report["dependency_metadata"] = {
"package_name": "directus",
"installed_version": "11.5.1",
}
result = _do_update(
report_id="vuln-0009",
update_reason="An unauthenticated PATCH wrote the file.",
fields={"poc_script_code": "PATCH /files/2f1c HTTP/1.1", "cvss_breakdown": _CVSS},
)
assert result["success"] is False
assert "dependency_cve" in result["error"]
assert "poc_script_code" not in dependency_report
def test_update_still_corrects_the_prose_of_a_dependency_finding(
report_state: ReportState,
) -> None:
"""Fields every class carries stay editable on a dependency record."""
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report["finding_class"] = "dependency_cve"
result = _do_update(
report_id="vuln-0009",
update_reason="The advisory names a later fixed release than the report says.",
fields={"remediation_steps": "Upgrade to 11.5.2 or later."},
)
assert result["success"] is True
assert dependency_report["remediation_steps"] == "Upgrade to 11.5.2 or later."
assert dependency_report["finding_class"] == "dependency_cve"
def test_update_refuses_code_locations_it_cannot_use(report_state: ReportState) -> None:
"""A location without a usable file and line is reported, not dropped in silence."""
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Naming the vulnerable handler.",
fields={"code_locations": [{"label": "the file write"}]},
)
assert result["success"] is False
assert any("start_line" in error for error in result["errors"])

2
uv.lock generated
View File

@@ -2378,7 +2378,7 @@ wheels = [
[[package]]
name = "strix-agent"
version = "1.5.3"
version = "1.6.0"
source = { editable = "." }
dependencies = [
{ name = "caido-sdk-client" },