Compare commits

..
Author SHA1 Message Date
Ahmed Allam 1ec146d809 cli: render contextual CVSS vector, advisory score, and reasoning for dependency findings 2026-08-17 09:59:57 +00:00
Ahmed Allam e82ee2c31e Contextual CVSS as a full 8-metric breakdown, computed like a normal finding 2026-08-17 09:45:04 +00:00
Ahmed Allam 256df30d28 reporting: require the source-to-sink trace in reachability evidence, not just CVSS reasoning 2026-08-17 09:11:59 +00:00
Ahmed Allam 84163e8e3b reporting: surface contextual CVSS in the markdown report; require reasoning only for surviving metrics 2026-08-17 09:10:47 +00:00
Ahmed Allam 360c549278 reporting: drop per-metric contextual CVSS reasoning, keep the summary 2026-08-17 09:08:07 +00:00
Ahmed Allam 474fb25bf3 feat(reporting): contextual CVSS environmental metrics on dependency reports 2026-08-17 09:05:20 +00:00
yoni-at-strixandGitHub 8551339130 feat: place caller-provided files into the sandbox workspace (extra_files, --workspace-file) (#1085)
* add extra-files plumbing so orchestrators can drop single files into the sandbox workspace

* reject extra-file paths that collide with a local source tree

* add --workspace-file so CLI users can place files in the sandbox workspace

* reject repeated and control-character workspace paths

* revalidate persisted workspace files when resuming a run

* drop the workspace-file size limit
2026-08-14 16:43:08 -04:00
Alex SchapiroandAhmed Allam 8ca0c4a9b8 Fix LiteLLM cost model resolution 2026-08-12 17:26:00 +03:00
10 changed files with 628 additions and 55 deletions
+21
View File
@@ -133,6 +133,27 @@ def format_vulnerability_report(report: dict[str, Any]) -> Text: # noqa: PLR091
text.append("CVSS Vector: ", style=field_style)
text.append("/".join(cvss_parts), style="dim")
dependency_metadata = report.get("dependency_metadata") or {}
if dependency_metadata:
contextual_vector = dependency_metadata.get("contextual_cvss_vector")
if contextual_vector:
text.append("\n\n")
text.append("Contextual CVSS Vector: ", style=field_style)
text.append(contextual_vector, style="dim")
advisory_cvss = dependency_metadata.get("advisory_cvss")
if advisory_cvss is not None and advisory_cvss != report.get("cvss"):
text.append("\n\n")
text.append("Advisory CVSS: ", style=field_style)
text.append(f"{float(advisory_cvss):.1f}", style="dim")
contextual_reasoning = dependency_metadata.get("contextual_cvss_reasoning")
if contextual_reasoning:
text.append("\n\n")
text.append("Contextual CVSS Reasoning", style=field_style)
text.append("\n")
text.append(contextual_reasoning)
description = report.get("description")
if description:
text.append("\n\n")
+54
View File
@@ -0,0 +1,54 @@
"""LiteLLM model-name resolution for local cost estimates."""
from __future__ import annotations
from functools import lru_cache
from typing import Any, cast
@lru_cache(maxsize=512)
def resolve_litellm_model(model: str) -> str | None:
"""Return a provider-qualified model name that LiteLLM can price."""
try:
import litellm
normalized = model.strip()
for prefix in ("litellm/", "any-llm/", "openai/"):
if normalized.startswith(prefix):
normalized = normalized.removeprefix(prefix)
break
if not normalized:
return None
model_cost = cast(
"dict[str, dict[str, Any]]",
getattr(litellm, "model_cost"), # noqa: B009
)
bare_entry = model_cost.get(normalized)
if "/" not in normalized and isinstance(bare_entry, dict):
provider = bare_entry.get("litellm_provider")
if isinstance(provider, str) and provider:
return f"{provider}/{normalized}"
if "/" in normalized and isinstance(bare_entry, dict):
return normalized
names = [normalized]
if "/" in normalized:
names.append(normalized.rsplit("/", 1)[-1])
for name in names:
matches = sorted(key for key in model_cost if key.endswith(f"/{name}"))
if not matches:
continue
prices = {
(
model_cost[key].get("input_cost_per_token"),
model_cost[key].get("output_cost_per_token"),
)
for key in matches
if isinstance(model_cost.get(key), dict)
}
if len(matches) == 1 or len(prices) == 1:
return matches[0]
return None # noqa: TRY300
except Exception: # noqa: BLE001
return None
+6 -2
View File
@@ -14,6 +14,7 @@ from agents.usage import Usage
from strix.config import codex
from strix.config.loader import load_settings
from strix.core.paths import run_dir_for
from strix.report.pricing import resolve_litellm_model
from strix.report.sarif import write_sarif
from strix.report.usage import LLMUsageLedger
from strix.report.writer import (
@@ -696,10 +697,13 @@ def _estimate_response_cost(kwargs: Any, completion_response: Any) -> float | No
candidates.append(model.rsplit("/", 1)[-1])
for candidate in candidates:
resolved = resolve_litellm_model(candidate)
if not resolved:
continue
try:
value = completion_cost(
completion_response={"model": candidate, "usage": usage_payload},
model=candidate,
completion_response={"model": resolved, "usage": usage_payload},
model=resolved,
)
except Exception: # nosec B112 # noqa: BLE001, S112
continue
+30 -29
View File
@@ -7,6 +7,8 @@ from typing import Any
from agents.usage import Usage, deserialize_usage, serialize_usage
from strix.report.pricing import resolve_litellm_model
logger = logging.getLogger(__name__)
@@ -18,7 +20,9 @@ class LLMUsageLedger:
self._total_usage = Usage()
self._agent_usage: dict[str, Usage] = {}
self._agent_metadata: dict[str, dict[str, str]] = {}
self._total_cost = 0.0
self._observed_cost = 0.0
self._estimated_cost = 0.0
self._has_observed_cost = False
# When True, tokens are still tracked but cost stays $0 — the run is on a
# model subscription, so there is no metered per-token charge to report.
self.zero_cost = False
@@ -44,10 +48,10 @@ class LLMUsageLedger:
if model:
metadata["model"] = model
if not self.zero_cost and not _is_litellm_routed(model):
if not self.zero_cost:
estimated = _estimate_litellm_cost(usage, model)
if estimated:
self._total_cost += estimated
self._estimated_cost += estimated
return True
@@ -55,15 +59,18 @@ class LLMUsageLedger:
if self.zero_cost:
return
if isinstance(cost, int | float) and cost > 0:
self._total_cost += float(cost)
self._observed_cost += float(cost)
self._has_observed_cost = True
@property
def total_cost(self) -> float:
return _round_cost(self._total_cost)
if self.zero_cost:
return 0.0
return _round_cost(self._observed_cost if self._has_observed_cost else self._estimated_cost)
def to_record(self) -> dict[str, Any]:
record = serialize_usage(self._total_usage)
record["cost"] = _round_cost(self._total_cost)
record["cost"] = self.total_cost
record["agents"] = []
agent_tokens = {aid: _resolve_total_tokens(u) for aid, u in self._agent_usage.items()}
@@ -72,7 +79,7 @@ class LLMUsageLedger:
usage = self._agent_usage[agent_id]
metadata = self._agent_metadata.get(agent_id, {})
agent_cost = (
self._total_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
self.total_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
)
agent_record = serialize_usage(usage)
@@ -92,7 +99,9 @@ class LLMUsageLedger:
self._total_usage = Usage()
self._agent_usage.clear()
self._agent_metadata.clear()
self._total_cost = 0.0
self._observed_cost = 0.0
self._estimated_cost = 0.0
self._has_observed_cost = False
if not isinstance(raw_usage, dict):
return
@@ -103,7 +112,9 @@ class LLMUsageLedger:
logger.exception("Failed to hydrate aggregate llm_usage from run.json")
self._total_usage = Usage()
self._total_cost = _float_or_zero(raw_usage.get("cost"))
persisted_cost = _float_or_zero(raw_usage.get("cost"))
self._observed_cost = persisted_cost
self._estimated_cost = persisted_cost
for raw_agent in raw_usage.get("agents") or []:
if not isinstance(raw_agent, dict):
@@ -136,15 +147,6 @@ def _resolve_total_tokens(usage: Usage) -> int:
return prompt + completion
def _is_litellm_routed(model: str | None) -> bool:
if not model:
return False
name = model.strip().lower()
if "/" not in name:
return False
return not name.startswith("openai/")
def _usage_has_activity(usage: Usage) -> bool:
return bool(
usage.requests
@@ -201,24 +203,23 @@ def _estimate_litellm_entry_cost(entry: Any, model: str) -> float | None:
candidates = [model]
if "/" in model:
candidates.append(model.split("/", 1)[-1])
candidates.append(model.rsplit("/", 1)[-1])
cost: Any = None
for candidate in candidates:
resolved = resolve_litellm_model(candidate)
if not resolved:
continue
try:
cost = completion_cost(
completion_response={"model": candidate, "usage": usage_payload},
model=model,
completion_response={"model": resolved, "usage": usage_payload},
model=resolved,
)
break
except Exception: # nosec B112 # noqa: BLE001, S112
continue
if cost is None:
logger.debug("LiteLLM cost estimate unavailable for model %s", model)
return None
return cost if isinstance(cost, int | float) and cost >= 0 else None
if cost > 0:
return float(cost)
logger.debug("LiteLLM cost estimate unavailable for model %s", model)
return None
def _litellm_model_name(model: str | None) -> str | None:
+10
View File
@@ -215,6 +215,11 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
cvss = report.get("cvss")
if cvss is not None:
metadata.append(("CVSS", cvss))
advisory_cvss = dep_meta.get("advisory_cvss")
if advisory_cvss is not None and advisory_cvss != cvss:
metadata.append(("Advisory CVSS", advisory_cvss))
if dep_meta.get("contextual_cvss_vector"):
metadata.append(("Contextual CVSS Vector", dep_meta["contextual_cvss_vector"]))
if report.get("fix_effort"):
metadata.append(("Fix Effort", str(report["fix_effort"]).title()))
for label, value in metadata:
@@ -241,6 +246,11 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
lines.append(str(report["technical_analysis"]))
lines.append("")
if dep_meta.get("contextual_cvss_reasoning"):
lines.append("## Contextual CVSS\n")
lines.append(str(dep_meta["contextual_cvss_reasoning"]))
lines.append("")
if report.get("poc_description") or report.get("poc_script_code"):
lines.append("## Proof of Concept\n")
if report.get("poc_description"):
+78 -7
View File
@@ -161,7 +161,23 @@ fi
verdict/evidence onto its siblings; run the symbol search against each
CVE's own affected-symbol list. The import check (step 1) is the only
part shared across a package's CVEs.
3. If the analysis was not performed or is inconclusive (obfuscated code,
3. **Source-to-sink trace — do this whenever step 2 found a symbol hit.** A
symbol hit alone says the code calls the vulnerable API; it does not say
who can reach it. Start at the sink (the exact line that calls the
vulnerable function) and walk backwards hop by hop to the source: the
entry point that carries untrusted input (HTTP route, CLI argument, queue
or webhook payload, uploaded file, config value). Read each intermediate
function; when a hop is a thin wrapper, go one step deeper — never stop at
the first caller. Record what each hop enforces: authentication, a role
check, validation, a feature flag, a size or type limit, a default that is
off in production.
Write the chain into `reachability_evidence` as
`entry point -> intermediate call -> package call` with a
repository-relative `file:line` for every hop, and say who controls the
input. If no source reaches the sink, say that too — the level stays
`vulnerable_symbol_used` (the call is real), and the trace is what tells
the reader it is only reachable from, say, an operator CLI.
4. If the analysis was not performed or is inconclusive (obfuscated code,
dynamic loading, unparsable sources) ⇒ `unknown` and say why in
`assumptions`.
@@ -225,15 +241,68 @@ findings and rejects empty PoC fields):
installed/affected version, fixed version, lockfile path, and the relevant
trivy output excerpt.
- **Always set `advisory_cvss` to the published advisory base score (0.010.0).**
Severity is derived *solely* from this number: read it off the advisory (`CVSS`
in trivy output, or the NVD/GHSA page) and pass the real value. The tool rejects
a call that omits it, because guessing a score both inflates low CVEs and
deflates critical ones.
It is the published reference, and it rates the finding whenever you give no
contextual breakdown: read it off the advisory (`CVSS` in trivy output, or the
NVD/GHSA page) and pass the real value. The tool rejects a call that omits it,
because guessing a score both inflates low CVEs and deflates critical ones.
- Set `cwe` to the most specific `CWE-NNN` when the advisory names one.
- Do NOT cap severity at LOW just because there is no dynamic reproduction — use
the advisory score.
- Set `reachability` + `reachability_evidence` from the usage analysis above;
use `assumptions` for anything softer (confidence, caveats, analysis limits).
- Set `contextual_cvss_breakdown` + `contextual_cvss_reasoning` when this
codebase clearly changes the risk the published score describes (see below).
### Contextual CVSS
The published score rates the CVE in the abstract. `contextual_cvss_breakdown`
rates it **here**, in this codebase — the same 8-metric CVSS v3.1 object as a
normal finding's `cvss_breakdown` (`attack_vector`, `attack_complexity`,
`privileges_required`, `user_interaction`, `scope`, `confidentiality`,
`integrity`, `availability`). You never pass a score: the contextual score and
vector are computed from the breakdown, and when you provide one it determines
the finding's severity. `advisory_cvss` stays the published reference.
Start from the advisory's own published metrics and change only what your
evidence proves is different in this codebase:
- `attack_vector` `N`/`A`/`L`/`P` — as deployed. A library reached only by a
local CLI is `L`, not `N`.
- `attack_complexity` `L`/`H` — raise to `H` when the vulnerable path needs a
precondition the code enforces (input validation, a non-default flag, an
internal-only route).
- `privileges_required` `N`/`L`/`H`, `user_interaction` `N`/`R` — what this
deployment requires before the path is reachable.
- `scope` `U`/`C` — whether exploitation here escapes the component boundary.
- `confidentiality`/`integrity`/`availability` `N`/`L`/`H` — the impact in this
codebase. `not_imported` code the build still ships is usually `N` across all
three.
Ground every metric in the **source-to-sink trace** from the usage analysis
(step 3 above), not in a general impression of the package. Derive the metrics
from that chain: `attack_vector`, `privileges_required`, and `user_interaction`
come from what the source requires; `attack_complexity` comes from the
preconditions the hops enforce; `confidentiality`, `integrity`, and
`availability` come from the data and privileges available at the sink.
No trace, no contextual breakdown: if you did not reach a symbol hit, or you
could not follow a hop, omit the contextual fields instead of guessing.
`contextual_cvss_reasoning` is required with the breakdown. Write two to four
sentences that another engineer can check without opening the repository. Name
the chain hop by hop as `entry point -> intermediate call -> package call`, with
a repository-relative `file:line` for each hop, say who controls the input, and
say what the contextual rating changes. Example: lowering `attack_vector` to
`L` and `confidentiality` to `L` with "The only caller of `yaml.load` is
`parse_manifest` in `scripts/import.py:88`, which `cli/commands.py:212` invokes
for an operator-supplied path behind the `--allow-unsafe-import` flag that
`deploy/prod.yaml` never sets. No HTTP route reaches that function, so an
attacker must already hold shell access on the job host, and the parsed data is
build metadata rather than customer records."
Omit all the contextual fields when the published rating already fits, and when
the evidence is thin. A contextual rating is a claim you must be able to
defend, and it never replaces `advisory_cvss` as the published reference.
Verify the CVE with `web_search` when available before reporting. Never guess or
hallucinate a CVE id.
@@ -244,10 +313,12 @@ hallucinate a CVE id.
`create_dependency_report`.
- Do not report a finding without a verified CVE id.
- Do not batch multiple CVEs into one report.
- Do not omit `advisory_cvss` — the tool rejects it, and it is the single input
that determines dependency severity.
- Do not omit `advisory_cvss` — the tool rejects it, and it rates every finding
that carries no contextual breakdown.
- Do not silently drop a known CVE because it lacks a dynamic PoC — that is the
exact failure this skill prevents.
- Do not downgrade advisory severity for lack of dynamic reproduction.
- Do not claim a `reachability` level the evidence does not prove — `unknown`
with a reason is always acceptable; an overclaimed level never is.
- Do not send `contextual_cvss_breakdown` without evidence-backed reasoning, and
do not use it to quietly de-rate a CVE you simply could not analyze.
+152 -15
View File
@@ -749,6 +749,61 @@ def _validate_manifest_path(manifest_path: str | None) -> str | None:
return None
_MAX_CONTEXTUAL_REASONING_CHARS = 2000
def _validate_contextual_cvss(
breakdown: dict[str, str] | None,
reasoning: str | None,
) -> list[str]:
errors: list[str] = []
if breakdown:
for name, valid in _CVSS_VALID.items():
value = breakdown.get(name)
if value not in valid:
errors.append(
f"Invalid contextual_cvss_breakdown {name}: {value}. Must be one of: {valid}"
)
if not (reasoning or "").strip():
errors.append(
"contextual_cvss_reasoning is required when contextual_cvss_breakdown is "
"set: state what you observed in this codebase that justifies the "
"contextual rating. A contextual score with no reasoning is not shown."
)
return errors
def _validate_advisory_cvss(advisory_cvss: float | None) -> str | None:
if advisory_cvss is None:
return (
"advisory_cvss is required: read the published advisory base score "
"(0.0-10.0) off the advisory (trivy CVSS / NVD / GHSA). It is the "
"published reference the finding is rated against — do not omit it "
"or the finding cannot be rated."
)
if not 0.0 <= advisory_cvss <= 10.0:
return f"advisory_cvss must be between 0.0 and 10.0, got {advisory_cvss}"
return None
def _resolve_dependency_rating(
advisory_cvss: float | None,
contextual_cvss_breakdown: dict[str, str] | None,
) -> tuple[float | None, str, float | None, str | None]:
"""Rate the finding.
A contextual breakdown works exactly like a normal finding's
``cvss_breakdown``: the agent supplies the 8 metrics as observed in this
codebase and the score/vector are computed from them. When provided it
rates the finding; the advisory score stays as the published reference.
"""
if contextual_cvss_breakdown:
score, severity, vector = _calculate_cvss(contextual_cvss_breakdown)
return score, severity, score, vector
score, severity = _dependency_severity(advisory_cvss)
return score, severity, None, None
def _build_dependency_metadata(
*,
package_name: str,
@@ -760,11 +815,18 @@ def _build_dependency_metadata(
manifest_path: str | None = None,
reachability: str | None = None,
reachability_evidence: str | None = None,
) -> dict[str, str]:
metadata = {
advisory_cvss: float | None = None,
contextual_cvss_breakdown: dict[str, str] | None = None,
contextual_cvss_score: float | None = None,
contextual_cvss_vector: str | None = None,
contextual_cvss_reasoning: str | None = None,
) -> dict[str, Any]:
metadata: dict[str, Any] = {
"package_name": package_name.strip(),
"installed_version": installed_version.strip(),
}
if advisory_cvss is not None:
metadata["advisory_cvss"] = advisory_cvss
if package_ecosystem and package_ecosystem.strip():
metadata["package_ecosystem"] = package_ecosystem.strip()
if manifest_path and manifest_path.strip():
@@ -781,6 +843,20 @@ def _build_dependency_metadata(
metadata["reachability"] = reachability.strip()
if reachability_evidence and reachability_evidence.strip():
metadata["reachability_evidence"] = reachability_evidence.strip()
# Contextual CVSS is only meaningful as the full breakdown, its computed
# score/vector, and the reasoning a reader can check — an incomplete set
# is dropped.
reasoning = str(contextual_cvss_reasoning or "").strip()
if (
contextual_cvss_breakdown
and contextual_cvss_score is not None
and contextual_cvss_vector
and reasoning
):
metadata["contextual_cvss_breakdown"] = contextual_cvss_breakdown
metadata["contextual_cvss_score"] = contextual_cvss_score
metadata["contextual_cvss_vector"] = contextual_cvss_vector
metadata["contextual_cvss_reasoning"] = reasoning[:_MAX_CONTEXTUAL_REASONING_CHARS]
return metadata
@@ -852,6 +928,8 @@ async def _do_create_dependency( # noqa: PLR0912
manifest_path: str | None = None,
reachability: str = "unknown",
reachability_evidence: str | None = None,
contextual_cvss_breakdown: dict[str, str] | None = None,
contextual_cvss_reasoning: str | None = None,
agent_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
@@ -904,19 +982,21 @@ async def _do_create_dependency( # noqa: PLR0912
"govulncheck call path). Never claim a reachability level without evidence."
)
if advisory_cvss is None:
errors.append(
"advisory_cvss is required: read the published advisory base score "
"(0.0-10.0) off the advisory (trivy CVSS / NVD / GHSA). Severity is "
"derived solely from it — do not omit it or the finding cannot be rated."
)
elif not 0.0 <= advisory_cvss <= 10.0:
errors.append(f"advisory_cvss must be between 0.0 and 10.0, got {advisory_cvss}")
errors.extend(_validate_contextual_cvss(contextual_cvss_breakdown, contextual_cvss_reasoning))
advisory_err = _validate_advisory_cvss(advisory_cvss)
if advisory_err:
errors.append(advisory_err)
if errors:
return {"success": False, "error": "Validation failed", "errors": errors}
cvss_score, severity = _dependency_severity(advisory_cvss)
try:
cvss_score, severity, contextual_score, contextual_vector = _resolve_dependency_rating(
advisory_cvss, contextual_cvss_breakdown
)
except ValueError as exc:
return {"success": False, "error": "Validation failed", "errors": [str(exc)]}
dependency_metadata = _build_dependency_metadata(
package_name=package_name,
installed_version=installed_version,
@@ -927,6 +1007,11 @@ async def _do_create_dependency( # noqa: PLR0912
manifest_path=manifest_path,
reachability=reachability,
reachability_evidence=reachability_evidence,
advisory_cvss=advisory_cvss,
contextual_cvss_breakdown=contextual_cvss_breakdown,
contextual_cvss_score=contextual_score,
contextual_cvss_vector=contextual_vector,
contextual_cvss_reasoning=contextual_cvss_reasoning,
)
evidence = _build_dependency_evidence(
cve=parsed_cve,
@@ -1038,6 +1123,8 @@ async def create_dependency_report(
dependency_path: str | None = None,
reachability: str = "unknown",
reachability_evidence: str | None = None,
contextual_cvss_breakdown: dict[str, str] | None = None,
contextual_cvss_reasoning: str | None = None,
) -> str:
"""File a known-CVE dependency (SCA) finding — one report per CVE x package.
@@ -1080,8 +1167,10 @@ async def create_dependency_report(
proved a path from application code to the vulnerable function.
- ``unknown`` — usage analysis was not performed or was inconclusive.
Severity is still derived solely from ``advisory_cvss`` — the
reachability level never changes the rating, only prioritization.
Severity comes from ``contextual_cvss_breakdown`` when you provide one
(computed exactly like a normal finding's ``cvss_breakdown``), otherwise
from ``advisory_cvss``. The reachability level alone never changes the
rating, only prioritization.
**Formatting**: use markdown in text fields (``**bold**``, ``inline
code`` for package/version identifiers, fenced code blocks for
@@ -1102,8 +1191,9 @@ async def create_dependency_report(
cwe: ``CWE-NNN`` (most specific) if certain, else omit.
advisory_cvss: **Required.** Published advisory base score
(0.0-10.0) — read it off the advisory (trivy CVSS / NVD / GHSA).
Severity is derived solely from this score, so it must be the
real published value; do not guess or omit it.
It is the published reference the finding is rated against and
rates the finding whenever you give no contextual breakdown, so
it must be the real published value; do not guess or omit it.
technical_analysis: Optional deeper mechanism/root-cause detail.
fix_effort: One of ``trivial`` / ``low`` / ``medium`` / ``high``
(dependency upgrades are usually ``trivial``/``low``).
@@ -1131,6 +1221,51 @@ async def create_dependency_report(
(required for any level other than ``unknown``): repo-relative
``file:line`` of the import or symbol usage, the matched
advisory symbols, or the govulncheck call-path excerpt.
Whenever you found the vulnerable symbol in use, also give the
**source-to-sink trace** here: start at the vulnerable package
call site and walk backwards hop by hop to the entry point
that carries untrusted input (HTTP route, CLI argument, queue
message, webhook, config file), going one step deeper whenever
a hop is a wrapper. Write it as ``entry point -> intermediate
call -> package call`` with a ``file:line`` per hop, name what
each hop enforces (auth, role check, validation, a flag that
is off in production), and say who controls the input. State
it plainly when no entry point reaches the sink — that is the
most useful result a reader can get.
contextual_cvss_breakdown: Optional full CVSS v3.1 rating of this
CVE **in this codebase** — the same 8-metric object as
``create_vulnerability_report``'s ``cvss_breakdown``:
``attack_vector`` (N/A/L/P), ``attack_complexity`` (L/H),
``privileges_required`` (N/L/H), ``user_interaction`` (N/R),
``scope`` (U/C), ``confidentiality`` / ``integrity`` /
``availability`` (N/L/H). All 8 metrics are required when the
field is set, and the contextual score/vector are computed
from them — you never supply a score. Start from the
advisory's published metrics and change only what the
**source-to-sink trace** you recorded in
``reachability_evidence`` proves is different here: derive
``attack_vector`` / ``privileges_required`` /
``user_interaction`` from what the entry point actually
requires, ``attack_complexity`` from the preconditions the
hops enforce, and the impact metrics from the data and
privileges reachable at the sink. When provided, this rating
determines the finding's severity; ``advisory_cvss`` stays as
the published reference. Omit the field when the trace does
not change the published rating, or when you could not
complete the trace.
contextual_cvss_reasoning: **Required whenever**
``contextual_cvss_breakdown`` is set. Two to four detailed
sentences that a reviewer can verify without opening the repo:
how the application uses the package, which call sites or
configuration you inspected (repo-relative ``file:line``),
which input reaches the vulnerable code and whether an
attacker controls it, and what the adjustment therefore
changes. State the source-to-sink chain explicitly, hop by
hop, as ``entry point -> intermediate call -> package call``
with a ``file:line`` for each hop. Cite concrete evidence,
never a generic statement such as "low risk". The user reads
this text next to the adjusted score, so an adjustment
without it is discarded.
"""
agent_id, agent_name = _caller_identity(ctx)
@@ -1155,6 +1290,8 @@ async def create_dependency_report(
manifest_path=manifest_path,
reachability=reachability,
reachability_evidence=reachability_evidence,
contextual_cvss_breakdown=contextual_cvss_breakdown,
contextual_cvss_reasoning=contextual_cvss_reasoning,
agent_id=agent_id,
agent_name=agent_name,
)
+1 -1
View File
@@ -143,7 +143,7 @@ def test_cost_callback_estimates_cost_with_bare_model_fallback() -> None:
}
def fake_completion_cost(**kwargs: object) -> float:
if kwargs["model"] == "gpt-4o-mini":
if kwargs["model"] == "openai/gpt-4o-mini":
return 0.025
raise ValueError(kwargs["model"])
+120
View File
@@ -0,0 +1,120 @@
from __future__ import annotations
from unittest.mock import patch
import litellm
from agents.usage import Usage
from strix.report.pricing import resolve_litellm_model
from strix.report.usage import LLMUsageLedger
def test_resolves_common_bare_model_names() -> None:
resolve_litellm_model.cache_clear()
assert resolve_litellm_model("deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
assert resolve_litellm_model("openai/deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
assert resolve_litellm_model("grok-4.5") == "xai/grok-4.5"
assert resolve_litellm_model("MiniMax-M3") == "minimax/MiniMax-M3"
def test_resolver_returns_none_for_unresolvable_model() -> None:
resolve_litellm_model.cache_clear()
assert resolve_litellm_model("provider/not-a-real-model") is None
def test_ledger_uses_estimate_when_routed_provider_reports_no_cost() -> None:
usage = Usage()
usage.requests = 1
usage.input_tokens = 1000
usage.output_tokens = 200
usage.total_tokens = 1200
ledger = LLMUsageLedger()
with patch("litellm.completion_cost", return_value=0.42):
ledger.record(agent_id="a", usage=usage, model="openai/deepseek-v4-flash")
assert ledger.total_cost == 0.42
def test_ledger_prefers_observed_cost_over_estimate() -> None:
usage = Usage()
usage.requests = 1
usage.input_tokens = 1000
usage.output_tokens = 200
usage.total_tokens = 1200
ledger = LLMUsageLedger()
with patch("litellm.completion_cost", return_value=0.42):
ledger.record(agent_id="a", usage=usage, model="openai/deepseek-v4-flash")
ledger.record_observed_cost(0.17)
assert ledger.total_cost == 0.17
def test_hydrated_estimate_continues_accumulating_new_estimates() -> None:
usage = Usage()
usage.requests = 1
usage.input_tokens = 1000
usage.output_tokens = 200
usage.total_tokens = 1200
ledger = LLMUsageLedger()
ledger.hydrate({"cost": 0.42})
with patch("litellm.completion_cost", return_value=0.17):
ledger.record(agent_id="a", usage=usage, model="openai/deepseek-v4-flash")
assert ledger.total_cost == 0.59
def test_zero_cost_disables_both_observed_and_estimated_costs() -> None:
usage = Usage()
usage.requests = 1
usage.input_tokens = 1000
usage.output_tokens = 200
usage.total_tokens = 1200
ledger = LLMUsageLedger()
ledger.zero_cost = True
with patch("litellm.completion_cost", return_value=0.42) as estimate:
ledger.record(agent_id="a", usage=usage, model="deepseek-v4-flash")
ledger.record_observed_cost(1.0)
estimate.assert_not_called()
assert ledger.total_cost == 0.0
def test_resolver_uses_provider_when_bare_entry_has_one() -> None:
original = litellm.model_cost
litellm.model_cost = {
"example": {
"litellm_provider": "example-provider",
"input_cost_per_token": 1.0,
"output_cost_per_token": 2.0,
}
}
try:
resolve_litellm_model.cache_clear()
assert resolve_litellm_model("example") == "example-provider/example"
finally:
litellm.model_cost = original
resolve_litellm_model.cache_clear()
def test_resolver_does_not_guess_between_differently_priced_providers() -> None:
original = litellm.model_cost
litellm.model_cost = {
"provider-a/example": {
"input_cost_per_token": 1.0,
"output_cost_per_token": 2.0,
},
"provider-b/example": {
"input_cost_per_token": 3.0,
"output_cost_per_token": 4.0,
},
}
try:
resolve_litellm_model.cache_clear()
assert resolve_litellm_model("example") is None
finally:
litellm.model_cost = original
resolve_litellm_model.cache_clear()
+156 -1
View File
@@ -160,6 +160,7 @@ async def test_dependency_report_sets_class_and_metadata(report_state: ReportSta
assert report["dependency_metadata"] == {
"package_name": "lodash",
"installed_version": "4.17.20",
"advisory_cvss": 7.2,
"package_ecosystem": "npm",
"manifest_path": "package-lock.json",
"fixed_version": "4.17.21",
@@ -372,7 +373,7 @@ async def test_dependency_report_omits_unknown_reachability(report_state: Report
fix_effort="low",
)
assert result["success"] is True
assert result["success"] is True, result
metadata = report_state.vulnerability_reports[0]["dependency_metadata"]
assert "reachability" not in metadata
assert "reachability_evidence" not in metadata
@@ -463,6 +464,7 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"advisory_cvss": 0.0,
"package_ecosystem": "npm",
"manifest_path": "package-lock.json",
"fixed_version": "1.0.1",
@@ -877,3 +879,156 @@ def test_vuln_tool_exposes_new_params() -> None:
dep_required = create_dependency_report.params_json_schema["required"]
assert "package_ecosystem" in dep_required
assert "advisory_cvss" in dep_required
def test_dep_tool_exposes_contextual_cvss_params() -> None:
dep_props = create_dependency_report.params_json_schema["properties"]
for field in (
"contextual_cvss_breakdown",
"contextual_cvss_reasoning",
):
assert field in dep_props
assert "source-to-sink" in dep_props["contextual_cvss_breakdown"]["description"].lower()
assert "source-to-sink" in dep_props["reachability_evidence"]["description"].lower()
assert "file:line" in dep_props["contextual_cvss_reasoning"]["description"].lower()
_CONTEXTUAL_BREAKDOWN = {
"attack_vector": "L",
"attack_complexity": "H",
"privileges_required": "H",
"user_interaction": "N",
"scope": "U",
"confidentiality": "L",
"integrity": "L",
"availability": "N",
}
@pytest.mark.asyncio
async def test_dependency_report_computes_contextual_cvss(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
advisory_cvss=7.2,
technical_analysis=None,
fixed_version="4.17.21",
cwe="CWE-94",
fix_effort="trivial",
manifest_path="package-lock.json",
contextual_cvss_breakdown=_CONTEXTUAL_BREAKDOWN,
contextual_cvss_reasoning="Only scripts/import.py reaches the sink.",
)
assert result["success"] is True, result
report = report_state.vulnerability_reports[0]
metadata = report["dependency_metadata"]
assert metadata["advisory_cvss"] == 7.2
assert metadata["contextual_cvss_breakdown"] == _CONTEXTUAL_BREAKDOWN
assert metadata["contextual_cvss_vector"] == ("CVSS:3.1/AV:L/AC:H/PR:H/UI:N/S:U/C:L/I:L/A:N")
assert metadata["contextual_cvss_score"] == pytest.approx(3.0, abs=0.05)
assert metadata["contextual_cvss_reasoning"] == "Only scripts/import.py reaches the sink."
# The contextual rating determines the finding's score/severity, exactly
# like a normal finding's cvss_breakdown.
assert report["cvss"] == metadata["contextual_cvss_score"]
assert report["severity"] == "low"
@pytest.mark.asyncio
async def test_dependency_report_rates_from_advisory_without_contextual(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
advisory_cvss=7.2,
technical_analysis=None,
fixed_version="4.17.21",
cwe="CWE-94",
fix_effort="trivial",
manifest_path="package-lock.json",
)
assert result["success"] is True, result
report = report_state.vulnerability_reports[0]
assert report["cvss"] == 7.2
assert report["severity"] == "high"
metadata = report["dependency_metadata"]
assert metadata["advisory_cvss"] == 7.2
assert "contextual_cvss_breakdown" not in metadata
assert "contextual_cvss_score" not in metadata
@pytest.mark.asyncio
async def test_dependency_report_rejects_incomplete_contextual_breakdown(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
advisory_cvss=7.2,
technical_analysis=None,
fixed_version="4.17.21",
cwe="CWE-94",
fix_effort="trivial",
manifest_path="package-lock.json",
contextual_cvss_breakdown={"attack_vector": "L", "attack_complexity": "Z"},
contextual_cvss_reasoning="Only scripts/import.py reaches the sink.",
)
assert result["success"] is False
assert any("attack_complexity" in error for error in result["errors"])
assert any("privileges_required" in error for error in result["errors"])
assert report_state.vulnerability_reports == []
@pytest.mark.asyncio
async def test_dependency_report_rejects_contextual_breakdown_without_reasoning(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
advisory_cvss=7.2,
technical_analysis=None,
fixed_version="4.17.21",
cwe="CWE-94",
fix_effort="trivial",
manifest_path="package-lock.json",
contextual_cvss_breakdown=_CONTEXTUAL_BREAKDOWN,
contextual_cvss_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 == []