mirror of
https://github.com/usestrix/strix.git
synced 2026-11-09 05:19:39 -10:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Alex Schapiro
|
58e68a5db3 | ||
|
Alex Schapiro
|
2ea56755e6 |
No files matched your search
+4
-68
@@ -186,84 +186,20 @@ RUN set -eux; \
|
||||
aarch64|arm64) GITLEAKS_ARCH="arm64" ;; \
|
||||
*) echo "Unsupported architecture: $ARCH" >&2; exit 1 ;; \
|
||||
esac; \
|
||||
TAG="$(curl -fsSLI -o /dev/null -w '%{url_effective}' https://github.com/gitleaks/gitleaks/releases/latest | sed 's#.*/##')"; \
|
||||
TAG="$(curl -fsSL https://api.github.com/repos/gitleaks/gitleaks/releases/latest | jq -r .tag_name)"; \
|
||||
curl -fsSL "https://github.com/gitleaks/gitleaks/releases/download/${TAG}/gitleaks_${TAG#v}_linux_${GITLEAKS_ARCH}.tar.gz" -o /tmp/gitleaks.tgz; \
|
||||
tar -xzf /tmp/gitleaks.tgz -C /tmp; \
|
||||
install -m 0755 /tmp/gitleaks /usr/local/bin/gitleaks; \
|
||||
rm -f /tmp/gitleaks /tmp/gitleaks.tgz
|
||||
|
||||
ARG TRIVY_VERSION=0.76.1
|
||||
ARG TRIVY_SHA256_X86=c7cbe3137fb8a48a17574bede3dd6beb16742d1e9772929940fc90082c39d9c0
|
||||
ARG TRIVY_SHA256_AARCH64=e174dff2a6dba5e3ae41f8d0537a6324405bd08132497340304c6b20e0f16a1f
|
||||
RUN set -eux; \
|
||||
case "$(uname -m)" in \
|
||||
x86_64) TRIVY_ARCH="64bit"; TRIVY_SHA256="${TRIVY_SHA256_X86}" ;; \
|
||||
aarch64|arm64) TRIVY_ARCH="ARM64"; TRIVY_SHA256="${TRIVY_SHA256_AARCH64}" ;; \
|
||||
*) echo "Unsupported architecture: $(uname -m)" >&2; exit 1 ;; \
|
||||
esac; \
|
||||
curl -fsSL "https://github.com/aquasecurity/trivy/releases/download/v${TRIVY_VERSION}/trivy_${TRIVY_VERSION}_Linux-${TRIVY_ARCH}.tar.gz" -o /tmp/trivy.tgz; \
|
||||
echo "${TRIVY_SHA256} /tmp/trivy.tgz" | sha256sum -c -; \
|
||||
tar -xzf /tmp/trivy.tgz -C /tmp trivy; \
|
||||
install -m 0755 /tmp/trivy /usr/local/bin/trivy; \
|
||||
rm -f /tmp/trivy /tmp/trivy.tgz; \
|
||||
trivy --version
|
||||
RUN curl -sfL https://raw.githubusercontent.com/aquasecurity/trivy/main/contrib/install.sh | sh -s -- -b /usr/local/bin
|
||||
|
||||
RUN apt-get install -y wapiti
|
||||
|
||||
# CVE-2026-111497: out-of-bounds write in the brotli decoder, fixed in brotli
|
||||
# 1.2.1 / Debian 1.2.0-4.1. kali-rolling still ships 1.2.0-4, and node, curl and
|
||||
# python3 in this image all load the shared libbrotlidec, so every HTTP client here
|
||||
# would decode a malicious `Content-Encoding: br` response with the vulnerable
|
||||
# code. Install the fixed Debian packages until Kali picks them up.
|
||||
ARG BROTLI_DEB_VERSION=1.2.0-4.1
|
||||
RUN set -eux; \
|
||||
ARCH="$(dpkg --print-architecture)"; \
|
||||
for pkg in libbrotli1 python3-brotli; do \
|
||||
curl -fsSL "https://deb.debian.org/debian/pool/main/b/brotli/${pkg}_${BROTLI_DEB_VERSION}_${ARCH}.deb" -o "/tmp/${pkg}.deb"; \
|
||||
done; \
|
||||
dpkg -i /tmp/libbrotli1.deb /tmp/python3-brotli.deb; \
|
||||
rm -f /tmp/libbrotli1.deb /tmp/python3-brotli.deb; \
|
||||
test "$(dpkg-query -W -f='${Version}' libbrotli1)" = "${BROTLI_DEB_VERSION}"; \
|
||||
node -e 'const z=require("zlib");z.brotliDecompressSync(z.brotliCompressSync(Buffer.from("ok")))'
|
||||
|
||||
ARG OPENGREP_VERSION=1.31.0
|
||||
ARG OPENGREP_SHA256_X86=dbbe3c2058dbe600c449a8ba22ea79a1643a6660658400090d0323a8d0d31147
|
||||
ARG OPENGREP_SHA256_AARCH64=2b713a29a00237f85408b83002fc111db5b94c7fa8d263f0a93184a960655d9d
|
||||
RUN set -eux; \
|
||||
case "$(uname -m)" in \
|
||||
x86_64) OPENGREP_ARCH="x86"; OPENGREP_SHA256="${OPENGREP_SHA256_X86}" ;; \
|
||||
aarch64|arm64) OPENGREP_ARCH="aarch64"; OPENGREP_SHA256="${OPENGREP_SHA256_AARCH64}" ;; \
|
||||
*) echo "Unsupported architecture: $(uname -m)" >&2; exit 1 ;; \
|
||||
esac; \
|
||||
curl -fsSL "https://github.com/opengrep/opengrep/releases/download/v${OPENGREP_VERSION}/opengrep_manylinux_${OPENGREP_ARCH}" -o /tmp/opengrep; \
|
||||
echo "${OPENGREP_SHA256} /tmp/opengrep" | sha256sum -c -; \
|
||||
install -m 0755 /tmp/opengrep /usr/local/bin/opengrep; \
|
||||
rm -f /tmp/opengrep; \
|
||||
opengrep --version
|
||||
|
||||
# Opengrep ships no rule registry of its own; bake the community ruleset so
|
||||
# scans run offline. Keep only rule YAML and LICENSE; drop tests, fixtures,
|
||||
# tooling, and rules marked with a proprietary license.
|
||||
ARG OPENGREP_RULES_REF=f1d2b562b414783763fd02a6ed2736eaed622efa
|
||||
RUN set -eux; \
|
||||
mkdir -p /opt/opengrep-rules; \
|
||||
curl -fsSL "https://codeload.github.com/opengrep/opengrep-rules/tar.gz/${OPENGREP_RULES_REF}" \
|
||||
| tar -xz --strip-components=1 -C /opt/opengrep-rules; \
|
||||
cd /opt/opengrep-rules; \
|
||||
rm -rf .github scripts stats; \
|
||||
find . -maxdepth 1 -type f ! -name LICENSE -delete; \
|
||||
find . -mindepth 2 -type f \( ! \( -name '*.yaml' -o -name '*.yml' \) -o -name '*.test.yaml' -o -name '*.test.yml' -o -name '.*' \) -delete; \
|
||||
grep -rlZ 'license: proprietary' . | xargs -0 -r rm -f; \
|
||||
find . -type d -empty -delete; \
|
||||
chmod -R a+rX /opt/opengrep-rules
|
||||
|
||||
# Strix-authored rules that close gaps in the community ruleset.
|
||||
COPY containers/opengrep-rules/ /opt/opengrep-rules/strix/
|
||||
RUN chmod -R a+rX /opt/opengrep-rules/strix
|
||||
|
||||
USER pentester
|
||||
|
||||
RUN pipx install bandit
|
||||
RUN pipx install semgrep && \
|
||||
pipx install bandit
|
||||
|
||||
RUN npm install -g jshint
|
||||
|
||||
|
||||
@@ -1,37 +0,0 @@
|
||||
rules:
|
||||
- id: github-owned-action-not-pinned-to-commit-sha
|
||||
patterns:
|
||||
- pattern-inside: "{steps: ...}"
|
||||
- pattern: |
|
||||
uses: "$USES"
|
||||
- metavariable-regex:
|
||||
metavariable: $USES
|
||||
regex: '^(actions|github)/'
|
||||
- metavariable-pattern:
|
||||
metavariable: $USES
|
||||
language: generic
|
||||
patterns:
|
||||
- pattern-not-regex: "@[0-9a-f]{40}$"
|
||||
message: >-
|
||||
GitHub-owned action `$USES` is referenced by a mutable tag or branch
|
||||
instead of a full-length commit SHA. Tags can be moved, so a compromised
|
||||
or force-pushed release changes what this workflow runs. Pin to the
|
||||
commit SHA of the release you reviewed and let a bot keep it current.
|
||||
languages:
|
||||
- yaml
|
||||
severity: WARNING
|
||||
metadata:
|
||||
cwe:
|
||||
- "CWE-1357: Reliance on Insufficiently Trustworthy Component"
|
||||
- "CWE-353: Missing Support for Integrity Check"
|
||||
owasp: A06:2021 - Vulnerable and Outdated Components
|
||||
references:
|
||||
- https://docs.github.com/en/actions/security-guides/security-hardening-for-github-actions#using-third-party-actions
|
||||
category: security
|
||||
technology:
|
||||
- github-actions
|
||||
subcategory:
|
||||
- vuln
|
||||
likelihood: LOW
|
||||
impact: LOW
|
||||
confidence: HIGH
|
||||
@@ -19,22 +19,6 @@ Configure Strix using environment variables or a config file.
|
||||
Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_LLM_INPUT_COST_PER_MTOK" type="number">
|
||||
Price per 1M input tokens in USD for `STRIX_LLM`. Use it for gateway or self-hosted model ids that LiteLLM can't price.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_LLM_OUTPUT_COST_PER_MTOK" type="number">
|
||||
Price per 1M output tokens in USD for `STRIX_LLM`.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_LLM_CACHED_INPUT_COST_PER_MTOK" type="number">
|
||||
Price per 1M cached input tokens in USD for `STRIX_LLM`. If unset, cached tokens use the input price.
|
||||
</ParamField>
|
||||
|
||||
<Note>
|
||||
Setting any of these prices replaces the provider-reported and LiteLLM cost for `STRIX_LLM` in reports and `--max-budget`. An unset input or output price counts as `0`, so set both. A separate `STRIX_DEDUPE_MODEL` keeps its LiteLLM price.
|
||||
</Note>
|
||||
|
||||
<ParamField path="STRIX_API_TYPE" type="string">
|
||||
Select the OpenAI API path for the model: `responses` or `chat_completions`.
|
||||
By default the model decides: Responses when LiteLLM's model catalog lists
|
||||
@@ -186,7 +170,7 @@ When remote vars are set, Strix dual-writes telemetry to both local JSONL and th
|
||||
|
||||
## Docker Configuration
|
||||
|
||||
<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.4.0" type="string">
|
||||
<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.3.0" type="string">
|
||||
Docker image to use for the sandbox container.
|
||||
</ParamField>
|
||||
|
||||
|
||||
@@ -49,7 +49,7 @@ Strix runs inside a Kali Linux-based Docker container with a comprehensive set o
|
||||
|
||||
| Tool | Description |
|
||||
| ------------------------------------------------------- | --------------------------------------------- |
|
||||
| [Opengrep](https://github.com/opengrep/opengrep) | Fast SAST and custom rule matching |
|
||||
| [Semgrep](https://github.com/semgrep/semgrep) | Fast SAST and custom rule matching |
|
||||
| [ast-grep](https://ast-grep.github.io) | Structural AST/CST-aware code search (`sg`) |
|
||||
| [Tree-sitter](https://tree-sitter.github.io/tree-sitter/) | Syntax tree parsing and symbol extraction (Java/JS/TS/Python/Go/Bash/JSON/YAML grammars pre-configured) |
|
||||
| [Bandit](https://bandit.readthedocs.io) | Python security linter |
|
||||
|
||||
@@ -33,7 +33,7 @@ sqlmap -u "https://example.com/page?id=1"
|
||||
|
||||
```bash
|
||||
# Fast SAST triage
|
||||
opengrep scan --config /opt/opengrep-rules ./src
|
||||
semgrep --config auto ./src
|
||||
|
||||
# Structural AST search
|
||||
sg scan ./src
|
||||
|
||||
@@ -46,7 +46,7 @@ Thorough penetration testing. Best for:
|
||||
|
||||
**Duration**: 1-4 hours depending on target complexity
|
||||
|
||||
**White-box behavior**: Runs broad source-aware triage (`opengrep`, AST structural search, secrets, supply-chain checks) and then systematically validates top candidates dynamically.
|
||||
**White-box behavior**: Runs broad source-aware triage (`semgrep`, AST structural search, secrets, supply-chain checks) and then systematically validates top candidates dynamically.
|
||||
|
||||
<Note>
|
||||
Deep mode is the default. It explores edge cases, chained vulnerabilities, and complex attack paths.
|
||||
|
||||
+1
-1
@@ -4,7 +4,7 @@ set -euo pipefail
|
||||
|
||||
APP=strix
|
||||
REPO="usestrix/strix"
|
||||
STRIX_IMAGE="ghcr.io/usestrix/strix-sandbox:1.4.0"
|
||||
STRIX_IMAGE="ghcr.io/usestrix/strix-sandbox:1.3.0"
|
||||
|
||||
MUTED='\033[0;2m'
|
||||
RED='\033[0;31m'
|
||||
|
||||
@@ -115,11 +115,11 @@ BLACK-BOX TESTING (domain/subdomain only):
|
||||
|
||||
WHITE-BOX TESTING (code provided):
|
||||
- MUST perform BOTH static AND dynamic analysis
|
||||
- Static: Use source-aware triage first to map risk quickly (`opengrep`, `ast-grep`, Tree-sitter tooling, `gitleaks`, `trufflehog`, `trivy fs`). Then review code for vulnerabilities
|
||||
- Static: Use source-aware triage first to map risk quickly (`semgrep`, `ast-grep`, Tree-sitter tooling, `gitleaks`, `trufflehog`, `trivy fs`). Then review code for vulnerabilities
|
||||
- Static coverage floor: execute at least one structural AST mapping pass (`sg` and/or Tree-sitter) per repository and keep artifact output
|
||||
- Static coverage target per repository: run one `opengrep` pass, one secrets pass (`gitleaks` and/or `trufflehog`), one `trivy fs` pass, and one AST-structural pass (`sg` and/or Tree-sitter)
|
||||
- Static coverage target per repository: run one `semgrep` pass, one secrets pass (`gitleaks` and/or `trufflehog`), one `trivy fs` pass, and one AST-structural pass (`sg` and/or Tree-sitter)
|
||||
- Keep AST artifacts bounded and high-signal: scope to relevant paths/hypotheses, avoid whole-repo generic function dumps
|
||||
- AST target selection rule: build `sg-targets.txt` from `opengrep.json` scope first (`paths.scanned`, fallback to unique `results[].path`), then run `xargs ... sg run` against that file list. Only use path-heuristic fallback if opengrep scope is unavailable.
|
||||
- AST target selection rule: build `sg-targets.txt` from `semgrep.json` scope first (`paths.scanned`, fallback to unique `results[].path`), then run `xargs ... sg run` against that file list. Only use path-heuristic fallback if semgrep scope is unavailable.
|
||||
- Dynamic: Run the application and test live to validate exploitability
|
||||
- NEVER rely solely on static code analysis when dynamic validation is possible
|
||||
- Begin with fast source triage and dynamic run preparation in parallel; use static findings to prioritize live testing.
|
||||
@@ -188,7 +188,7 @@ EFFICIENCY TACTICS:
|
||||
- For Caido proxy automation inside Python, explicitly import from
|
||||
`caido_api`:
|
||||
`from caido_api import list_requests, view_request, repeat_request, list_sitemap, view_sitemap_entry, scope_rules`
|
||||
- Prefer established fuzzers/scanners where applicable: ffuf, sqlmap, nuclei, wapiti, arjun, httpx, katana, opengrep, bandit, trufflehog, nmap. Use scripts mainly to coordinate or validate around them, not to replace them without reason
|
||||
- Prefer established fuzzers/scanners where applicable: ffuf, sqlmap, nuclei, wapiti, arjun, httpx, katana, semgrep, bandit, trufflehog, nmap. Use scripts mainly to coordinate or validate around them, not to replace them without reason
|
||||
- For trial-heavy vectors (SQLi, XSS, XXE, SSRF, RCE, auth/JWT, deserialization), DO NOT iterate payloads manually in the browser. Always spray payloads via Python scripts through `exec_command` or terminal tools.
|
||||
- When using established fuzzers/scanners, use the proxy for inspection where helpful
|
||||
- Generate/adapt large payload corpora: combine encodings (URL, unicode, base64), comment styles, wrappers, time-based/differential probes. Expand with wordlists/templates
|
||||
@@ -458,7 +458,7 @@ JAVASCRIPT ANALYSIS:
|
||||
- js-beautify - JS beautifier/deobfuscator
|
||||
|
||||
CODE ANALYSIS:
|
||||
- opengrep - Static analysis/SAST (bundled rules in /opt/opengrep-rules)
|
||||
- semgrep - Static analysis/SAST
|
||||
- ast-grep (sg) - Structural AST/CST-aware code search
|
||||
- tree-sitter - Syntax-aware parsing and symbol extraction support
|
||||
- bandit - Python security linter
|
||||
|
||||
@@ -61,15 +61,6 @@ class LlmSettings(BaseSettings):
|
||||
repr=False,
|
||||
)
|
||||
reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
|
||||
input_cost_per_mtok: float | None = Field(
|
||||
default=None, ge=0, allow_inf_nan=False, alias="STRIX_LLM_INPUT_COST_PER_MTOK"
|
||||
)
|
||||
output_cost_per_mtok: float | None = Field(
|
||||
default=None, ge=0, allow_inf_nan=False, alias="STRIX_LLM_OUTPUT_COST_PER_MTOK"
|
||||
)
|
||||
cached_input_cost_per_mtok: float | None = Field(
|
||||
default=None, ge=0, allow_inf_nan=False, alias="STRIX_LLM_CACHED_INPUT_COST_PER_MTOK"
|
||||
)
|
||||
force_required_tool_choice: bool = Field(
|
||||
default=False,
|
||||
alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
|
||||
@@ -164,7 +155,7 @@ class RuntimeSettings(BaseSettings):
|
||||
model_config = _BASE_CONFIG
|
||||
|
||||
image: str = Field(
|
||||
default="ghcr.io/usestrix/strix-sandbox:1.4.0",
|
||||
default="ghcr.io/usestrix/strix-sandbox:1.3.0",
|
||||
alias="STRIX_IMAGE",
|
||||
)
|
||||
backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
|
||||
|
||||
@@ -130,6 +130,10 @@ _DETAIL_PRIORITY_KEYS = (
|
||||
"finding_type",
|
||||
"cve",
|
||||
"cwe",
|
||||
"attack_techniques",
|
||||
"attack_source",
|
||||
"atlas_techniques",
|
||||
"atlas_source",
|
||||
"cvss",
|
||||
"filed_at",
|
||||
"target",
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
"""MITRE ATLAS technique ids for findings in AI systems (LLM apps, agents, RAG).
|
||||
|
||||
The catalog is a pinned snapshot of the ATLAS techniques, generated from
|
||||
MITRE's ``atlas-data`` release.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from functools import cache
|
||||
from importlib.resources import files
|
||||
from typing import Any
|
||||
|
||||
|
||||
_TECHNIQUE_ID = re.compile(r"\bAML\.T\d{4}(?:\.\d{3})?\b", re.IGNORECASE)
|
||||
|
||||
|
||||
@cache
|
||||
def _catalog() -> dict[str, Any]:
|
||||
raw = files("strix.report").joinpath("atlas_catalog.json").read_text(encoding="utf-8")
|
||||
data: dict[str, Any] = json.loads(raw)
|
||||
return data
|
||||
|
||||
|
||||
def atlas_version() -> str:
|
||||
return str(_catalog()["version"])
|
||||
|
||||
|
||||
def technique(technique_id: str) -> dict[str, Any] | None:
|
||||
entry = _catalog()["techniques"].get(technique_id)
|
||||
return entry if isinstance(entry, dict) else None
|
||||
|
||||
|
||||
def technique_url(technique_id: str) -> str:
|
||||
return f"https://atlas.mitre.org/techniques/{technique_id}"
|
||||
|
||||
|
||||
def normalize_atlas_techniques(raw: Any) -> tuple[list[str], list[str]]:
|
||||
"""Parse ATLAS ids from a list or string and check them against the catalog.
|
||||
|
||||
Returns the ids in first-seen order and one error per id that is not a
|
||||
known ATLAS technique.
|
||||
"""
|
||||
if raw is None:
|
||||
return [], []
|
||||
items = raw if isinstance(raw, list) else [raw]
|
||||
techniques: list[str] = []
|
||||
errors: list[str] = []
|
||||
for item in items:
|
||||
text = str(item or "").strip()
|
||||
if not text:
|
||||
continue
|
||||
found = [match.upper() for match in _TECHNIQUE_ID.findall(text)]
|
||||
if not found:
|
||||
errors.append(f"invalid ATLAS technique: '{text}' (expected an id like 'AML.T0051')")
|
||||
continue
|
||||
for technique_id in found:
|
||||
if technique_id in techniques:
|
||||
continue
|
||||
if technique(technique_id) is None:
|
||||
errors.append(
|
||||
f"unknown ATLAS technique: '{technique_id}' is not a "
|
||||
f"MITRE ATLAS v{atlas_version()} technique"
|
||||
)
|
||||
continue
|
||||
techniques.append(technique_id)
|
||||
return techniques, errors
|
||||
@@ -0,0 +1 @@
|
||||
{"version":"2026.09","techniques":{"AML.T0000":{"name":"Search Open Technical Databases"},"AML.T0000.000":{"name":"Journals and Conference Proceedings"},"AML.T0000.001":{"name":"Pre-Print Repositories"},"AML.T0000.002":{"name":"Technical Blogs"},"AML.T0000.003":{"name":"Scan Databases"},"AML.T0001":{"name":"Search Open AI Vulnerability Analysis"},"AML.T0002":{"name":"Acquire Public AI Artifacts"},"AML.T0002.000":{"name":"Datasets"},"AML.T0002.001":{"name":"Models"},"AML.T0002.002":{"name":"AI Agent Configuration"},"AML.T0003":{"name":"Search Victim-Owned Websites"},"AML.T0004":{"name":"Search Application Repositories"},"AML.T0005":{"name":"Create Proxy AI Model"},"AML.T0005.000":{"name":"Train Proxy via Gathered AI Artifacts"},"AML.T0005.001":{"name":"Train Proxy via Replication"},"AML.T0005.002":{"name":"Use Pre-Trained Model"},"AML.T0006":{"name":"Active Scanning"},"AML.T0006.000":{"name":"Enumerate Hosted AI Resources"},"AML.T0006.001":{"name":"Query Platform Metadata APIs"},"AML.T0006.002":{"name":"Scan for Exposed AI Infrastructure"},"AML.T0006.003":{"name":"Probe AI Agent Trigger Channels"},"AML.T0007":{"name":"Discover AI Artifacts"},"AML.T0008":{"name":"Acquire Infrastructure"},"AML.T0008.000":{"name":"AI Development Workspaces"},"AML.T0008.001":{"name":"Consumer Hardware"},"AML.T0008.002":{"name":"Domains"},"AML.T0008.003":{"name":"Physical Countermeasures"},"AML.T0008.004":{"name":"Serverless"},"AML.T0008.005":{"name":"AI Service Proxies"},"AML.T0010":{"name":"AI Supply Chain Compromise"},"AML.T0010.000":{"name":"Hardware"},"AML.T0010.001":{"name":"AI Software"},"AML.T0010.002":{"name":"Data"},"AML.T0010.003":{"name":"Model"},"AML.T0010.004":{"name":"Container Registry"},"AML.T0010.005":{"name":"AI Agent Tool"},"AML.T0011":{"name":"User Execution"},"AML.T0011.000":{"name":"Unsafe AI Artifacts"},"AML.T0011.001":{"name":"Malicious Package"},"AML.T0011.002":{"name":"Poisoned AI Agent Tool"},"AML.T0011.003":{"name":"Malicious Link"},"AML.T0012":{"name":"Valid Accounts"},"AML.T0013":{"name":"Discover AI Model Ontology"},"AML.T0014":{"name":"Discover AI Model Family"},"AML.T0015":{"name":"Evade AI Model"},"AML.T0016":{"name":"Obtain Capabilities"},"AML.T0016.000":{"name":"Adversarial AI Attack Implementations"},"AML.T0016.001":{"name":"Software Tools"},"AML.T0016.002":{"name":"Generative AI"},"AML.T0016.003":{"name":"Exploits"},"AML.T0016.004":{"name":"AI Agent Tools"},"AML.T0017":{"name":"Develop Capabilities"},"AML.T0017.000":{"name":"Adversarial AI Attacks"},"AML.T0017.001":{"name":"Autonomous Exploit Development"},"AML.T0017.002":{"name":"AI Agent Tools"},"AML.T0018":{"name":"Manipulate AI Model"},"AML.T0018.000":{"name":"Poison AI Model"},"AML.T0018.001":{"name":"Modify AI Model Architecture"},"AML.T0018.002":{"name":"Embed Malware"},"AML.T0018.003":{"name":"Modify Prompt Construction Logic"},"AML.T0020":{"name":"Training Data Poisoning"},"AML.T0021":{"name":"Establish Accounts"},"AML.T0024":{"name":"Exfiltration via AI Inference API"},"AML.T0024.000":{"name":"Infer Training Data Membership"},"AML.T0024.001":{"name":"Invert AI Model"},"AML.T0024.002":{"name":"Extract AI Model"},"AML.T0025":{"name":"Exfiltration via Cyber Means"},"AML.T0029":{"name":"Denial of AI Service"},"AML.T0031":{"name":"Erode AI Model Integrity"},"AML.T0034":{"name":"Cost Harvesting"},"AML.T0034.000":{"name":"Excessive Queries"},"AML.T0034.001":{"name":"Resource-Intensive Queries"},"AML.T0034.002":{"name":"Agentic Resource Consumption"},"AML.T0035":{"name":"AI Artifact Collection"},"AML.T0036":{"name":"Data from Information Repositories"},"AML.T0037":{"name":"Data from Local System"},"AML.T0040":{"name":"AI Model Inference API Access"},"AML.T0041":{"name":"Physical Environment Access"},"AML.T0042":{"name":"Verify Attack"},"AML.T0043":{"name":"Craft Adversarial Data"},"AML.T0043.000":{"name":"White-Box Optimization"},"AML.T0043.001":{"name":"Black-Box Optimization"},"AML.T0043.002":{"name":"Black-Box Transfer"},"AML.T0043.003":{"name":"Manual Modification"},"AML.T0043.004":{"name":"Insert Backdoor Trigger"},"AML.T0044":{"name":"Full AI Model Access"},"AML.T0046":{"name":"Spamming AI System with Chaff Data"},"AML.T0047":{"name":"AI-Enabled Product or Service"},"AML.T0048":{"name":"External Harms"},"AML.T0048.000":{"name":"Financial Harm"},"AML.T0048.001":{"name":"Reputational Harm"},"AML.T0048.002":{"name":"Societal Harm"},"AML.T0048.003":{"name":"User Harm"},"AML.T0048.004":{"name":"AI Intellectual Property Theft"},"AML.T0049":{"name":"Exploit Public-Facing Application"},"AML.T0050":{"name":"Command and Scripting Interpreter"},"AML.T0051":{"name":"LLM Prompt Injection"},"AML.T0051.000":{"name":"Direct"},"AML.T0051.001":{"name":"Indirect"},"AML.T0051.002":{"name":"Triggered"},"AML.T0052":{"name":"Phishing"},"AML.T0052.000":{"name":"Spearphishing via Social Engineering LLM"},"AML.T0052.001":{"name":"Deepfake-Assisted Phishing"},"AML.T0053":{"name":"AI Agent Tool Invocation"},"AML.T0054":{"name":"LLM Jailbreak"},"AML.T0055":{"naLine truncated
|
||||
@@ -0,0 +1,70 @@
|
||||
"""MITRE ATT&CK Enterprise technique ids for findings.
|
||||
|
||||
The catalog is a pinned snapshot of the active (not revoked, not deprecated)
|
||||
Enterprise techniques, generated from MITRE's ``attack-stix-data`` release.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from functools import cache
|
||||
from importlib.resources import files
|
||||
from typing import Any
|
||||
|
||||
|
||||
_TECHNIQUE_ID = re.compile(r"\bT\d{4}(?:\.\d{3})?\b", re.IGNORECASE)
|
||||
|
||||
|
||||
@cache
|
||||
def _catalog() -> dict[str, Any]:
|
||||
raw = files("strix.report").joinpath("attack_catalog.json").read_text(encoding="utf-8")
|
||||
data: dict[str, Any] = json.loads(raw)
|
||||
return data
|
||||
|
||||
|
||||
def attack_version() -> str:
|
||||
return str(_catalog()["version"])
|
||||
|
||||
|
||||
def technique(technique_id: str) -> dict[str, Any] | None:
|
||||
entry = _catalog()["techniques"].get(technique_id)
|
||||
return entry if isinstance(entry, dict) else None
|
||||
|
||||
|
||||
def technique_url(technique_id: str) -> str:
|
||||
return f"https://attack.mitre.org/techniques/{technique_id.replace('.', '/')}/"
|
||||
|
||||
|
||||
def normalize_attack_techniques(raw: Any) -> tuple[list[str], list[str]]:
|
||||
"""Parse technique ids from a list or string and check them against the catalog.
|
||||
|
||||
Returns the ids in first-seen order and one error per id that is not an
|
||||
active Enterprise technique.
|
||||
"""
|
||||
if raw is None:
|
||||
return [], []
|
||||
items = raw if isinstance(raw, list) else [raw]
|
||||
techniques: list[str] = []
|
||||
errors: list[str] = []
|
||||
for item in items:
|
||||
text = str(item or "").strip()
|
||||
if not text:
|
||||
continue
|
||||
found = [match.upper() for match in _TECHNIQUE_ID.findall(text)]
|
||||
if not found:
|
||||
errors.append(
|
||||
f"invalid ATT&CK technique: '{text}' (expected an id like 'T1190' or 'T1059.007')"
|
||||
)
|
||||
continue
|
||||
for technique_id in found:
|
||||
if technique_id in techniques:
|
||||
continue
|
||||
if technique(technique_id) is None:
|
||||
errors.append(
|
||||
f"unknown ATT&CK technique: '{technique_id}' is not an active "
|
||||
f"MITRE ATT&CK Enterprise v{attack_version()} technique"
|
||||
)
|
||||
continue
|
||||
techniques.append(technique_id)
|
||||
return techniques, errors
|
||||
@@ -0,0 +1 @@
|
||||
{"domain":"enterprise-attack","version":"20.0","techniques":{"T1001":{"name":"Data Obfuscation","tactics":["command-and-control"]},"T1001.001":{"name":"Junk Data","tactics":["command-and-control"]},"T1001.002":{"name":"Steganography","tactics":["command-and-control"]},"T1001.003":{"name":"Protocol or Service Impersonation","tactics":["command-and-control"]},"T1003":{"name":"OS Credential Dumping","tactics":["credential-access"]},"T1003.001":{"name":"LSASS Memory","tactics":["credential-access"]},"T1003.002":{"name":"Security Account Manager","tactics":["credential-access"]},"T1003.003":{"name":"NTDS","tactics":["credential-access"]},"T1003.004":{"name":"LSA Secrets","tactics":["credential-access"]},"T1003.005":{"name":"Cached Domain Credentials","tactics":["credential-access"]},"T1003.006":{"name":"DCSync","tactics":["credential-access"]},"T1003.007":{"name":"Proc Filesystem","tactics":["credential-access"]},"T1003.008":{"name":"/etc/passwd and /etc/shadow","tactics":["credential-access"]},"T1005":{"name":"Data from Local System","tactics":["collection"]},"T1006":{"name":"Direct Volume Access","tactics":["stealth"]},"T1007":{"name":"System Service Discovery","tactics":["discovery"]},"T1008":{"name":"Fallback Channels","tactics":["command-and-control"]},"T1010":{"name":"Application Window Discovery","tactics":["discovery"]},"T1011":{"name":"Exfiltration Over Other Network Medium","tactics":["exfiltration"]},"T1011.001":{"name":"Exfiltration Over Bluetooth","tactics":["exfiltration"]},"T1012":{"name":"Query Registry","tactics":["discovery"]},"T1014":{"name":"Rootkit","tactics":["stealth"]},"T1016":{"name":"System Network Configuration Discovery","tactics":["discovery"]},"T1016.001":{"name":"Internet Connection Discovery","tactics":["discovery"]},"T1016.002":{"name":"Wi-Fi Discovery","tactics":["discovery"]},"T1018":{"name":"Remote System Discovery","tactics":["discovery"]},"T1020":{"name":"Automated Exfiltration","tactics":["exfiltration"]},"T1020.001":{"name":"Traffic Duplication","tactics":["exfiltration"]},"T1021":{"name":"Remote Services","tactics":["lateral-movement"]},"T1021.001":{"name":"Remote Desktop Protocol","tactics":["lateral-movement"]},"T1021.002":{"name":"SMB/Windows Admin Shares","tactics":["lateral-movement"]},"T1021.003":{"name":"Distributed Component Object Model","tactics":["lateral-movement"]},"T1021.004":{"name":"SSH","tactics":["lateral-movement"]},"T1021.005":{"name":"VNC","tactics":["lateral-movement"]},"T1021.006":{"name":"Windows Remote Management","tactics":["lateral-movement"]},"T1021.007":{"name":"Cloud Services","tactics":["lateral-movement"]},"T1021.008":{"name":"Direct Cloud VM Connections","tactics":["lateral-movement"]},"T1025":{"name":"Data from Removable Media","tactics":["collection"]},"T1027":{"name":"Obfuscated Files or Information","tactics":["stealth"]},"T1027.001":{"name":"Binary Padding","tactics":["stealth"]},"T1027.002":{"name":"Software Packing","tactics":["stealth"]},"T1027.003":{"name":"Steganography","tactics":["stealth"]},"T1027.004":{"name":"Compile After Delivery","tactics":["stealth"]},"T1027.005":{"name":"Indicator Removal from Tools","tactics":["stealth"]},"T1027.006":{"name":"HTML Smuggling","tactics":["stealth"]},"T1027.007":{"name":"Dynamic API Resolution","tactics":["stealth"]},"T1027.008":{"name":"Stripped Payloads","tactics":["stealth"]},"T1027.009":{"name":"Embedded Payloads","tactics":["stealth"]},"T1027.010":{"name":"Command Obfuscation","tactics":["stealth"]},"T1027.011":{"name":"Fileless Storage","tactics":["stealth"]},"T1027.012":{"name":"LNK Icon Smuggling","tactics":["stealth"]},"T1027.013":{"name":"Encrypted/Encoded File","tactics":["stealth"]},"T1027.014":{"name":"Polymorphic Code","tactics":["stealth"]},"T1027.015":{"name":"Compression","tactics":["stealth"]},"T1027.016":{"name":"Junk Code Insertion","tactics":["stealth"]},"T1027.017":{"name":"SVG Smuggling","tactics":["stealth"]},"T1027.018":{"name":"Invisible Unicode","tactics":["stealth"]},"T1029":{"name":"Scheduled Transfer","tactics":["exfiltration"]},"T1030":{"name":"Data Transfer Size Limits","tactics":["exfiltration"]},"T1033":{"name":"System Owner/User Discovery","tactics":["discovery"]},"T1036":{"name":"Masquerading","tactics":["stealth"]},"T1036.001":{"name":"Invalid Code Signature","tactics":["stealth"]},"T1036.002":{"name":"Right-to-Left Override","tactics":["stealth"]},"T1036.003":{"name":"Rename Legitimate Utilities","tactics":["stealth"]},"T1036.004":{"name":"Masquerade Task or Service","tactics":["stealth"]},"T1036.005":{"name":"Match Legitimate Resource Name or Location","tactics":["stealth"]},"T1036.006":{"name":"Space after Filename","tactics":["stealth"]},"T1036.007":{"name":"Double File Extension","tactics":["stealth"]},"T1036.008":{"name":"Masquerade File Type","tactics":["stealth"]},"T1036.009":{"name":"Break Process Trees","tactics":["stealth"]},"T1036.010":{"name":"Masquerade Account Name","tactics":["stealth"]},"T1036.011":{"name":"Overwrite Process ArgumentsLine truncated
|
||||
+2
-20
@@ -300,15 +300,7 @@ def _parse_dedupe_response(content: str) -> dict[str, Any]:
|
||||
raise ValueError(f"No JSON object found in dedupe response: {content[:500]}")
|
||||
parsed = json.loads(text[start : end + 1])
|
||||
|
||||
is_duplicate = parsed.get("is_duplicate")
|
||||
if not isinstance(is_duplicate, bool):
|
||||
raise TypeError("is_duplicate must be a boolean")
|
||||
duplicate_id = parsed.get("duplicate_id") or ""
|
||||
if is_duplicate and (not isinstance(duplicate_id, str) or not duplicate_id):
|
||||
raise ValueError("A duplicate decision must name an existing report ID")
|
||||
# Keep the ID verbatim: truncating or stripping it could turn an unknown ID
|
||||
# into a different, existing one before the membership check below.
|
||||
duplicate_id = str(duplicate_id)
|
||||
duplicate_id = str(parsed.get("duplicate_id") or "")[:64]
|
||||
reason = str(parsed.get("reason") or "")[:500]
|
||||
try:
|
||||
confidence = float(parsed.get("confidence", 0.0))
|
||||
@@ -316,22 +308,13 @@ def _parse_dedupe_response(content: str) -> dict[str, Any]:
|
||||
confidence = 0.0
|
||||
|
||||
return {
|
||||
"is_duplicate": is_duplicate,
|
||||
"is_duplicate": bool(parsed.get("is_duplicate", False)),
|
||||
"duplicate_id": duplicate_id,
|
||||
"confidence": confidence,
|
||||
"reason": reason,
|
||||
}
|
||||
|
||||
|
||||
def _validate_duplicate_reference(
|
||||
result: dict[str, Any], existing_reports: list[dict[str, Any]]
|
||||
) -> None:
|
||||
if result["is_duplicate"] and not any(
|
||||
report.get("id") == result["duplicate_id"] for report in existing_reports
|
||||
):
|
||||
raise ValueError("Duplicate decision refers to an unknown report ID")
|
||||
|
||||
|
||||
def _extract_text(response: ModelResponse) -> str:
|
||||
parts: list[str] = []
|
||||
for item in response.output:
|
||||
@@ -414,7 +397,6 @@ async def check_duplicate(
|
||||
}
|
||||
|
||||
result = _parse_dedupe_response(content)
|
||||
_validate_duplicate_reference(result, existing_reports)
|
||||
|
||||
logger.info(
|
||||
"Deduplication check: is_duplicate=%s, confidence=%.2f, reason=%s",
|
||||
|
||||
@@ -543,6 +543,8 @@ def _result_properties(
|
||||
"method",
|
||||
"cve",
|
||||
"cwe",
|
||||
"attack_techniques",
|
||||
"atlas_techniques",
|
||||
"impact",
|
||||
"technical_analysis",
|
||||
"remediation_steps",
|
||||
|
||||
@@ -91,6 +91,8 @@ UPDATABLE_REPORT_FIELDS = frozenset(
|
||||
"method",
|
||||
"cve",
|
||||
"cwe",
|
||||
"attack_techniques",
|
||||
"atlas_techniques",
|
||||
"code_locations",
|
||||
"http_exchange_ids",
|
||||
"fix_verification",
|
||||
@@ -353,6 +355,8 @@ class ReportState:
|
||||
method: str | None = None,
|
||||
cve: str | None = None,
|
||||
cwe: str | None = None,
|
||||
attack_techniques: list[str] | None = None,
|
||||
atlas_techniques: list[str] | None = None,
|
||||
code_locations: list[dict[str, Any]] | None = None,
|
||||
http_exchange_ids: list[str] | None = None,
|
||||
fix_verification: str | None = None,
|
||||
@@ -411,6 +415,10 @@ class ReportState:
|
||||
report["cve"] = cve.strip()
|
||||
if cwe:
|
||||
report["cwe"] = cwe.strip()
|
||||
if attack_techniques:
|
||||
report["attack_techniques"] = list(attack_techniques)
|
||||
if atlas_techniques:
|
||||
report["atlas_techniques"] = list(atlas_techniques)
|
||||
if code_locations:
|
||||
report["code_locations"] = code_locations
|
||||
if http_exchange_ids:
|
||||
|
||||
+2
-51
@@ -8,38 +8,11 @@ from typing import Any
|
||||
from agents.usage import Usage, deserialize_usage, serialize_usage
|
||||
from pydantic import BaseModel, TypeAdapter, ValidationError
|
||||
|
||||
from strix.config.loader import load_settings
|
||||
from strix.report.pricing import resolve_litellm_model
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
Prices = tuple[float | None, float | None, float | None]
|
||||
|
||||
|
||||
def _configured_prices() -> tuple[str | None, Prices | None]:
|
||||
llm = load_settings().llm
|
||||
prices = (llm.input_cost_per_mtok, llm.output_cost_per_mtok, llm.cached_input_cost_per_mtok)
|
||||
if all(price is None for price in prices):
|
||||
return None, None
|
||||
return (llm.model or "").strip() or None, prices
|
||||
|
||||
|
||||
def _configured_usage_cost(usage: Usage, prices: Prices) -> float:
|
||||
input_rate, output_rate, cached_rate = prices
|
||||
input_tokens = max(0, _int_or_zero(getattr(usage, "input_tokens", 0)))
|
||||
cached_details = getattr(usage, "input_tokens_details", None)
|
||||
cached_tokens = max(0, _int_or_zero(getattr(cached_details, "cached_tokens", 0)))
|
||||
cached_tokens = min(cached_tokens, input_tokens)
|
||||
output_tokens = max(0, _int_or_zero(getattr(usage, "output_tokens", 0)))
|
||||
billable_input = input_tokens - cached_tokens
|
||||
cost = (
|
||||
billable_input * (input_rate or 0)
|
||||
+ cached_tokens * (cached_rate if cached_rate is not None else input_rate or 0)
|
||||
+ output_tokens * (output_rate or 0)
|
||||
) / 1_000_000
|
||||
return _round_cost(cost)
|
||||
|
||||
|
||||
class ProviderUsage(BaseModel):
|
||||
"""Running totals for one upstream provider OpenRouter routed calls to."""
|
||||
@@ -64,7 +37,6 @@ class LLMUsageLedger:
|
||||
self._total_usage = Usage()
|
||||
self._agent_usage: dict[str, Usage] = {}
|
||||
self._agent_metadata: dict[str, dict[str, str]] = {}
|
||||
self._agent_estimated_cost: dict[str, float] = {}
|
||||
self._observed_cost = 0.0
|
||||
self._estimated_cost = 0.0
|
||||
self._has_observed_cost = False
|
||||
@@ -101,9 +73,6 @@ class LLMUsageLedger:
|
||||
estimated = _estimate_litellm_cost(usage, model)
|
||||
if estimated:
|
||||
self._estimated_cost += estimated
|
||||
self._agent_estimated_cost[normalized_agent_id] = (
|
||||
self._agent_estimated_cost.get(normalized_agent_id, 0.0) + estimated
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
@@ -145,23 +114,11 @@ class LLMUsageLedger:
|
||||
def total_cost(self) -> float:
|
||||
if self.zero_cost:
|
||||
return 0.0
|
||||
model, prices = _configured_prices()
|
||||
if prices is not None:
|
||||
return _round_cost(
|
||||
sum(self._configured_agent_cost(aid, model, prices) for aid in self._agent_usage)
|
||||
)
|
||||
return _round_cost(self._observed_cost if self._has_observed_cost else self._estimated_cost)
|
||||
|
||||
def _configured_agent_cost(self, agent_id: str, model: str | None, prices: Prices) -> float:
|
||||
if self._agent_metadata.get(agent_id, {}).get("model") == model:
|
||||
return _configured_usage_cost(self._agent_usage[agent_id], prices)
|
||||
return self._agent_estimated_cost.get(agent_id, 0.0)
|
||||
|
||||
def to_record(self) -> dict[str, Any]:
|
||||
record = serialize_usage(self._total_usage)
|
||||
model, prices = _configured_prices()
|
||||
total_cost = self.total_cost
|
||||
record["cost"] = total_cost
|
||||
record["cost"] = self.total_cost
|
||||
record["providers"] = {name: tally.model_dump() for name, tally in self._providers.items()}
|
||||
record["agents"] = []
|
||||
|
||||
@@ -171,11 +128,7 @@ class LLMUsageLedger:
|
||||
usage = self._agent_usage[agent_id]
|
||||
metadata = self._agent_metadata.get(agent_id, {})
|
||||
agent_cost = (
|
||||
self._configured_agent_cost(agent_id, model, prices)
|
||||
if prices is not None and not self.zero_cost
|
||||
else 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)
|
||||
@@ -195,7 +148,6 @@ class LLMUsageLedger:
|
||||
self._total_usage = Usage()
|
||||
self._agent_usage.clear()
|
||||
self._agent_metadata.clear()
|
||||
self._agent_estimated_cost.clear()
|
||||
self._observed_cost = 0.0
|
||||
self._estimated_cost = 0.0
|
||||
self._has_observed_cost = False
|
||||
@@ -231,7 +183,6 @@ class LLMUsageLedger:
|
||||
logger.exception("Failed to hydrate llm_usage for agent %s", agent_id)
|
||||
self._agent_usage[agent_id] = Usage()
|
||||
|
||||
self._agent_estimated_cost[agent_id] = _float_or_zero(raw_agent.get("cost"))
|
||||
metadata: dict[str, str] = {}
|
||||
agent_name = raw_agent.get("agent_name")
|
||||
model = raw_agent.get("model")
|
||||
|
||||
@@ -241,6 +241,8 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
|
||||
("Method", report.get("method")),
|
||||
("CVE", report.get("cve")),
|
||||
("CWE", report.get("cwe")),
|
||||
("MITRE ATT&CK", ", ".join(report.get("attack_techniques") or [])),
|
||||
("MITRE ATLAS", ", ".join(report.get("atlas_techniques") or [])),
|
||||
]
|
||||
cvss = report.get("cvss")
|
||||
if cvss is not None:
|
||||
|
||||
@@ -40,7 +40,7 @@ The skills are dynamically injected into the agent's system prompt, allowing it
|
||||
|
||||
Notable source-aware skills:
|
||||
- `source_aware_whitebox` (coordination): white-box orchestration playbook
|
||||
- `source_aware_sast` (custom): opengrep/AST/secrets/supply-chain static triage workflow
|
||||
- `source_aware_sast` (custom): semgrep/AST/secrets/supply-chain static triage workflow
|
||||
- `dependency_cve_scanning` (custom): trivy-based SCA workflow for reporting known dependency CVEs via `create_dependency_report`
|
||||
- `npx_confusion` (custom): npx/npm exec/bunx fallback and adjacent package-runner identity confusion, with runner-specific registry and reporting gates
|
||||
- `semantic_confusion` (vulnerabilities): cross-boundary parser, normalization, and representation mismatch analysis
|
||||
|
||||
@@ -58,13 +58,6 @@ Structure agents by function:
|
||||
- Business logic flaws
|
||||
- Infrastructure vulnerabilities
|
||||
|
||||
**Sensitive Browser Confirmations**
|
||||
- For every browser workflow containing a security-sensitive or consequential user action—whether authenticated or unauthenticated—assign `browser_security` to the agent testing that workflow or create a dedicated UI-redressing agent. This includes actions that change protected state, grant authority, submit credentials, expose sensitive data, or cross a trust boundary.
|
||||
- Its task must cover both iframe and cross-window/DoubleClickjacking paths; framing protections close only iframe coverage.
|
||||
- A workflow crossing origins or ports remains eligible when every involved origin is an explicit scan target. Same-origin policy blocks DOM reads, not top-level navigation, popups, opener relationships, focus changes, or window closing; cross-origin separation alone does not close UI-redressing coverage.
|
||||
- Never expand authorization from a link alone. If a required origin is not an explicit target, leave the path `needs_follow_up` and do not claim clickjacking or the related workflow is cleared; scope expansion must come from scan configuration.
|
||||
- Before accepting `no_issue_found` or `ruled_out` for cross-window UI redressing, reject closure evidence that cites popup-as-tab behavior, ignored geometry, unavailable OS-level input, or only a manual control click; keep the exact two-click path `needs_follow_up`. A successfully executed protected consent, approval, or grant must be impact-calibrated, not dismissed because it is idempotent, pre-seeded, or issues no immediate token or data.
|
||||
|
||||
**Exploitation and Validation**
|
||||
- Proof-of-concept development
|
||||
- Impact demonstration
|
||||
|
||||
@@ -14,22 +14,22 @@ Increase white-box coverage by combining source-aware triage with dynamic valida
|
||||
## Recommended Workflow
|
||||
|
||||
1. Build a quick source map before deep exploitation, including at least one AST-structural pass (`sg` or `tree-sitter`) scoped to relevant paths.
|
||||
- For `sg` baseline, derive `sg-targets.txt` from `opengrep.json` scope first (`paths.scanned`, fallback to unique `results[].path`) and run `xargs ... sg run` on that list.
|
||||
- Only fall back to path heuristics when opengrep scope is unavailable.
|
||||
- For `sg` baseline, derive `sg-targets.txt` from `semgrep.json` scope first (`paths.scanned`, fallback to unique `results[].path`) and run `xargs ... sg run` on that list.
|
||||
- Only fall back to path heuristics when semgrep scope is unavailable.
|
||||
2. Run first-pass static triage to rank high-risk paths.
|
||||
3. Use triage outputs to prioritize dynamic PoC validation.
|
||||
4. Keep findings evidence-driven: no report without validation.
|
||||
|
||||
## Source-Aware Triage Stack
|
||||
|
||||
- `opengrep`: fast security-first triage and custom pattern scans
|
||||
- `semgrep`: fast security-first triage and custom pattern scans
|
||||
- `ast-grep` (`sg`): structural pattern hunting and targeted repo mapping
|
||||
- `tree-sitter`: syntax-aware parsing support for symbol and route extraction
|
||||
- `gitleaks` + `trufflehog`: complementary secret detection (working tree and history coverage)
|
||||
- `trivy fs`: dependency, misconfiguration, license, and secret checks
|
||||
|
||||
Coverage target per repository:
|
||||
- one `opengrep` pass
|
||||
- one `semgrep` pass
|
||||
- one AST structural pass (`sg` and/or `tree-sitter`)
|
||||
- one secrets pass (`gitleaks` and/or `trufflehog`)
|
||||
- one `trivy fs` pass
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
---
|
||||
name: source-aware-sast
|
||||
description: Practical source-aware SAST and AST playbook for opengrep, ast-grep, gitleaks, and trivy fs
|
||||
description: Practical source-aware SAST and AST playbook for semgrep, ast-grep, gitleaks, and trivy fs
|
||||
---
|
||||
|
||||
# Source-Aware SAST Playbook
|
||||
@@ -23,19 +23,19 @@ Run this baseline once per repository before deep narrowing:
|
||||
ART=/workspace/.source-aware
|
||||
mkdir -p "$ART"
|
||||
|
||||
opengrep scan --config /opt/opengrep-rules --taint-intrafile \
|
||||
--json --output "$ART/opengrep.json" .
|
||||
# Build deterministic AST targets from opengrep scope (no hardcoded path guessing)
|
||||
semgrep scan --config p/default --config p/golang --config p/secrets \
|
||||
--metrics=off --json --output "$ART/semgrep.json" .
|
||||
# Build deterministic AST targets from semgrep scope (no hardcoded path guessing)
|
||||
python3 - <<'PY'
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
art = Path("/workspace/.source-aware")
|
||||
opengrep_json = art / "opengrep.json"
|
||||
semgrep_json = art / "semgrep.json"
|
||||
targets_file = art / "sg-targets.txt"
|
||||
|
||||
try:
|
||||
data = json.loads(opengrep_json.read_text(encoding="utf-8"))
|
||||
data = json.loads(semgrep_json.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
targets_file.write_text("", encoding="utf-8")
|
||||
raise
|
||||
@@ -63,18 +63,17 @@ trivy fs --scanners vuln,misconfig --timeout 30m --offline-scan \
|
||||
--format json --output "$ART/trivy-fs.json" . || true
|
||||
```
|
||||
|
||||
## Opengrep First Pass
|
||||
## Semgrep First Pass
|
||||
|
||||
Use Opengrep as the default static triage pass. Rules are bundled offline in `/opt/opengrep-rules` (one directory per language, plus `generic/secrets`):
|
||||
Use Semgrep as the default static triage pass:
|
||||
|
||||
```bash
|
||||
# Full bundled ruleset
|
||||
opengrep scan --config /opt/opengrep-rules --taint-intrafile \
|
||||
--json --output /workspace/.source-aware/opengrep.json .
|
||||
# Preferred deterministic profile set (works with --metrics=off)
|
||||
semgrep scan --config p/default --config p/golang --config p/secrets \
|
||||
--metrics=off --json --output /workspace/.source-aware/semgrep.json .
|
||||
|
||||
# Narrow to the target stack when the full set is slow or noisy
|
||||
opengrep scan --config /opt/opengrep-rules/python --config /opt/opengrep-rules/generic/secrets \
|
||||
--taint-intrafile --json --output /workspace/.source-aware/opengrep-python.json .
|
||||
# If you choose auto config, do not combine it with --metrics=off
|
||||
semgrep scan --config auto --json --output /workspace/.source-aware/semgrep-auto.json .
|
||||
```
|
||||
|
||||
If diff scope is active, restrict to changed files first, then expand only when needed.
|
||||
|
||||
@@ -205,10 +205,10 @@ Jinja2 backend without autoescape: `{{7*7}}`, RCE gadgets if sandbox misconfigur
|
||||
|
||||
## Tooling
|
||||
|
||||
Static analysis is the fastest way to reach the sinks above in white-box scope. The sandbox ships `python`/`pipx`, `opengrep`, `bandit`, `ast-grep`, and `ripgrep`.
|
||||
Static analysis is the fastest way to reach the sinks above in white-box scope. The sandbox ships `python`/`pipx`, `semgrep`, `bandit`, `ast-grep`, and `ripgrep`.
|
||||
|
||||
- **bandit** (preinstalled) — Python security linter; flags `mark_safe`, `extra()`, `RawSQL`, `subprocess`, weak crypto, hardcoded secrets: `bandit -r . -ll`
|
||||
- **opengrep** (preinstalled) with the bundled Django rules — higher-signal than bandit for framework-specific bugs (`.extra()`, `RawSQL`, `|safe`, `csrf_exempt`, `ALLOWED_HOSTS=['*']`): `opengrep scan --config /opt/opengrep-rules/python/django .`
|
||||
- **semgrep** (preinstalled) with the Django ruleset — higher-signal than bandit for framework-specific bugs (`.extra()`, `RawSQL`, `|safe`, `csrf_exempt`, `ALLOWED_HOSTS=['*']`): `semgrep --config p/django .`
|
||||
- **pip-audit** (PyPA) — dependency CVE scanner for known-vuln Django/DRF/simplejwt versions: `pipx install pip-audit && pip-audit -r requirements.txt`
|
||||
- **ast-grep** (preinstalled) — quick structural grep for risky calls without a full SAST run: `ast-grep run -p 'mark_safe($X)' -l python`
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@ Thorough understanding before exploitation. Test every parameter, every endpoint
|
||||
|
||||
**Whitebox (source available)**
|
||||
- Map every file, module, and code path in the repository
|
||||
- Start with broad source-aware triage (`opengrep`, `ast-grep`, `gitleaks`, `trufflehog`, `trivy fs`) and use outputs to drive deep review
|
||||
- Start with broad source-aware triage (`semgrep`, `ast-grep`, `gitleaks`, `trufflehog`, `trivy fs`) and use outputs to drive deep review
|
||||
- Execute at least one structural AST pass (`sg` and/or Tree-sitter) per repository and store artifacts for reuse
|
||||
- Keep AST artifacts bounded and query-driven (target relevant paths/sinks first; avoid whole-repo generic function dumps)
|
||||
- Use syntax-aware parsing (Tree-sitter tooling) to improve symbol, route, and sink extraction quality
|
||||
|
||||
@@ -15,7 +15,7 @@ Optimize for fast feedback on critical security issues. Skip exhaustive enumerat
|
||||
|
||||
**Whitebox (source available)**
|
||||
- Focus on recent changes: git diffs, new commits, modified files—these are most likely to contain fresh bugs
|
||||
- Run a fast static triage on changed files first (`opengrep`, then targeted `sg` queries)
|
||||
- Run a fast static triage on changed files first (`semgrep`, then targeted `sg` queries)
|
||||
- Run at least one lightweight AST pass (`sg` or Tree-sitter) so structural mapping is not skipped
|
||||
- Keep AST commands tightly scoped to changed or high-risk paths; avoid broad repository-wide pattern dumps
|
||||
- Run quick secret and dependency checks (`gitleaks`, `trufflehog`, `trivy fs`) scoped to changed areas when possible
|
||||
|
||||
@@ -15,7 +15,7 @@ Systematic testing across the full attack surface. Understand the application be
|
||||
|
||||
**Whitebox (source available)**
|
||||
- Map codebase structure: modules, entry points, routing
|
||||
- Run `opengrep` first-pass triage to prioritize risky flows before deep manual review
|
||||
- Run `semgrep` first-pass triage to prioritize risky flows before deep manual review
|
||||
- Run at least one AST-structural mapping pass (`sg` and/or Tree-sitter), then use outputs for route, sink, and trust-boundary mapping
|
||||
- Keep AST output bounded to relevant paths and hypotheses; avoid whole-repo generic function dumps
|
||||
- Identify architecture pattern (MVC, microservices, monolith)
|
||||
|
||||
@@ -1,69 +0,0 @@
|
||||
---
|
||||
name: opengrep
|
||||
description: Exact Opengrep CLI structure, bundled offline rules, scoped rule selection, taint mode, and automation-safe output patterns.
|
||||
---
|
||||
|
||||
# Opengrep CLI Playbook
|
||||
|
||||
Official docs:
|
||||
- https://github.com/opengrep/opengrep
|
||||
- https://github.com/opengrep/opengrep-rules
|
||||
|
||||
Canonical syntax:
|
||||
`opengrep scan [flags] [targets]`
|
||||
|
||||
Bundled rules:
|
||||
- `/opt/opengrep-rules` holds the full ruleset, one directory per language (`python`, `javascript`, `typescript`, `go`, `java`, `php`, `ruby`, …) plus `generic/secrets`, `dockerfile`, `terraform`, `yaml`, and `strix` (Strix-authored rules, e.g. GitHub-owned actions not pinned to a commit SHA)
|
||||
- Framework rules sit under the language (`python/django`, `python/flask`, `javascript/express`, …)
|
||||
|
||||
High-signal flags:
|
||||
- `--config <path>` rule file or directory; repeat for several
|
||||
- `--taint-intrafile` intra-file inter-procedural taint (follows data across functions in one file)
|
||||
- `--json` / `--sarif` output format
|
||||
- `--output <file>` write findings to file
|
||||
- `--severity <level>` filter by severity (`INFO`, `WARNING`, `ERROR`)
|
||||
- `--error` return non-zero exit when findings exist
|
||||
- `--quiet` suppress progress noise
|
||||
- `--jobs <n>` parallel workers
|
||||
- `--timeout <seconds>` per-rule, per-file timeout
|
||||
- `--exclude <pattern>` / `--include <pattern>` path filters
|
||||
- `--exclude-rule <rule_id>` suppress specific rule
|
||||
- `--baseline-commit <sha>` only report findings introduced after baseline
|
||||
|
||||
Agent-safe baseline for automation:
|
||||
`opengrep scan --config /opt/opengrep-rules --taint-intrafile --json --output opengrep.json --quiet --jobs 4 --timeout 20 /workspace`
|
||||
|
||||
Common patterns:
|
||||
- Full bundled scan:
|
||||
`opengrep scan --config /opt/opengrep-rules --json --output opengrep.json --quiet /workspace`
|
||||
- High-severity focused pass:
|
||||
`opengrep scan --config /opt/opengrep-rules --severity ERROR --json --output opengrep_high.json --quiet /workspace`
|
||||
- SARIF output:
|
||||
`opengrep scan --config /opt/opengrep-rules --sarif --output opengrep.sarif --quiet /workspace`
|
||||
- Language- or framework-specific rules:
|
||||
`opengrep scan --config /opt/opengrep-rules/python --config /opt/opengrep-rules/generic/secrets --json --output opengrep_python.json --quiet /workspace`
|
||||
- Scoped directory scan:
|
||||
`opengrep scan --config /opt/opengrep-rules --json --output opengrep_api.json --quiet /workspace/services/api`
|
||||
- Custom rule:
|
||||
`opengrep scan --config ./rules/my-rule.yaml --json --output opengrep_custom.json --quiet /workspace`
|
||||
|
||||
Critical correctness rules:
|
||||
- Always provide an explicit `--config` from `/opt/opengrep-rules` or a local rule file. Registry shorthands (`p/...`, `auto`) fetch rules from the network; do not use them.
|
||||
- Opengrep has no telemetry; there is no `--metrics` flag, and passing one fails the scan.
|
||||
- Prefer `--json --output <file>` or `--sarif --output <file>` for machine-readable downstream processing. JSON keeps `paths.scanned` and `results[].path`.
|
||||
- Keep the target path explicit; use an absolute or clearly scoped workspace path instead of `.` when possible.
|
||||
|
||||
Usage rules:
|
||||
- Start with the full `/opt/opengrep-rules` set unless the task clearly calls for a narrower language or framework directory.
|
||||
- Add `--taint-intrafile` for injection-class hunting (SQLi, command injection, SSRF, path traversal) in supported languages.
|
||||
- Use `--quiet` in automation to reduce noisy logs.
|
||||
- Use `--jobs` and `--timeout` explicitly for reproducible runtime behavior.
|
||||
- `PartialParsing` errors on templates are normal; they do not invalidate the run.
|
||||
|
||||
Failure recovery:
|
||||
- If scans are too slow, narrow the target path and the rule directories before changing engine settings.
|
||||
- If scans time out, increase `--timeout` modestly or lower `--jobs`.
|
||||
- If output is too broad, scope `--config`, add `--severity`, or exclude known irrelevant paths.
|
||||
|
||||
If uncertain, query web_search with:
|
||||
`site:github.com/opengrep/opengrep <flag>`
|
||||
@@ -0,0 +1,72 @@
|
||||
---
|
||||
name: semgrep
|
||||
description: Exact Semgrep CLI structure, metrics-off scanning, scoped ruleset selection, and automation-safe output patterns.
|
||||
---
|
||||
|
||||
# Semgrep CLI Playbook
|
||||
|
||||
Official docs:
|
||||
- https://semgrep.dev/docs/cli-reference
|
||||
- https://semgrep.dev/docs/getting-started/cli
|
||||
- https://semgrep.dev/docs/semgrep-code/semgrep-pro-engine-intro
|
||||
|
||||
Canonical syntax:
|
||||
`semgrep scan [flags]`
|
||||
|
||||
High-signal flags:
|
||||
- `--config <rule_or_ruleset>` ruleset, registry pack, local rule file, or directory
|
||||
- `--metrics=off` disable telemetry and metrics reporting
|
||||
- `--json` JSON output
|
||||
- `--sarif` SARIF output
|
||||
- `--output <file>` write findings to file
|
||||
- `--severity <level>` filter by severity
|
||||
- `--error` return non-zero exit when findings exist
|
||||
- `--quiet` suppress progress noise
|
||||
- `--jobs <n>` parallel workers
|
||||
- `--timeout <seconds>` per-file timeout
|
||||
- `--exclude <pattern>` exclude path pattern
|
||||
- `--include <pattern>` include path pattern
|
||||
- `--exclude-rule <rule_id>` suppress specific rule
|
||||
- `--baseline-commit <sha>` only report findings introduced after baseline
|
||||
- `--pro` enable Pro engine if available
|
||||
- `--oss-only` force OSS engine only
|
||||
|
||||
Agent-safe baseline for automation:
|
||||
`semgrep scan --config p/default --metrics=off --json --output semgrep.json --quiet --jobs 4 --timeout 20 /workspace`
|
||||
|
||||
Common patterns:
|
||||
- Default security scan:
|
||||
`semgrep scan --config p/default --metrics=off --json --output semgrep.json --quiet /workspace`
|
||||
- High-severity focused pass:
|
||||
`semgrep scan --config p/default --severity ERROR --metrics=off --json --output semgrep_high.json --quiet /workspace`
|
||||
- OWASP-oriented scan:
|
||||
`semgrep scan --config p/owasp-top-ten --metrics=off --sarif --output semgrep.sarif --quiet /workspace`
|
||||
- Language- or framework-specific rules:
|
||||
`semgrep scan --config p/python --config p/secrets --metrics=off --json --output semgrep_python.json --quiet /workspace`
|
||||
- Scoped directory scan:
|
||||
`semgrep scan --config p/default --metrics=off --json --output semgrep_api.json --quiet /workspace/services/api`
|
||||
- Pro engine check or run:
|
||||
`semgrep scan --config p/default --pro --metrics=off --json --output semgrep_pro.json --quiet /workspace`
|
||||
|
||||
Critical correctness rules:
|
||||
- Always include `--metrics=off`; Semgrep sends telemetry by default.
|
||||
- Always provide an explicit `--config`; do not rely on vague or implied defaults.
|
||||
- Prefer `--json --output <file>` or `--sarif --output <file>` for machine-readable downstream processing.
|
||||
- Keep the target path explicit; use an absolute or clearly scoped workspace path instead of `.` when possible.
|
||||
- If Pro availability matters, check it explicitly with a bounded command before assuming cross-file analysis exists.
|
||||
|
||||
Usage rules:
|
||||
- Start with `p/default` unless the task clearly calls for a narrower pack.
|
||||
- Add focused packs such as `p/secrets`, `p/python`, or `p/javascript` only when they match the target stack.
|
||||
- Use `--quiet` in automation to reduce noisy logs.
|
||||
- Use `--jobs` and `--timeout` explicitly for reproducible runtime behavior.
|
||||
- Do not use `-h`/`--help` for routine operation unless absolutely necessary.
|
||||
|
||||
Failure recovery:
|
||||
- If scans are too slow, narrow the target path and reduce the active rulesets before changing engine settings.
|
||||
- If scans time out, increase `--timeout` modestly or lower `--jobs`.
|
||||
- If output is too broad, scope `--config`, add `--severity`, or exclude known irrelevant paths.
|
||||
- If Pro mode fails, rerun with `--oss-only` or without `--pro` and note the loss of cross-file coverage.
|
||||
|
||||
If uncertain, query web_search with:
|
||||
`site:semgrep.dev semgrep <flag> cli`
|
||||
@@ -1,6 +1,6 @@
|
||||
---
|
||||
name: browser-security
|
||||
description: Browser-internals security testing for clickjacking, UI redressing, DoubleClickjacking, browsing-context relationships, postMessage, client-side path traversal, XS-Leaks, service workers, Web Workers, navigation behavior, CSP interactions, caches, and cross-origin state machines
|
||||
description: Browser-internals security testing for browsing-context relationships, postMessage, client-side path traversal, XS-Leaks, service workers, Web Workers, navigation behavior, CSP interactions, caches, and cross-origin state machines
|
||||
---
|
||||
|
||||
# Browser Security
|
||||
@@ -43,15 +43,6 @@ Draw the context graph. Security checks on `event.origin`, `event.source`, or a
|
||||
- Check whether a blocked intermediate frame leaves a useful browsing-context relationship intact.
|
||||
- Use random per-flow names or `_blank` with `noopener` where an opener relationship is unnecessary.
|
||||
|
||||
### Clickjacking and UI Redressing
|
||||
|
||||
- **UI-redressing closure gate:** Never mark clickjacking as ruled out from `X-Frame-Options` or CSP `frame-ancestors` alone. Before closure, test both frame embedding and cross-window/popup retargeting.
|
||||
- Do not treat a different origin or port as protection against UI redressing. Same-origin policy blocks DOM access, not top-level navigation, `window.open()`, opener relationships, focus changes, or window closing. Test only explicitly authorized targets; if a required origin is outside scope, keep the path `needs_follow_up` and do not claim the vulnerability class is cleared.
|
||||
- For framed cases, prove the overlay and sensitive click. For DoubleClickjacking, use two real browser clicks and trace click 1 → window/focus/navigation/positioning change → click 2 → sensitive control. Never substitute scripted `.click()` calls.
|
||||
- Reproduce in the victim state required by the workflow—authenticated when necessary—with realistic user activation and stable geometry; capture window bounds and timing.
|
||||
- **Evidence-limited closure:** Verify a repeatable sensitive server-side state change with a recording or deterministic event trace and a negative control. If the exact sequence cannot be tested, record `needs_follow_up`, not `ruled_out` or `no_issue_found`. Popup-as-tab behavior, ignored geometry, unavailable OS-level input, or a manual control click are not counterevidence. Close the path only when a named mitigation blocks it, or when geometry is honored and both physical clicks are delivered without a sensitive action.
|
||||
- **Impact floor:** Successful execution of a protected consent, approval, or grant is an integrity impact even when idempotent, pre-seeded, or followed by no immediate token or data access. Use downstream capability to calibrate severity, not erase the demonstrated protected action.
|
||||
|
||||
### Client-Side Path Traversal
|
||||
|
||||
Trace the complete source-to-request pipeline:
|
||||
@@ -155,7 +146,7 @@ Also inspect DevTools network initiators, service workers, storage, CSP violatio
|
||||
|
||||
### Source Review
|
||||
|
||||
- Search for `postMessage`, message listeners, `window.open`, named targets, opener/parent access, frame creation, `dblclick`, `mousedown`/`mouseup`, focus/blur, move/resize, navigation handlers, and sandbox attributes.
|
||||
- Search for `postMessage`, message listeners, `window.open`, named targets, opener/parent access, frame creation, and sandbox attributes.
|
||||
- Search for router parameter APIs flowing into `fetch`, Axios, navigation, or HTML rendering.
|
||||
- Search for service-worker registration, Cache API writes, worker constructors, Blob URLs, and dynamic imports.
|
||||
- Search for raw HTML sinks and trust escape hatches in every supported frontend framework.
|
||||
@@ -170,7 +161,7 @@ Also inspect DevTools network initiators, service workers, storage, CSP violatio
|
||||
5. **Prove the primitive** - Data transfer, path change, state oracle, cache modification, or context capture.
|
||||
6. **Escalate deliberately** - Chain to a privileged action, sensitive disclosure, SSRF, or executable DOM sink.
|
||||
7. **Cross-browser check** - At minimum record Chromium/Firefox/Safari applicability when the primitive is browser-specific.
|
||||
8. **State interaction requirements** - Record each physical click or drag, popup permission state, active context, focus, geometry, navigation, timing window, login state, and visual deception.
|
||||
8. **State interaction requirements** - Click, drag, popup permission, timing window, login state, and visual deception.
|
||||
|
||||
## Validation
|
||||
|
||||
@@ -180,7 +171,7 @@ Also inspect DevTools network initiators, service workers, storage, CSP violatio
|
||||
4. For XS-Leaks, repeat randomized success/failure trials and quantify separation and noise.
|
||||
5. For workers/caches, show which later context consumes the modified data.
|
||||
6. For client-side traversal, capture the final network request and the security-relevant response/action.
|
||||
7. For interaction-dependent chains, show both click targets, intermediate context changes, and the resulting sensitive state change in a recording or deterministic event trace.
|
||||
7. For interaction-dependent chains, provide a screen recording or deterministic event trace.
|
||||
|
||||
## False Positives
|
||||
|
||||
|
||||
@@ -16,6 +16,8 @@ from typing import TYPE_CHECKING, Any
|
||||
|
||||
from agents import RunContextWrapper, function_tool
|
||||
|
||||
from strix.report.atlas import normalize_atlas_techniques
|
||||
from strix.report.attack import normalize_attack_techniques
|
||||
from strix.tools.nullish import clean_optional
|
||||
from strix.tools.proxy.tools import existing_request_ids
|
||||
|
||||
@@ -462,6 +464,20 @@ def _collect_update_changes( # noqa: PLR0912, PLR0915
|
||||
if cwe:
|
||||
changes["cwe"] = cwe
|
||||
|
||||
if fields.get("attack_techniques") is not None:
|
||||
attack_techniques, attack_errors = normalize_attack_techniques(
|
||||
fields.get("attack_techniques")
|
||||
)
|
||||
errors.extend(attack_errors)
|
||||
if not attack_errors:
|
||||
changes["attack_techniques"] = attack_techniques
|
||||
|
||||
if fields.get("atlas_techniques") is not None:
|
||||
atlas_techniques, atlas_errors = normalize_atlas_techniques(fields.get("atlas_techniques"))
|
||||
errors.extend(atlas_errors)
|
||||
if not atlas_errors:
|
||||
changes["atlas_techniques"] = atlas_techniques
|
||||
|
||||
raw_http_exchange_ids = fields.get("http_exchange_ids")
|
||||
http_exchange_ids, http_exchange_errors = _normalize_http_exchange_ids(raw_http_exchange_ids)
|
||||
errors.extend(http_exchange_errors)
|
||||
@@ -803,6 +819,8 @@ async def _do_create(
|
||||
confidence_rationale: str | None = None,
|
||||
fix_verification: str | None = None,
|
||||
fix_pr_body: str | None = None,
|
||||
attack_techniques: list[str] | None = None,
|
||||
atlas_techniques: list[str] | None = None,
|
||||
agent_id: str | None = None,
|
||||
agent_name: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
@@ -845,6 +863,10 @@ async def _do_create(
|
||||
errors.extend(_validate_fix_verification(parsed_locations, fix_verification))
|
||||
cve, cwe, identifier_errors = _validate_identifiers(cve, cwe)
|
||||
errors.extend(identifier_errors)
|
||||
normalized_attack_techniques, attack_errors = normalize_attack_techniques(attack_techniques)
|
||||
errors.extend(attack_errors)
|
||||
normalized_atlas_techniques, atlas_errors = normalize_atlas_techniques(atlas_techniques)
|
||||
errors.extend(atlas_errors)
|
||||
normalized_http_exchange_ids, http_exchange_errors = _normalize_http_exchange_ids(
|
||||
http_exchange_ids
|
||||
)
|
||||
@@ -907,6 +929,8 @@ async def _do_create(
|
||||
"method": method,
|
||||
"cve": cve,
|
||||
"cwe": cwe,
|
||||
"attack_techniques": normalized_attack_techniques or None,
|
||||
"atlas_techniques": normalized_atlas_techniques or None,
|
||||
"code_locations": parsed_locations,
|
||||
"fix_verification": fix_verification,
|
||||
"fix_pr_body": fix_pr_body,
|
||||
@@ -1000,6 +1024,8 @@ async def create_vulnerability_report(
|
||||
method: str | None = None,
|
||||
cve: str | None = None,
|
||||
cwe: str | None = None,
|
||||
attack_techniques: list[str] | None = None,
|
||||
atlas_techniques: list[str] | None = None,
|
||||
code_locations: list[dict[str, Any]] | None = None,
|
||||
http_exchange_ids: list[str] | None = None,
|
||||
confidence_rationale: str | None = None,
|
||||
@@ -1296,6 +1322,18 @@ async def create_vulnerability_report(
|
||||
method: HTTP method when relevant.
|
||||
cve: ``CVE-YYYY-NNNNN`` if certain, else omit.
|
||||
cwe: ``CWE-NNN`` (most specific child) if certain, else omit.
|
||||
attack_techniques: MITRE ATT&CK Enterprise technique ids (e.g.
|
||||
``["T1190", "T1552.001"]``) for what the PoC actually did:
|
||||
the entry point it exploited and each step it demonstrated
|
||||
after that (credentials it read, commands it ran, data it
|
||||
took). Tag only techniques the evidence shows; leave out
|
||||
techniques the finding merely makes possible. Omit when unsure.
|
||||
atlas_techniques: MITRE ATLAS technique ids (e.g. ``["AML.T0051"]``)
|
||||
when the target is an AI system (LLM app, agent, RAG pipeline)
|
||||
and the PoC showed the attack, for example prompt injection
|
||||
(AML.T0051), jailbreak (AML.T0054), system prompt extraction
|
||||
(AML.T0056) or data leakage (AML.T0057). Same rule: only what the
|
||||
evidence shows. Omit for other findings.
|
||||
code_locations: White-box findings — list of location objects.
|
||||
http_exchange_ids: Proxy request IDs that prove this finding.
|
||||
Copy these IDs from ``list_requests`` or ``view_request``.
|
||||
@@ -1450,6 +1488,8 @@ async def create_vulnerability_report(
|
||||
method=method,
|
||||
cve=cve,
|
||||
cwe=cwe,
|
||||
attack_techniques=attack_techniques,
|
||||
atlas_techniques=atlas_techniques,
|
||||
code_locations=code_locations,
|
||||
http_exchange_ids=http_exchange_ids,
|
||||
fix_verification=fix_verification,
|
||||
@@ -1485,6 +1525,8 @@ async def update_vulnerability_report(
|
||||
method: str | None = None,
|
||||
cve: str | None = None,
|
||||
cwe: str | None = None,
|
||||
attack_techniques: list[str] | None = None,
|
||||
atlas_techniques: list[str] | None = None,
|
||||
code_locations: list[dict[str, Any]] | None = None,
|
||||
http_exchange_ids: list[str] | None = None,
|
||||
fix_verification: str | None = None,
|
||||
@@ -1558,6 +1600,9 @@ async def update_vulnerability_report(
|
||||
method: Replacement HTTP method.
|
||||
cve: Replacement CVE id.
|
||||
cwe: Replacement CWE id.
|
||||
attack_techniques: Replacement MITRE ATT&CK technique ids. Same
|
||||
rule as on create: only what the evidence demonstrates.
|
||||
atlas_techniques: Replacement MITRE ATLAS technique ids. Same rule.
|
||||
code_locations: Replacement code locations.
|
||||
http_exchange_ids: Replacement proxy request ids. Pass an empty
|
||||
list to remove all linked exchanges.
|
||||
@@ -1604,6 +1649,8 @@ async def update_vulnerability_report(
|
||||
"method": method,
|
||||
"cve": cve,
|
||||
"cwe": cwe,
|
||||
"attack_techniques": attack_techniques,
|
||||
"atlas_techniques": atlas_techniques,
|
||||
"code_locations": code_locations,
|
||||
"http_exchange_ids": http_exchange_ids,
|
||||
"fix_verification": fix_verification,
|
||||
@@ -1905,6 +1952,7 @@ async def _do_create_dependency( # noqa: PLR0912, PLR0915
|
||||
fix_verification: str | None = None,
|
||||
contextual_cvss_breakdown: dict[str, str] | None = None,
|
||||
contextual_cvss_reasoning: str | None = None,
|
||||
attack_techniques: list[str] | None = None,
|
||||
agent_id: str | None = None,
|
||||
agent_name: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
@@ -1935,6 +1983,9 @@ async def _do_create_dependency( # noqa: PLR0912, PLR0915
|
||||
if cwe_err:
|
||||
errors.append(cwe_err)
|
||||
|
||||
normalized_attack_techniques, attack_errors = normalize_attack_techniques(attack_techniques)
|
||||
errors.extend(attack_errors)
|
||||
|
||||
fix_effort = (fix_effort or "").strip().lower()
|
||||
if fix_effort not in _VALID_FIX_EFFORT:
|
||||
errors.append(
|
||||
@@ -2061,6 +2112,7 @@ async def _do_create_dependency( # noqa: PLR0912, PLR0915
|
||||
cvss=cvss_score if advisory_cvss is not None else None,
|
||||
cve=parsed_cve,
|
||||
cwe=cwe,
|
||||
attack_techniques=normalized_attack_techniques or None,
|
||||
code_locations=parsed_locations,
|
||||
fix_verification=fix_verification,
|
||||
finding_class="dependency_cve",
|
||||
@@ -2121,6 +2173,7 @@ async def create_dependency_report(
|
||||
fix_verification: str | None = None,
|
||||
contextual_cvss_breakdown: dict[str, str] | None = None,
|
||||
contextual_cvss_reasoning: str | None = None,
|
||||
attack_techniques: list[str] | None = None,
|
||||
) -> str:
|
||||
"""File a known-CVE dependency (SCA) finding — one report per CVE x package.
|
||||
|
||||
@@ -2185,6 +2238,10 @@ async def create_dependency_report(
|
||||
package_ecosystem: e.g. ``npm`` / ``pypi`` / ``maven`` / ``go``.
|
||||
fixed_version: First non-vulnerable version, if known.
|
||||
cwe: ``CWE-NNN`` (most specific) if certain, else omit.
|
||||
attack_techniques: MITRE ATT&CK Enterprise technique ids the
|
||||
reachability evidence demonstrates (e.g. ``["T1190"]`` for a
|
||||
proven exploitable path from a public endpoint). Omit when the
|
||||
evidence does not show one.
|
||||
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
|
||||
@@ -2310,6 +2367,7 @@ async def create_dependency_report(
|
||||
fix_verification=fix_verification,
|
||||
contextual_cvss_breakdown=contextual_cvss_breakdown,
|
||||
contextual_cvss_reasoning=contextual_cvss_reasoning,
|
||||
attack_techniques=attack_techniques,
|
||||
agent_id=agent_id,
|
||||
agent_name=agent_name,
|
||||
)
|
||||
@@ -2338,6 +2396,7 @@ _REPORT_SUMMARY_FIELDS = (
|
||||
"finding_class",
|
||||
"cve",
|
||||
"cwe",
|
||||
"attack_techniques",
|
||||
"target",
|
||||
"endpoint",
|
||||
"method",
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
from strix.report.atlas import normalize_atlas_techniques, technique_url
|
||||
|
||||
|
||||
def test_normalize_atlas_techniques_validates_ids() -> None:
|
||||
ids, errors = normalize_atlas_techniques(["aml.t0051", "AML.T0051", "AML.T0054 jailbreak"])
|
||||
assert ids == ["AML.T0051", "AML.T0054"]
|
||||
assert errors == []
|
||||
|
||||
|
||||
def test_normalize_atlas_techniques_rejects_unknown_and_malformed() -> None:
|
||||
ids, errors = normalize_atlas_techniques(["AML.T9999", "T1190"])
|
||||
assert ids == []
|
||||
assert len(errors) == 2
|
||||
assert "unknown ATLAS technique" in errors[0]
|
||||
assert "invalid ATLAS technique" in errors[1]
|
||||
|
||||
|
||||
def test_technique_url() -> None:
|
||||
assert technique_url("AML.T0051.000") == "https://atlas.mitre.org/techniques/AML.T0051.000"
|
||||
@@ -1,159 +0,0 @@
|
||||
"""Only validated model decisions may suppress a finding."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from agents.items import ModelResponse
|
||||
from agents.usage import Usage
|
||||
from openai.types.responses import ResponseOutputMessage, ResponseOutputText
|
||||
|
||||
from strix.config import loader
|
||||
from strix.report import dedupe
|
||||
from strix.report.state import ReportState, get_global_report_state, set_global_report_state
|
||||
from strix.tools.reporting.tool import _do_create
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[ReportState]:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
previous = get_global_report_state()
|
||||
state = ReportState(run_name="dedupe-decisions")
|
||||
set_global_report_state(state)
|
||||
state.add_vulnerability_report(title="Existing finding", severity="low")
|
||||
try:
|
||||
yield state
|
||||
finally:
|
||||
set_global_report_state(previous)
|
||||
|
||||
|
||||
def _mock_judge(monkeypatch: pytest.MonkeyPatch, decision: dict[str, Any]) -> AsyncMock:
|
||||
settings = loader.load_settings().model_copy(deep=True)
|
||||
settings.llm.model = "openai/synthetic-model"
|
||||
model = AsyncMock()
|
||||
model.get_response.return_value = ModelResponse(
|
||||
output=[
|
||||
ResponseOutputMessage(
|
||||
id="msg_synthetic",
|
||||
type="message",
|
||||
role="assistant",
|
||||
status="completed",
|
||||
content=[
|
||||
ResponseOutputText(
|
||||
type="output_text",
|
||||
annotations=[],
|
||||
text=json.dumps({"confidence": 0.95, "reason": "Synthetic", **decision}),
|
||||
)
|
||||
],
|
||||
)
|
||||
],
|
||||
usage=Usage(),
|
||||
response_id="synthetic",
|
||||
)
|
||||
monkeypatch.setattr(dedupe, "load_settings", lambda: settings)
|
||||
monkeypatch.setattr(dedupe, "configure_sdk_model_defaults", lambda _: None)
|
||||
monkeypatch.setattr(dedupe, "resolve_dedupe_model", lambda *_: model)
|
||||
return model
|
||||
|
||||
|
||||
async def _file_finding() -> dict[str, Any]:
|
||||
return await _do_create(
|
||||
title="New finding",
|
||||
description="The endpoint reflects unencoded input.",
|
||||
impact="The reflected script executes in a victim's browser.",
|
||||
target="https://example.invalid",
|
||||
technical_analysis="Input is interpolated into HTML without encoding.",
|
||||
poc_description="Open the endpoint with the reflected parameter.",
|
||||
poc_script_code="print('synthetic reproduction')",
|
||||
remediation_steps="Encode output for its HTML context.",
|
||||
evidence="The synthetic response contains the unencoded parameter.",
|
||||
assumptions="A victim follows the crafted link.",
|
||||
counterevidence="No encoding or content security policy is present.",
|
||||
confidence="high",
|
||||
severity_change_conditions="A strict content security policy would lower the impact.",
|
||||
fix_effort="low",
|
||||
cvss_breakdown={
|
||||
"attack_vector": "N",
|
||||
"attack_complexity": "L",
|
||||
"privileges_required": "N",
|
||||
"user_interaction": "R",
|
||||
"scope": "U",
|
||||
"confidentiality": "L",
|
||||
"integrity": "L",
|
||||
"availability": "N",
|
||||
},
|
||||
endpoint="/search",
|
||||
method="GET",
|
||||
cve=None,
|
||||
cwe="CWE-79",
|
||||
code_locations=None,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["false", "true", 0, 1, None, [], {}])
|
||||
def test_dedupe_parser_requires_a_boolean(value: Any) -> None:
|
||||
with pytest.raises(TypeError, match="is_duplicate"):
|
||||
dedupe._parse_dedupe_response(json.dumps({"is_duplicate": value}))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"decision",
|
||||
[
|
||||
{"is_duplicate": "false", "duplicate_id": "vuln-0001"},
|
||||
{"is_duplicate": "true", "duplicate_id": "vuln-0001"},
|
||||
{"is_duplicate": 1, "duplicate_id": "vuln-0001"},
|
||||
{"is_duplicate": True, "duplicate_id": "vuln-9999"},
|
||||
{"is_duplicate": True, "duplicate_id": ""},
|
||||
{"is_duplicate": True, "duplicate_id": None},
|
||||
{"is_duplicate": True, "duplicate_id": ["vuln-0001"]},
|
||||
{"is_duplicate": True},
|
||||
],
|
||||
)
|
||||
async def test_invalid_dedupe_decision_does_not_suppress_finding(
|
||||
report_state: ReportState, monkeypatch: pytest.MonkeyPatch, decision: dict[str, Any]
|
||||
) -> None:
|
||||
model = _mock_judge(monkeypatch, decision)
|
||||
|
||||
result = await _file_finding()
|
||||
|
||||
assert result["success"] is True
|
||||
model.get_response.assert_awaited_once()
|
||||
saved = json.loads((report_state.get_run_dir() / "vulnerabilities.json").read_text())
|
||||
assert [report["title"] for report in saved] == ["Existing finding", "New finding"]
|
||||
assert (report_state.get_run_dir() / "vulnerabilities" / "vuln-0002.md").is_file()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("is_duplicate", [False, True])
|
||||
async def test_valid_dedupe_decisions_keep_their_meaning(
|
||||
report_state: ReportState, monkeypatch: pytest.MonkeyPatch, is_duplicate: bool
|
||||
) -> None:
|
||||
# An extraneous ID must not turn a valid false decision into a duplicate.
|
||||
_mock_judge(monkeypatch, {"is_duplicate": is_duplicate, "duplicate_id": "vuln-0001"})
|
||||
|
||||
result = await _file_finding()
|
||||
|
||||
assert result["success"] is not is_duplicate
|
||||
saved = json.loads((report_state.get_run_dir() / "vulnerabilities.json").read_text())
|
||||
assert len(saved) == (1 if is_duplicate else 2)
|
||||
if is_duplicate:
|
||||
assert result["duplicate_of"] == "vuln-0001"
|
||||
|
||||
|
||||
async def test_dedupe_id_is_matched_without_truncation(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
existing_id = "x" * 64
|
||||
_mock_judge(monkeypatch, {"is_duplicate": True, "duplicate_id": existing_id + "-unknown"})
|
||||
|
||||
result = await dedupe.check_duplicate(
|
||||
{"title": "New finding"}, [{"id": existing_id, "title": "Existing finding"}]
|
||||
)
|
||||
|
||||
assert result["is_duplicate"] is False
|
||||
assert "error" in result
|
||||
+1
-41
@@ -2,14 +2,11 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from agents.usage import Usage
|
||||
|
||||
from strix.config import loader
|
||||
from strix.core.agents import AgentCoordinator
|
||||
from strix.core.hooks import (
|
||||
BudgetExceededError,
|
||||
@@ -18,11 +15,6 @@ from strix.core.hooks import (
|
||||
SubagentBudgetReservedError,
|
||||
recomputed_budget_flags,
|
||||
)
|
||||
from strix.report.usage import LLMUsageLedger
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _make_hooks(max_budget: float | None) -> ReportUsageHooks:
|
||||
@@ -93,38 +85,6 @@ async def test_over_budget_raises() -> None:
|
||||
await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_manual_rates_make_unpriced_model_trip_budget(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM", "test-model")
|
||||
monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "10")
|
||||
monkeypatch.delenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", raising=False)
|
||||
monkeypatch.delenv("STRIX_LLM_CACHED_INPUT_COST_PER_MTOK", raising=False)
|
||||
monkeypatch.setattr(loader, "_override", tmp_path / "missing.json")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
usage = Usage()
|
||||
usage.requests = 1
|
||||
usage.input_tokens = 1000
|
||||
usage.total_tokens = 1000
|
||||
ledger = LLMUsageLedger()
|
||||
|
||||
class ReportState:
|
||||
def record_sdk_usage(self, **kwargs: Any) -> None:
|
||||
ledger.record(**kwargs)
|
||||
|
||||
def get_total_llm_cost(self) -> float:
|
||||
return ledger.total_cost
|
||||
|
||||
hooks = _make_hooks(0.005)
|
||||
response: Any = SimpleNamespace(usage=usage)
|
||||
with (
|
||||
patch("strix.core.hooks.get_global_report_state", return_value=ReportState()),
|
||||
pytest.raises(BudgetExceededError),
|
||||
):
|
||||
await hooks.on_llm_end(_make_context(), MagicMock(), response)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_budget_check_uses_live_cost_accessor() -> None:
|
||||
# The check must read the live ledger, not the persisted run-record snapshot,
|
||||
|
||||
@@ -1,50 +1,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
import litellm
|
||||
import pytest
|
||||
from agents.usage import Usage
|
||||
from pydantic import ValidationError
|
||||
|
||||
from strix.config import loader
|
||||
from strix.config.settings import LlmSettings
|
||||
from strix.report.pricing import resolve_litellm_model
|
||||
from strix.report.usage import LLMUsageLedger
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
PRICE_ENV = (
|
||||
"STRIX_LLM_INPUT_COST_PER_MTOK",
|
||||
"STRIX_LLM_OUTPUT_COST_PER_MTOK",
|
||||
"STRIX_LLM_CACHED_INPUT_COST_PER_MTOK",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolated_settings(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
for name in PRICE_ENV:
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
monkeypatch.setenv("STRIX_LLM", "gateway/unpriced-model")
|
||||
monkeypatch.setattr(loader, "_override", tmp_path / "missing.json")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
|
||||
|
||||
def _usage(input_tokens: int, output_tokens: int = 0, cached_tokens: int = 0) -> Usage:
|
||||
usage = Usage()
|
||||
usage.requests = 1
|
||||
usage.input_tokens = input_tokens
|
||||
usage.output_tokens = output_tokens
|
||||
usage.total_tokens = input_tokens + output_tokens
|
||||
usage.input_tokens_details.cached_tokens = cached_tokens
|
||||
return usage
|
||||
|
||||
|
||||
def test_resolves_common_bare_model_names() -> None:
|
||||
resolve_litellm_model.cache_clear()
|
||||
assert resolve_litellm_model("deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
|
||||
@@ -96,112 +60,6 @@ def test_ledger_prefers_observed_cost_over_estimate() -> None:
|
||||
assert ledger.total_cost == 0.17
|
||||
|
||||
|
||||
def test_unpriced_model_without_manual_rates_costs_zero() -> None:
|
||||
ledger = LLMUsageLedger()
|
||||
|
||||
ledger.record(agent_id="a", usage=_usage(1000, 200), model="gateway/unpriced-model")
|
||||
|
||||
assert ledger.total_cost == 0.0
|
||||
|
||||
|
||||
def test_manual_rates_price_usage_and_override_observed_cost(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "2")
|
||||
monkeypatch.setenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", "8")
|
||||
ledger = LLMUsageLedger()
|
||||
|
||||
ledger.record(agent_id="a", usage=_usage(1000, 200), model="gateway/unpriced-model")
|
||||
ledger.record_observed_cost(0.17)
|
||||
|
||||
assert ledger.total_cost == 0.0036
|
||||
assert ledger.to_record()["agents"][0]["cost"] == 0.0036
|
||||
|
||||
|
||||
def test_manual_rates_from_config_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
path = tmp_path / "config.json"
|
||||
path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"env": {
|
||||
"STRIX_LLM_INPUT_COST_PER_MTOK": "2",
|
||||
"STRIX_LLM_OUTPUT_COST_PER_MTOK": "8",
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setattr(loader, "_override", path)
|
||||
ledger = LLMUsageLedger()
|
||||
|
||||
ledger.record(agent_id="a", usage=_usage(1000, 200), model="gateway/unpriced-model")
|
||||
|
||||
assert ledger.total_cost == 0.0036
|
||||
|
||||
|
||||
def test_manual_rates_only_price_the_configured_model(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "2")
|
||||
monkeypatch.setenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", "8")
|
||||
ledger = LLMUsageLedger()
|
||||
|
||||
ledger.record(agent_id="a", usage=_usage(1000, 200), model="gateway/unpriced-model")
|
||||
with patch("litellm.completion_cost", return_value=0.42):
|
||||
ledger.record(agent_id="dedupe", usage=_usage(500, 50), model="deepseek/deepseek-v4-flash")
|
||||
record = ledger.to_record()
|
||||
|
||||
assert ledger.total_cost == 0.4236
|
||||
assert {agent["agent_id"]: agent["cost"] for agent in record["agents"]} == {
|
||||
"a": 0.0036,
|
||||
"dedupe": 0.42,
|
||||
}
|
||||
|
||||
|
||||
def test_cached_tokens_fall_back_to_input_rate(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "5")
|
||||
monkeypatch.setenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", "7")
|
||||
ledger = LLMUsageLedger()
|
||||
|
||||
ledger.record(
|
||||
agent_id="a", usage=_usage(10, 1, cached_tokens=2), model="gateway/unpriced-model"
|
||||
)
|
||||
|
||||
assert ledger.total_cost == 0.000057
|
||||
|
||||
|
||||
def test_hydrated_usage_uses_configured_rates_not_persisted_cost(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "2")
|
||||
monkeypatch.setenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", "8")
|
||||
original = LLMUsageLedger()
|
||||
original.record(agent_id="a", usage=_usage(1000, 200), model="gateway/unpriced-model")
|
||||
persisted = original.to_record()
|
||||
persisted["cost"] = 42.0
|
||||
persisted["agents"][0]["cost"] = 42.0
|
||||
hydrated = LLMUsageLedger()
|
||||
|
||||
hydrated.hydrate(persisted)
|
||||
|
||||
assert hydrated.total_cost == 0.0036
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["invalid", "-1", "nan", "inf"])
|
||||
def test_invalid_manual_rate_is_rejected(value: str, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", value)
|
||||
with pytest.raises(ValidationError):
|
||||
LlmSettings()
|
||||
|
||||
|
||||
def test_zero_cost_overrides_manual_rates(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "10")
|
||||
ledger = LLMUsageLedger()
|
||||
ledger.zero_cost = True
|
||||
|
||||
ledger.record(agent_id="a", usage=_usage(1000), model="gateway/unpriced-model")
|
||||
|
||||
assert ledger.total_cost == 0.0
|
||||
|
||||
|
||||
def test_hydrated_estimate_continues_accumulating_new_estimates() -> None:
|
||||
usage = Usage()
|
||||
usage.requests = 1
|
||||
|
||||
@@ -2408,3 +2408,57 @@ async def test_delete_tool_carries_the_caller_identity(report_state: ReportState
|
||||
assert history[0]["agent_id"] == "aaaa1111"
|
||||
assert history[0]["agent_name"] == "Recon Agent"
|
||||
assert "update_vulnerability_report" in delete_vulnerability_report.description
|
||||
|
||||
|
||||
async def test_create_records_attack_techniques(report_state: ReportState) -> None:
|
||||
result = await _do_create(
|
||||
**_CONFIRMED_KWARGS,
|
||||
attack_techniques=["t1190", "T1552.001 - Credentials In Files", "T1190"],
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert report_state.vulnerability_reports[0]["attack_techniques"] == ["T1190", "T1552.001"]
|
||||
|
||||
|
||||
async def test_create_rejects_unknown_attack_technique(report_state: ReportState) -> None:
|
||||
result = await _do_create(**_CONFIRMED_KWARGS, attack_techniques=["T9999", "exploit"])
|
||||
|
||||
assert result["success"] is False
|
||||
assert any("unknown ATT&CK technique: 'T9999'" in error for error in result["errors"])
|
||||
assert any("invalid ATT&CK technique: 'exploit'" in error for error in result["errors"])
|
||||
assert report_state.vulnerability_reports == []
|
||||
|
||||
|
||||
async def test_create_without_attack_techniques_omits_the_field(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
result = await _do_create(**_CONFIRMED_KWARGS)
|
||||
|
||||
assert result["success"] is True
|
||||
assert "attack_techniques" not in report_state.vulnerability_reports[0]
|
||||
|
||||
|
||||
def test_update_replaces_attack_techniques(report_state: ReportState) -> None:
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="The chained PoC read cloud credentials.",
|
||||
fields={"attack_techniques": ["T1190", "T1552.005"]},
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert report_state.vulnerability_reports[0]["attack_techniques"] == ["T1190", "T1552.005"]
|
||||
|
||||
|
||||
def test_update_rejects_revoked_attack_technique(report_state: ReportState) -> None:
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="Tag the technique.",
|
||||
fields={"attack_techniques": ["T1086"]},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "attack_techniques" not in report_state.vulnerability_reports[0]
|
||||
@@ -371,3 +371,10 @@ def test_calibration_metadata_survives_into_result_properties(tmp_path: Path) ->
|
||||
assert strix["confidence_rationale"] == "Reproduced once out of three attempts."
|
||||
assert strix["severity_change_conditions"] == "Critical if the WAF rule is removed."
|
||||
assert strix["fix_verification"] == "Not retested."
|
||||
|
||||
|
||||
def test_write_sarif_carries_attack_techniques(tmp_path: Path) -> None:
|
||||
write_sarif(tmp_path, [_finding(attack_techniques=["T1190", "T1059"])])
|
||||
result = _read(tmp_path)["runs"][0]["results"][0]
|
||||
|
||||
assert result["properties"]["strix"]["attack_techniques"] == ["T1190", "T1059"]
|
||||
Reference in new issue
Block a user