mirror of
https://github.com/usestrix/strix.git
synced 2026-08-18 01:39:19 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2bcd0d267c | ||
|
|
9c5307d278 | ||
|
|
7b855c5cb0 | ||
|
|
56c68011c6 | ||
|
|
1300d5a615 | ||
|
|
b1b42393ab |
@@ -133,27 +133,6 @@ 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")
|
||||
|
||||
@@ -1,54 +0,0 @@
|
||||
"""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
|
||||
@@ -14,7 +14,6 @@ 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 (
|
||||
@@ -697,13 +696,10 @@ 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": resolved, "usage": usage_payload},
|
||||
model=resolved,
|
||||
completion_response={"model": candidate, "usage": usage_payload},
|
||||
model=candidate,
|
||||
)
|
||||
except Exception: # nosec B112 # noqa: BLE001, S112
|
||||
continue
|
||||
|
||||
+29
-30
@@ -7,8 +7,6 @@ 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__)
|
||||
|
||||
@@ -20,9 +18,7 @@ class LLMUsageLedger:
|
||||
self._total_usage = Usage()
|
||||
self._agent_usage: dict[str, Usage] = {}
|
||||
self._agent_metadata: dict[str, dict[str, str]] = {}
|
||||
self._observed_cost = 0.0
|
||||
self._estimated_cost = 0.0
|
||||
self._has_observed_cost = False
|
||||
self._total_cost = 0.0
|
||||
# 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
|
||||
@@ -48,10 +44,10 @@ class LLMUsageLedger:
|
||||
if model:
|
||||
metadata["model"] = model
|
||||
|
||||
if not self.zero_cost:
|
||||
if not self.zero_cost and not _is_litellm_routed(model):
|
||||
estimated = _estimate_litellm_cost(usage, model)
|
||||
if estimated:
|
||||
self._estimated_cost += estimated
|
||||
self._total_cost += estimated
|
||||
|
||||
return True
|
||||
|
||||
@@ -59,18 +55,15 @@ class LLMUsageLedger:
|
||||
if self.zero_cost:
|
||||
return
|
||||
if isinstance(cost, int | float) and cost > 0:
|
||||
self._observed_cost += float(cost)
|
||||
self._has_observed_cost = True
|
||||
self._total_cost += float(cost)
|
||||
|
||||
@property
|
||||
def total_cost(self) -> float:
|
||||
if self.zero_cost:
|
||||
return 0.0
|
||||
return _round_cost(self._observed_cost if self._has_observed_cost else self._estimated_cost)
|
||||
return _round_cost(self._total_cost)
|
||||
|
||||
def to_record(self) -> dict[str, Any]:
|
||||
record = serialize_usage(self._total_usage)
|
||||
record["cost"] = self.total_cost
|
||||
record["cost"] = _round_cost(self._total_cost)
|
||||
record["agents"] = []
|
||||
|
||||
agent_tokens = {aid: _resolve_total_tokens(u) for aid, u in self._agent_usage.items()}
|
||||
@@ -79,7 +72,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)
|
||||
@@ -99,9 +92,7 @@ class LLMUsageLedger:
|
||||
self._total_usage = Usage()
|
||||
self._agent_usage.clear()
|
||||
self._agent_metadata.clear()
|
||||
self._observed_cost = 0.0
|
||||
self._estimated_cost = 0.0
|
||||
self._has_observed_cost = False
|
||||
self._total_cost = 0.0
|
||||
|
||||
if not isinstance(raw_usage, dict):
|
||||
return
|
||||
@@ -112,9 +103,7 @@ class LLMUsageLedger:
|
||||
logger.exception("Failed to hydrate aggregate llm_usage from run.json")
|
||||
self._total_usage = Usage()
|
||||
|
||||
persisted_cost = _float_or_zero(raw_usage.get("cost"))
|
||||
self._observed_cost = persisted_cost
|
||||
self._estimated_cost = persisted_cost
|
||||
self._total_cost = _float_or_zero(raw_usage.get("cost"))
|
||||
|
||||
for raw_agent in raw_usage.get("agents") or []:
|
||||
if not isinstance(raw_agent, dict):
|
||||
@@ -147,6 +136,15 @@ 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
|
||||
@@ -203,23 +201,24 @@ def _estimate_litellm_entry_cost(entry: Any, model: str) -> float | None:
|
||||
|
||||
candidates = [model]
|
||||
if "/" in model:
|
||||
candidates.append(model.rsplit("/", 1)[-1])
|
||||
candidates.append(model.split("/", 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": resolved, "usage": usage_payload},
|
||||
model=resolved,
|
||||
completion_response={"model": candidate, "usage": usage_payload},
|
||||
model=model,
|
||||
)
|
||||
break
|
||||
except Exception: # nosec B112 # noqa: BLE001, S112
|
||||
continue
|
||||
if cost > 0:
|
||||
return float(cost)
|
||||
logger.debug("LiteLLM cost estimate unavailable for model %s", model)
|
||||
return None
|
||||
|
||||
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
|
||||
|
||||
|
||||
def _litellm_model_name(model: str | None) -> str | None:
|
||||
|
||||
@@ -215,11 +215,6 @@ 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:
|
||||
@@ -246,11 +241,6 @@ 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"):
|
||||
|
||||
@@ -161,23 +161,7 @@ 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. **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,
|
||||
3. If the analysis was not performed or is inconclusive (obfuscated code,
|
||||
dynamic loading, unparsable sources) ⇒ `unknown` and say why in
|
||||
`assumptions`.
|
||||
|
||||
@@ -241,68 +225,15 @@ 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.0–10.0).**
|
||||
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.
|
||||
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.
|
||||
- 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.
|
||||
@@ -313,12 +244,10 @@ 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 rates every finding
|
||||
that carries no contextual breakdown.
|
||||
- Do not omit `advisory_cvss` — the tool rejects it, and it is the single input
|
||||
that determines dependency severity.
|
||||
- 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.
|
||||
|
||||
+15
-152
@@ -749,61 +749,6 @@ 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,
|
||||
@@ -815,18 +760,11 @@ def _build_dependency_metadata(
|
||||
manifest_path: str | None = None,
|
||||
reachability: str | None = None,
|
||||
reachability_evidence: str | None = None,
|
||||
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] = {
|
||||
) -> dict[str, str]:
|
||||
metadata = {
|
||||
"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():
|
||||
@@ -843,20 +781,6 @@ 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
|
||||
|
||||
|
||||
@@ -928,8 +852,6 @@ 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]:
|
||||
@@ -982,21 +904,19 @@ async def _do_create_dependency( # noqa: PLR0912
|
||||
"govulncheck call path). Never claim a reachability level without evidence."
|
||||
)
|
||||
|
||||
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 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}")
|
||||
|
||||
if errors:
|
||||
return {"success": False, "error": "Validation failed", "errors": errors}
|
||||
|
||||
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)]}
|
||||
cvss_score, severity = _dependency_severity(advisory_cvss)
|
||||
dependency_metadata = _build_dependency_metadata(
|
||||
package_name=package_name,
|
||||
installed_version=installed_version,
|
||||
@@ -1007,11 +927,6 @@ 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,
|
||||
@@ -1123,8 +1038,6 @@ 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.
|
||||
|
||||
@@ -1167,10 +1080,8 @@ 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 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.
|
||||
Severity is still derived solely from ``advisory_cvss`` — the
|
||||
reachability level never changes the rating, only prioritization.
|
||||
|
||||
**Formatting**: use markdown in text fields (``**bold**``, ``inline
|
||||
code`` for package/version identifiers, fenced code blocks for
|
||||
@@ -1191,9 +1102,8 @@ 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).
|
||||
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.
|
||||
Severity is derived solely from this score, 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``).
|
||||
@@ -1221,51 +1131,6 @@ 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)
|
||||
|
||||
@@ -1290,8 +1155,6 @@ 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,
|
||||
)
|
||||
|
||||
@@ -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"] == "openai/gpt-4o-mini":
|
||||
if kwargs["model"] == "gpt-4o-mini":
|
||||
return 0.025
|
||||
raise ValueError(kwargs["model"])
|
||||
|
||||
|
||||
@@ -1,120 +0,0 @@
|
||||
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()
|
||||
@@ -160,7 +160,6 @@ 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",
|
||||
@@ -373,7 +372,7 @@ async def test_dependency_report_omits_unknown_reachability(report_state: Report
|
||||
fix_effort="low",
|
||||
)
|
||||
|
||||
assert result["success"] is True, result
|
||||
assert result["success"] is True
|
||||
metadata = report_state.vulnerability_reports[0]["dependency_metadata"]
|
||||
assert "reachability" not in metadata
|
||||
assert "reachability_evidence" not in metadata
|
||||
@@ -464,7 +463,6 @@ 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",
|
||||
@@ -879,156 +877,3 @@ 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 == []
|
||||
|
||||
Reference in New Issue
Block a user