mirror of
https://github.com/usestrix/strix.git
synced 2026-11-09 18:08:59 -10:00
Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e05948080b | ||
|
|
9ebeffbc74 | ||
|
|
4b19046320 | ||
|
|
675608dc7a | ||
|
|
18fa4c8449 | ||
|
|
215482eb22 | ||
|
|
0601550662 | ||
|
|
851743adcf | ||
|
|
09536cf148 | ||
|
|
cb581bf93d | ||
|
|
6b7ae2ebe4 | ||
|
|
92b30bcad5 | ||
|
|
dfdbe4b2ab |
No files matched your search
@@ -1,48 +1,39 @@
|
||||
name: Package
|
||||
name: Binary
|
||||
|
||||
on:
|
||||
workflow_call:
|
||||
pull_request:
|
||||
paths:
|
||||
- pyproject.toml
|
||||
- uv.lock
|
||||
- strix.spec
|
||||
- scripts/**
|
||||
- strix/interface/tui/**
|
||||
- .github/actions/**
|
||||
- .github/workflows/binary.yml
|
||||
push:
|
||||
branches:
|
||||
- main
|
||||
paths:
|
||||
- pyproject.toml
|
||||
- uv.lock
|
||||
- strix.spec
|
||||
- scripts/**
|
||||
- strix/interface/tui/**
|
||||
- .github/actions/**
|
||||
- .github/workflows/binary.yml
|
||||
workflow_dispatch:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
wheel:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: actions/setup-go@44694675825211faa026b3c33043df3e48a5fa00 # v6.0.0
|
||||
with:
|
||||
go-version-file: strix/interface/tui/go.mod
|
||||
cache-dependency-path: strix/interface/tui/go.sum
|
||||
- uses: ./.github/actions/setup-python-env
|
||||
- run: uv build --wheel
|
||||
- name: Wheel ships the TUI sidecar and the viewer bundle
|
||||
run: |
|
||||
uv run --frozen python - <<'PY'
|
||||
import glob, zipfile
|
||||
(wheel,) = glob.glob("dist/*.whl")
|
||||
names = zipfile.ZipFile(wheel).namelist()
|
||||
assert "strix/bin/strix-tui" in names, names
|
||||
assert "strix/interface/viewer/static/index.html" in names, names
|
||||
assert not any(name.startswith("strix/interface/viewer/frontend/") for name in names)
|
||||
assert not any(name.startswith("strix/interface/tui/cmd/") for name in names)
|
||||
PY
|
||||
- name: Install into a clean environment and run the CLI
|
||||
env:
|
||||
UV_EXCLUDE_NEWER: 5 days
|
||||
run: |
|
||||
uv venv --python 3.12 /tmp/strix-smoke
|
||||
uv pip install --python /tmp/strix-smoke/bin/python dist/*.whl
|
||||
/tmp/strix-smoke/bin/strix --version
|
||||
/tmp/strix-smoke/bin/strix --help >/dev/null
|
||||
- name: Vertex AI and Bedrock dependencies install with the wheel
|
||||
run: /tmp/strix-smoke/bin/python -c "import google.auth.transport.requests, boto3"
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.ref }}
|
||||
cancel-in-progress: true
|
||||
|
||||
jobs:
|
||||
binary:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 20
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
@@ -32,6 +32,7 @@ jobs:
|
||||
wheel-platform: win_amd64
|
||||
|
||||
runs-on: ${{ matrix.os }}
|
||||
timeout-minutes: 45
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
@@ -110,6 +111,7 @@ jobs:
|
||||
if: startsWith(github.ref, 'refs/tags/')
|
||||
needs: build
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 10
|
||||
permissions:
|
||||
contents: write
|
||||
|
||||
|
||||
@@ -1,61 +0,0 @@
|
||||
name: Python
|
||||
|
||||
on:
|
||||
workflow_call:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
lock:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: astral-sh/setup-uv@d4b2f3b6ecc6e67c4457f6d3e41ec42d3d0fcb86 # v5.4.2
|
||||
- run: uv lock --check
|
||||
|
||||
ruff:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: ./.github/actions/setup-python-env
|
||||
- run: uv run --frozen ruff check .
|
||||
- run: uv run --frozen ruff format --check .
|
||||
|
||||
mypy:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: ./.github/actions/setup-python-env
|
||||
- run: uv run --frozen mypy strix/
|
||||
|
||||
bandit:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: ./.github/actions/setup-python-env
|
||||
- run: uv run --frozen bandit -r strix/ -c pyproject.toml
|
||||
|
||||
pytest:
|
||||
name: pytest (${{ matrix.python-version }})
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
python-version: ['3.12', '3.13', '3.14']
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: ./.github/actions/setup-python-env
|
||||
with:
|
||||
python-version: ${{ matrix.python-version }}
|
||||
- run: uv run --frozen pytest -q
|
||||
@@ -1,26 +0,0 @@
|
||||
name: TUI
|
||||
|
||||
on:
|
||||
workflow_call:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
go:
|
||||
runs-on: ubuntu-latest
|
||||
defaults:
|
||||
run:
|
||||
working-directory: strix/interface/tui
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: actions/setup-go@44694675825211faa026b3c33043df3e48a5fa00 # v6.0.0
|
||||
with:
|
||||
go-version-file: strix/interface/tui/go.mod
|
||||
cache-dependency-path: strix/interface/tui/go.sum
|
||||
- run: test -z "$(gofmt -l .)"
|
||||
- run: go vet ./...
|
||||
- run: CGO_ENABLED=0 go build -trimpath -o /dev/null ./cmd/strix-tui
|
||||
- run: go test -race ./...
|
||||
@@ -1,27 +0,0 @@
|
||||
name: Viewer
|
||||
|
||||
on:
|
||||
workflow_call:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
defaults:
|
||||
run:
|
||||
working-directory: strix/interface/viewer/frontend
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
|
||||
with:
|
||||
node-version: '24'
|
||||
cache: npm
|
||||
cache-dependency-path: strix/interface/viewer/frontend/package-lock.json
|
||||
- run: npm ci --no-audit --no-fund
|
||||
- run: npm run build
|
||||
- name: Committed viewer bundle matches the build
|
||||
run: git -C "$GITHUB_WORKSPACE" diff --exit-code --stat -- strix/interface/viewer/static && test -z "$(git -C "$GITHUB_WORKSPACE" status --porcelain -- strix/interface/viewer/static)"
|
||||
+97
-28
@@ -15,38 +15,107 @@ concurrency:
|
||||
cancel-in-progress: true
|
||||
|
||||
jobs:
|
||||
python:
|
||||
uses: ./.github/workflows/ci-python.yml
|
||||
lint:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 15
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: ./.github/actions/setup-python-env
|
||||
- run: uv lock --check
|
||||
- run: uv run --frozen ruff check .
|
||||
- run: uv run --frozen ruff format --check .
|
||||
- run: uv run --frozen mypy strix/
|
||||
- run: uv run --frozen bandit -r strix/ -c pyproject.toml
|
||||
|
||||
pytest:
|
||||
name: pytest (${{ matrix.python-version }})
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 15
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
python-version: ['3.12', '3.14']
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: ./.github/actions/setup-python-env
|
||||
with:
|
||||
python-version: ${{ matrix.python-version }}
|
||||
- run: uv run --frozen pytest -q
|
||||
|
||||
tui:
|
||||
uses: ./.github/workflows/ci-tui.yml
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 15
|
||||
defaults:
|
||||
run:
|
||||
working-directory: strix/interface/tui
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: actions/setup-go@44694675825211faa026b3c33043df3e48a5fa00 # v6.0.0
|
||||
with:
|
||||
go-version-file: strix/interface/tui/go.mod
|
||||
cache-dependency-path: strix/interface/tui/go.sum
|
||||
- run: test -z "$(gofmt -l .)"
|
||||
- run: go vet ./...
|
||||
- run: CGO_ENABLED=0 go build -trimpath -o /dev/null ./cmd/strix-tui
|
||||
- run: go test -race ./...
|
||||
|
||||
viewer:
|
||||
uses: ./.github/workflows/ci-viewer.yml
|
||||
|
||||
package:
|
||||
uses: ./.github/workflows/ci-package.yml
|
||||
|
||||
workflows:
|
||||
uses: ./.github/workflows/ci-workflows.yml
|
||||
|
||||
ci-passed:
|
||||
name: ci-passed
|
||||
if: always()
|
||||
needs:
|
||||
- python
|
||||
- tui
|
||||
- viewer
|
||||
- package
|
||||
- workflows
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 15
|
||||
defaults:
|
||||
run:
|
||||
working-directory: strix/interface/viewer/frontend
|
||||
steps:
|
||||
- name: Every job succeeded
|
||||
env:
|
||||
NEEDS: ${{ toJSON(needs) }}
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
|
||||
with:
|
||||
node-version: '24'
|
||||
cache: npm
|
||||
cache-dependency-path: strix/interface/viewer/frontend/package-lock.json
|
||||
- run: npm ci --no-audit --no-fund
|
||||
- run: npm run build
|
||||
- name: Committed viewer bundle matches the build
|
||||
run: git -C "$GITHUB_WORKSPACE" diff --exit-code --stat -- strix/interface/viewer/static && test -z "$(git -C "$GITHUB_WORKSPACE" status --porcelain -- strix/interface/viewer/static)"
|
||||
|
||||
wheel:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 15
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: actions/setup-go@44694675825211faa026b3c33043df3e48a5fa00 # v6.0.0
|
||||
with:
|
||||
go-version-file: strix/interface/tui/go.mod
|
||||
cache-dependency-path: strix/interface/tui/go.sum
|
||||
- uses: ./.github/actions/setup-python-env
|
||||
- run: uv build --wheel
|
||||
- name: Wheel ships the TUI sidecar and the viewer bundle
|
||||
run: |
|
||||
failing=$(jq -r 'to_entries[] | select(.value.result != "success") | "\(.key): \(.value.result)"' <<< "$NEEDS")
|
||||
if [ -n "$failing" ]; then
|
||||
echo "$failing"
|
||||
exit 1
|
||||
fi
|
||||
uv run --frozen python - <<'PY'
|
||||
import glob, zipfile
|
||||
(wheel,) = glob.glob("dist/*.whl")
|
||||
names = zipfile.ZipFile(wheel).namelist()
|
||||
assert "strix/bin/strix-tui" in names, names
|
||||
assert "strix/interface/viewer/static/index.html" in names, names
|
||||
assert not any(name.startswith("strix/interface/viewer/frontend/") for name in names)
|
||||
assert not any(name.startswith("strix/interface/tui/cmd/") for name in names)
|
||||
PY
|
||||
- name: Install into a clean environment and run the CLI
|
||||
env:
|
||||
UV_EXCLUDE_NEWER: 5 days
|
||||
run: |
|
||||
uv venv --python 3.12 /tmp/strix-smoke
|
||||
uv pip install --python /tmp/strix-smoke/bin/python dist/*.whl
|
||||
/tmp/strix-smoke/bin/strix --version
|
||||
/tmp/strix-smoke/bin/strix --help >/dev/null
|
||||
- name: Vertex AI and Bedrock dependencies install with the wheel
|
||||
run: /tmp/strix-smoke/bin/python -c "import google.auth.transport.requests, boto3"
|
||||
@@ -1,14 +1,27 @@
|
||||
name: Workflows
|
||||
|
||||
on:
|
||||
workflow_call:
|
||||
pull_request:
|
||||
paths:
|
||||
- .github/**
|
||||
push:
|
||||
branches:
|
||||
- main
|
||||
paths:
|
||||
- .github/**
|
||||
workflow_dispatch:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.ref }}
|
||||
cancel-in-progress: true
|
||||
|
||||
jobs:
|
||||
actionlint:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 10
|
||||
env:
|
||||
ACTIONLINT_VERSION: 1.7.12
|
||||
ACTIONLINT_SCRIPT_REF: 914e7df21a07ef503a81201c76d2b11c789d3fca # v1.7.12
|
||||
@@ -27,6 +40,7 @@ jobs:
|
||||
|
||||
zizmor:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 10
|
||||
steps:
|
||||
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
|
||||
with:
|
||||
@@ -45,23 +45,3 @@ repos:
|
||||
hooks:
|
||||
- id: bandit
|
||||
args: [-c, pyproject.toml]
|
||||
|
||||
# Additional Python code quality checks
|
||||
- repo: https://github.com/asottile/pyupgrade
|
||||
rev: v3.21.2
|
||||
hooks:
|
||||
- id: pyupgrade
|
||||
args: [--py312-plus]
|
||||
|
||||
ci:
|
||||
autofix_commit_msg: |
|
||||
[pre-commit.ci] auto fixes from pre-commit.com hooks
|
||||
|
||||
for more information, see https://pre-commit.ci
|
||||
autofix_prs: true
|
||||
autoupdate_branch: ""
|
||||
autoupdate_commit_msg: "[pre-commit.ci] pre-commit autoupdate"
|
||||
autoupdate_schedule: weekly
|
||||
# pre-commit.ci cannot run `language: system` hooks; mypy runs via `make check-all`.
|
||||
skip: [mypy]
|
||||
submodules: false
|
||||
+68
-4
@@ -186,20 +186,84 @@ RUN set -eux; \
|
||||
aarch64|arm64) GITLEAKS_ARCH="arm64" ;; \
|
||||
*) echo "Unsupported architecture: $ARCH" >&2; exit 1 ;; \
|
||||
esac; \
|
||||
TAG="$(curl -fsSL https://api.github.com/repos/gitleaks/gitleaks/releases/latest | jq -r .tag_name)"; \
|
||||
TAG="$(curl -fsSLI -o /dev/null -w '%{url_effective}' https://github.com/gitleaks/gitleaks/releases/latest | sed 's#.*/##')"; \
|
||||
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
|
||||
|
||||
RUN curl -sfL https://raw.githubusercontent.com/aquasecurity/trivy/main/contrib/install.sh | sh -s -- -b /usr/local/bin
|
||||
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 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 semgrep && \
|
||||
pipx install bandit
|
||||
RUN pipx install bandit
|
||||
|
||||
RUN npm install -g jshint
|
||||
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
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,6 +19,22 @@ 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
|
||||
@@ -170,7 +186,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.3.0" type="string">
|
||||
<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.4.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 |
|
||||
| ------------------------------------------------------- | --------------------------------------------- |
|
||||
| [Semgrep](https://github.com/semgrep/semgrep) | Fast SAST and custom rule matching |
|
||||
| [Opengrep](https://github.com/opengrep/opengrep) | 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
|
||||
semgrep --config auto ./src
|
||||
opengrep scan --config /opt/opengrep-rules ./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 (`semgrep`, AST structural search, secrets, supply-chain checks) and then systematically validates top candidates dynamically.
|
||||
**White-box behavior**: Runs broad source-aware triage (`opengrep`, 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.3.0"
|
||||
STRIX_IMAGE="ghcr.io/usestrix/strix-sandbox:1.4.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 (`semgrep`, `ast-grep`, Tree-sitter tooling, `gitleaks`, `trufflehog`, `trivy fs`). Then review code for vulnerabilities
|
||||
- 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 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 `semgrep` 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 `opengrep` 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 `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.
|
||||
- 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.
|
||||
- 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, semgrep, 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, opengrep, 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:
|
||||
- semgrep - Static analysis/SAST
|
||||
- opengrep - Static analysis/SAST (bundled rules in /opt/opengrep-rules)
|
||||
- ast-grep (sg) - Structural AST/CST-aware code search
|
||||
- tree-sitter - Syntax-aware parsing and symbol extraction support
|
||||
- bandit - Python security linter
|
||||
|
||||
@@ -61,6 +61,15 @@ 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",
|
||||
@@ -155,7 +164,7 @@ class RuntimeSettings(BaseSettings):
|
||||
model_config = _BASE_CONFIG
|
||||
|
||||
image: str = Field(
|
||||
default="ghcr.io/usestrix/strix-sandbox:1.3.0",
|
||||
default="ghcr.io/usestrix/strix-sandbox:1.4.0",
|
||||
alias="STRIX_IMAGE",
|
||||
)
|
||||
backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
|
||||
|
||||
@@ -130,10 +130,6 @@ _DETAIL_PRIORITY_KEYS = (
|
||||
"finding_type",
|
||||
"cve",
|
||||
"cwe",
|
||||
"attack_techniques",
|
||||
"attack_source",
|
||||
"atlas_techniques",
|
||||
"atlas_source",
|
||||
"cvss",
|
||||
"filed_at",
|
||||
"target",
|
||||
|
||||
@@ -1,68 +0,0 @@
|
||||
"""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
|
||||
@@ -1 +0,0 @@
|
||||
{"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
|
||||
@@ -1,70 +0,0 @@
|
||||
"""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
|
||||
@@ -1 +0,0 @@
|
||||
{"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
|
||||
+20
-2
@@ -300,7 +300,15 @@ 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])
|
||||
|
||||
duplicate_id = str(parsed.get("duplicate_id") or "")[:64]
|
||||
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)
|
||||
reason = str(parsed.get("reason") or "")[:500]
|
||||
try:
|
||||
confidence = float(parsed.get("confidence", 0.0))
|
||||
@@ -308,13 +316,22 @@ def _parse_dedupe_response(content: str) -> dict[str, Any]:
|
||||
confidence = 0.0
|
||||
|
||||
return {
|
||||
"is_duplicate": bool(parsed.get("is_duplicate", False)),
|
||||
"is_duplicate": is_duplicate,
|
||||
"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:
|
||||
@@ -397,6 +414,7 @@ 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,8 +543,6 @@ def _result_properties(
|
||||
"method",
|
||||
"cve",
|
||||
"cwe",
|
||||
"attack_techniques",
|
||||
"atlas_techniques",
|
||||
"impact",
|
||||
"technical_analysis",
|
||||
"remediation_steps",
|
||||
|
||||
@@ -91,8 +91,6 @@ UPDATABLE_REPORT_FIELDS = frozenset(
|
||||
"method",
|
||||
"cve",
|
||||
"cwe",
|
||||
"attack_techniques",
|
||||
"atlas_techniques",
|
||||
"code_locations",
|
||||
"http_exchange_ids",
|
||||
"fix_verification",
|
||||
@@ -355,8 +353,6 @@ 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,
|
||||
@@ -415,10 +411,6 @@ 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:
|
||||
|
||||
+51
-2
@@ -8,11 +8,38 @@ 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."""
|
||||
@@ -37,6 +64,7 @@ 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
|
||||
@@ -73,6 +101,9 @@ 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
|
||||
|
||||
@@ -114,11 +145,23 @@ 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)
|
||||
record["cost"] = self.total_cost
|
||||
model, prices = _configured_prices()
|
||||
total_cost = self.total_cost
|
||||
record["cost"] = total_cost
|
||||
record["providers"] = {name: tally.model_dump() for name, tally in self._providers.items()}
|
||||
record["agents"] = []
|
||||
|
||||
@@ -128,7 +171,11 @@ 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._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
|
||||
)
|
||||
|
||||
agent_record = serialize_usage(usage)
|
||||
@@ -148,6 +195,7 @@ 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
|
||||
@@ -183,6 +231,7 @@ 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,8 +241,6 @@ 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): semgrep/AST/secrets/supply-chain static triage workflow
|
||||
- `source_aware_sast` (custom): opengrep/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
|
||||
|
||||
@@ -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 `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.
|
||||
- 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.
|
||||
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
|
||||
|
||||
- `semgrep`: fast security-first triage and custom pattern scans
|
||||
- `opengrep`: 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 `semgrep` pass
|
||||
- one `opengrep` pass
|
||||
- one AST structural pass (`sg` and/or `tree-sitter`)
|
||||
- one secrets pass (`gitleaks` and/or `trufflehog`)
|
||||
- one `trivy fs` pass
|
||||
|
||||
@@ -114,99 +114,103 @@ not just "upgrade <vulnerable pkg> to <fixed>".
|
||||
|
||||
### Usage / reachability analysis (required for every dependency CVE)
|
||||
|
||||
For every CVE you are about to report, run a static usage analysis and record
|
||||
the result in the structured `reachability` + `reachability_evidence` fields,
|
||||
and report every usage site you find as a `code_locations` entry (see Reporting).
|
||||
The level is an **evidence ladder, never an exploitability verdict** — claim
|
||||
only what you proved, and cite the proof. It never changes severity (that is
|
||||
`advisory_cvss` alone); it exists so the reader can prioritize.
|
||||
Every CVE you report carries a verdict in `reachability`, the proof in
|
||||
`reachability_evidence`, and every usage site in `code_locations` (see
|
||||
Reporting). The verdict answers one question for THIS advisory in THIS
|
||||
codebase: does the vulnerable code run here? There are three answers and no
|
||||
"unknown". The tool rejects a report without a verdict or without evidence,
|
||||
so finish the analysis before you file.
|
||||
|
||||
**Go — use govulncheck (real call-graph analysis):**
|
||||
- **`reachable`** — the advisory's vulnerable code executes in this codebase:
|
||||
application code, or a package it calls on the application's behalf, calls
|
||||
the affected function, mode or feature. Evidence: the call site as
|
||||
`file:line` and the source-to-sink trace (step 3).
|
||||
- **`not_reachable`** — the package is loaded by the shipped code, but this
|
||||
CVE's vulnerable path does not execute here. Evidence: what you checked and
|
||||
why the path is dead. Typical reasons: no caller of the affected API after
|
||||
a complete search; the only call site is behind a condition that is false
|
||||
here (a polyfill for an older runtime, a flag that is off, a dev-only
|
||||
command); the parent package of a transitive dependency never references
|
||||
the affected API; the flaw sits in a mode or option this code never uses;
|
||||
the argument that triggers the flaw is a constant here.
|
||||
- **`not_applicable`** — the advisory does not apply to this codebase at all:
|
||||
nothing the shipped code loads imports the package; it is build, test or
|
||||
dev tooling the artifact never executes; the installed copy is a patched or
|
||||
unaffected build; the flaw is specific to a platform this code does not run
|
||||
on; or the only code that could use it cannot be inspected (a compiled
|
||||
blob, a private bundle with no source) and nothing you can inspect reaches
|
||||
it. Evidence: the search scope and the negative result.
|
||||
|
||||
```bash
|
||||
# Symbol-level: reports only vulnerabilities whose vulnerable functions are
|
||||
# actually reachable from application code. Needs the Go toolchain + module
|
||||
# deps; if either is missing, fall back to the checks below rather than
|
||||
# claiming a level.
|
||||
if command -v govulncheck >/dev/null && go version >/dev/null 2>&1; then
|
||||
govulncheck -format json ./... > "$ART/govulncheck.json" || true
|
||||
fi
|
||||
```
|
||||
The verdict is per CVE, not per package. Sibling CVEs on one package usually
|
||||
affect different symbols or modes (one hits a parser, another a header
|
||||
check): run the check against each advisory's own affected functions and
|
||||
never copy one CVE's verdict or evidence onto another. The import check
|
||||
(step 1) is the only step shared across a package's CVEs.
|
||||
|
||||
- A finding with a call stack ⇒ `reachability=reachable_call_path`, put the
|
||||
call-path excerpt (entrypoint → vulnerable function) in
|
||||
`reachability_evidence`.
|
||||
- Listed as affecting a required module but with no reachable symbol ⇒ fall
|
||||
back to the import/symbol checks below (`imported` / `not_imported`).
|
||||
`reachable` is not an exploit, and `not_reachable` / `not_applicable` are
|
||||
not proof of safety (dynamic `import()`, reflection and framework wiring
|
||||
evade static search). Say what you checked; never write "not exploitable".
|
||||
|
||||
**All other ecosystems — import check, then symbol match:**
|
||||
**Procedure**
|
||||
|
||||
1. **Import check.** Search application code (exclude lockfiles, vendored
|
||||
deps, `node_modules`, build output) for imports of the vulnerable package:
|
||||
`ast-grep`/`rg` for `import`/`require`/`from X import` of the package (and
|
||||
its ecosystem import name, which may differ from the registry name, e.g.
|
||||
`PyYAML` → `yaml`). No hits ⇒ `not_imported`, with the search scope stated
|
||||
in `reachability_evidence`. For a **transitive** dependency, the check is
|
||||
whether application code imports it directly; if not, it is reachable only
|
||||
through the direct dependency — check whether the direct dep's usage can
|
||||
hit it (if unclear, use `imported` when the direct dep is used at all).
|
||||
2. **Symbol match — per CVE, not per package.** Read each CVE's own advisory
|
||||
(GHSA/NVD/OSV `affected[].ecosystem_specific.imports` or the advisory
|
||||
text) for the affected functions/classes/APIs. Search application code for
|
||||
those symbols (`ast-grep` pattern or `rg -n`). Hits ⇒
|
||||
`vulnerable_symbol_used`, with repo-relative `file:line` of each hit (up
|
||||
to a handful) in `reachability_evidence`. Imported but no affected-symbol
|
||||
usage found (or the advisory names no symbols) ⇒ `imported`.
|
||||
Different CVEs on the same package usually affect **different** symbols
|
||||
(one hits a parser, another a header check) — never copy one CVE's
|
||||
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,
|
||||
dynamic loading, unparsable sources) ⇒ `unknown` and say why in
|
||||
`assumptions`.
|
||||
deps, `node_modules`, build output) for imports of the vulnerable package
|
||||
under its ecosystem import name (`PyYAML` → `yaml`): `ast-grep`/`rg` for
|
||||
`import`/`require`/`from X import`. For a **transitive** dependency nothing
|
||||
in application code imports it, so walk every introducing chain from the
|
||||
lockfile (e.g. `express > body-parser > qs`) and inspect the code of the
|
||||
package one hop above the vulnerable one on each chain (`body-parser`, not
|
||||
just `express`; its installed sources, or the class references of its
|
||||
jar). Nothing imports it and no chain loads it ⇒ `not_applicable`, citing
|
||||
the scope you searched. Loaded ⇒ continue.
|
||||
2. **Affected-symbol check, per CVE.** Read the advisory (GHSA/NVD/OSV
|
||||
`affected[].ecosystem_specific.imports`, the advisory text, the fix commit
|
||||
when the text is vague) for the functions, classes, options or modes that
|
||||
trigger the flaw. Search application code, and the parent package of a
|
||||
transitive dependency, for them (`ast-grep` pattern or `rg -n`; sibling
|
||||
CVEs can share one multi-pattern search per package). No caller after a
|
||||
complete search ⇒ `not_reachable`, citing the symbols and the scope. A
|
||||
caller ⇒ step 3. When the flaw is in code every use of the package runs
|
||||
(the parser the package exists for), the import itself is the call site ⇒
|
||||
step 3.
|
||||
3. **Source-to-sink trace.** Start at the sink (the exact line that calls the
|
||||
affected function) and walk backwards hop by hop to the source: the entry
|
||||
point that carries the 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. Read what reaches the function: a constant argument that cannot
|
||||
contain the triggering pattern, a flag that is off, a runtime-version
|
||||
gate, a path only a dev tool runs means the vulnerable path does not run
|
||||
⇒ `not_reachable`, citing the condition. Otherwise ⇒ `reachable`. Write
|
||||
the chain into `reachability_evidence` as
|
||||
`entry point -> intermediate call -> package call` with a repo-relative
|
||||
`file:line` for every hop, what each hop enforces (authentication, a role
|
||||
check, validation, a size or type limit), and who controls the input. A
|
||||
trusted or operator-only source does not soften the verdict: the code
|
||||
runs, so it is `reachable`, and the trace is what tells the reader who can
|
||||
reach it; the contextual CVSS carries that.
|
||||
4. **Go — govulncheck first.** When the Go toolchain and module deps are
|
||||
present, `govulncheck -format json ./... > "$ART/govulncheck.json"` gives a
|
||||
real call graph: a finding with a call stack ⇒ `reachable`, with the call
|
||||
path (entrypoint → vulnerable function) as the trace; an affected module
|
||||
with no reachable symbol ⇒ `not_reachable` when application code imports
|
||||
it (cite the govulncheck result and the import), `not_applicable` when
|
||||
nothing imports it. When govulncheck cannot analyze the module (build
|
||||
errors, missing deps), run steps 1–3 instead of guessing.
|
||||
|
||||
Cheap-first budgeting: the import check is one search per package — always do
|
||||
it. Do the per-CVE symbol match for every CVE whose advisory names affected
|
||||
symbols (they can be batched into one multi-pattern search per package);
|
||||
prioritize `critical`/`high`/KEV when the budget is tight; a CVE whose symbol
|
||||
search was skipped may still be reported as `imported` (the import check is
|
||||
real evidence), but its `reachability_evidence` must state that the
|
||||
affected-symbol check was not performed, so a skipped search is never
|
||||
mistaken for a completed one with no hits. Never let this analysis stall
|
||||
reporting — `unknown` with a reason beats an unverified claim.
|
||||
Work in severity order (critical, high and KEV first) so that if time runs
|
||||
out the low tail is what is missing, and name that tail in the scan summary.
|
||||
A CVE whose analysis you did not finish is neither filed with a guessed
|
||||
verdict nor filed as `not_applicable`: finish it, or report the gap. Never let
|
||||
the usage analysis silently drop a CVE: every advisory the scanner matched is
|
||||
either filed with a verdict or named in the summary as not analyzed.
|
||||
|
||||
Anti-overclaim rules:
|
||||
### Reachability is a verdict on the code path, not on severity
|
||||
|
||||
- `not_imported` still does NOT mean safe (dynamic `import()`/reflection/
|
||||
framework wiring evade static search) — never phrase it as "not exploitable".
|
||||
- `reachable_call_path` is reserved for call-graph tools (govulncheck); a
|
||||
symbol grep hit is `vulnerable_symbol_used`, no matter how convinced you are.
|
||||
- The tool rejects any level other than `unknown` without
|
||||
`reachability_evidence`.
|
||||
|
||||
### Reachability is a confidence modifier, not a gate
|
||||
|
||||
Do NOT suppress or downgrade a known CVE just because you could not prove the
|
||||
vulnerable code path is reachable. Report it, set `advisory_cvss` from the
|
||||
advisory, record the usage analysis in `reachability`/`reachability_evidence`,
|
||||
and use `assumptions` for anything softer. If you *can* actually trigger the
|
||||
Do NOT suppress or downgrade a known CVE because its vulnerable code path is
|
||||
not reached. Report it with its verdict, set `advisory_cvss` from the
|
||||
advisory, put what the analysis proved in `reachability_evidence`, and use
|
||||
`assumptions` for anything softer. If you *can* actually trigger the
|
||||
vulnerable path or chain it into a dynamic exploit, additionally report that
|
||||
as a normal dynamic finding with `create_vulnerability_report` (the standalone
|
||||
CVE stays in its own `create_dependency_report`).
|
||||
@@ -230,8 +234,8 @@ findings and rejects empty PoC fields):
|
||||
pins the vulnerable version. Each entry needs the repo-relative `file`, the
|
||||
exact `start_line` / `end_line`, the verbatim `snippet` at those lines, and a
|
||||
short `label` (e.g. "imports js-yaml", "calls yaml.load on CLI input"). Read
|
||||
the real file to confirm the lines; never guess. When the package is
|
||||
`not_imported`, the manifest line is the only entry. If an entry carries
|
||||
the real file to confirm the lines; never guess. When the verdict is
|
||||
`not_applicable`, the manifest line is the only entry. If an entry carries
|
||||
`fix_before` / `fix_after`, also fill `fix_verification`.
|
||||
- **Always fill the structured dependency fields** (they power the dedicated
|
||||
dependency-report card; do not leave them only in free-text):
|
||||
@@ -260,9 +264,8 @@ findings and rejects empty PoC fields):
|
||||
- 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 —
|
||||
the tool rejects a report with no evidence, so for `unknown` write what you
|
||||
searched and why the result is inconclusive;
|
||||
use `assumptions` for anything softer (confidence, caveats, analysis limits).
|
||||
the tool rejects a report without a verdict or without evidence; use
|
||||
`assumptions` for anything softer (confidence, caveats, analysis limits).
|
||||
- **Always set `contextual_cvss_breakdown` + `contextual_cvss_reasoning`.** Every
|
||||
dependency finding carries a contextual rating of the CVE in this codebase
|
||||
(see below). Start from the published metrics and change only what your
|
||||
@@ -296,8 +299,8 @@ evidence proves is different in this codebase:
|
||||
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.
|
||||
codebase. A `not_applicable` package 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
|
||||
@@ -308,7 +311,7 @@ preconditions the hops enforce; `confidentiality`, `integrity`, and
|
||||
|
||||
When you have no source-to-sink trace, still rate the finding: copy the
|
||||
published metrics, change only the metrics the usage level itself proves, and
|
||||
say so in the reasoning. For example, for a `not_imported` package that the
|
||||
say so in the reasoning. For example, for a `not_applicable` package that the
|
||||
build still ships, keep the published metrics and lower `confidentiality`,
|
||||
`integrity`, and `availability` to `N`, because no code path reaches the
|
||||
vulnerable symbol. Never invent a hop you did not read.
|
||||
@@ -344,8 +347,8 @@ hallucinate a CVE id.
|
||||
- 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 claim a `reachability` verdict the evidence does not prove, and do
|
||||
not file a CVE whose analysis you did not finish — report the gap instead.
|
||||
- Do not send a report without `contextual_cvss_breakdown` and
|
||||
`contextual_cvss_reasoning` — the reader rates and ranks the finding with them.
|
||||
- Do not use the contextual breakdown to quietly de-rate a CVE you could not
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
---
|
||||
name: source-aware-sast
|
||||
description: Practical source-aware SAST and AST playbook for semgrep, ast-grep, gitleaks, and trivy fs
|
||||
description: Practical source-aware SAST and AST playbook for opengrep, 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"
|
||||
|
||||
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)
|
||||
opengrep scan --config /opt/opengrep-rules --taint-intrafile \
|
||||
--json --output "$ART/opengrep.json" .
|
||||
# Build deterministic AST targets from opengrep scope (no hardcoded path guessing)
|
||||
python3 - <<'PY'
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
art = Path("/workspace/.source-aware")
|
||||
semgrep_json = art / "semgrep.json"
|
||||
opengrep_json = art / "opengrep.json"
|
||||
targets_file = art / "sg-targets.txt"
|
||||
|
||||
try:
|
||||
data = json.loads(semgrep_json.read_text(encoding="utf-8"))
|
||||
data = json.loads(opengrep_json.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
targets_file.write_text("", encoding="utf-8")
|
||||
raise
|
||||
@@ -63,17 +63,18 @@ trivy fs --scanners vuln,misconfig --timeout 30m --offline-scan \
|
||||
--format json --output "$ART/trivy-fs.json" . || true
|
||||
```
|
||||
|
||||
## Semgrep First Pass
|
||||
## Opengrep First Pass
|
||||
|
||||
Use Semgrep as the default static triage pass:
|
||||
Use Opengrep as the default static triage pass. Rules are bundled offline in `/opt/opengrep-rules` (one directory per language, plus `generic/secrets`):
|
||||
|
||||
```bash
|
||||
# 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 .
|
||||
# Full bundled ruleset
|
||||
opengrep scan --config /opt/opengrep-rules --taint-intrafile \
|
||||
--json --output /workspace/.source-aware/opengrep.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 .
|
||||
# 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 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`, `semgrep`, `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`, `opengrep`, `bandit`, `ast-grep`, and `ripgrep`.
|
||||
|
||||
- **bandit** (preinstalled) — Python security linter; flags `mark_safe`, `extra()`, `RawSQL`, `subprocess`, weak crypto, hardcoded secrets: `bandit -r . -ll`
|
||||
- **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 .`
|
||||
- **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 .`
|
||||
- **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 (`semgrep`, `ast-grep`, `gitleaks`, `trufflehog`, `trivy fs`) and use outputs to drive deep review
|
||||
- Start with broad source-aware triage (`opengrep`, `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 (`semgrep`, then targeted `sg` queries)
|
||||
- Run a fast static triage on changed files first (`opengrep`, 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 `semgrep` first-pass triage to prioritize risky flows before deep manual review
|
||||
- Run `opengrep` 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)
|
||||
|
||||
@@ -207,7 +207,7 @@ agent-browser snapshot -i
|
||||
```
|
||||
|
||||
Credentials in shell history are a leak. For anything sensitive, use the
|
||||
auth vault (see [references/authentication.md](references/authentication.md)):
|
||||
auth vault (see `references/authentication.md`):
|
||||
|
||||
```bash
|
||||
agent-browser auth save my-app --url https://app.example.com/login \
|
||||
@@ -367,7 +367,7 @@ agent-browser click @e3
|
||||
agent-browser record stop
|
||||
```
|
||||
|
||||
See [references/video-recording.md](references/video-recording.md) for
|
||||
See `references/video-recording.md` for
|
||||
codec options, GIF export, and more.
|
||||
|
||||
### Iframes
|
||||
@@ -491,8 +491,8 @@ snapshot — fall back to `eval` in the iframe's origin or use the
|
||||
|
||||
**Authentication expires mid-workflow**
|
||||
Use `--session-name <name>` or `state save`/`state load` so your session
|
||||
survives browser restarts. See [references/session-management.md](references/session-management.md)
|
||||
and [references/authentication.md](references/authentication.md).
|
||||
survives browser restarts. See `references/session-management.md`
|
||||
and `references/authentication.md`.
|
||||
|
||||
## Global flags worth knowing
|
||||
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
---
|
||||
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>`
|
||||
@@ -1,72 +0,0 @@
|
||||
---
|
||||
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: semantic-confusion
|
||||
description: Cross-component semantic confusion testing for parser differentials, normalization mismatches, overloaded fields, lifecycle state drift, internal redirects, protocol translation, and validator-to-sink inconsistencies
|
||||
description: Cross-component semantic confusion testing for parser differentials, normalization mismatches, email and IDNA identity drift, overloaded fields, lifecycle state drift, internal redirects, protocol translation, and validator-to-sink inconsistencies
|
||||
---
|
||||
|
||||
# Semantic Confusion
|
||||
@@ -59,6 +59,27 @@ The highest-signal condition is `security_check(value_A)` followed by `sink(tran
|
||||
- Test malformed Unicode and replacement behavior; a rejected code point may become an allowed delimiter or wildcard later.
|
||||
- Test path, query, and fragment separately. Browsers and routers commonly transform each source differently.
|
||||
|
||||
### Email and Account Identity Drift
|
||||
|
||||
Treat an email address as both a principal selector and a delivery destination. Trace the complete boundary before testing:
|
||||
|
||||
```text
|
||||
submitted or identity-provider address
|
||||
-> application normalization / IDNA
|
||||
-> database comparison and collation
|
||||
-> selected local principal
|
||||
-> token, session, invitation, or link binding
|
||||
-> recipient or linked identity consumed by the final service
|
||||
```
|
||||
|
||||
- Cover recovery, magic links, invitations, registration uniqueness, email changes, and OAuth/OIDC account linking. Test case/accent-insensitive collations, Unicode normalization, IDNA U-label/A-label conversion, trailing dots, and provider canonicalization only when the observed stack supports them.
|
||||
- In black-box testing, report the observed equivalence and routing behavior. Do not name a specific collation, normalizer, or provider algorithm without source, configuration, or version evidence.
|
||||
- Determine whether two non-identical addresses can select the same protected principal while SMTP or an identity provider routes them to different owners. The address that receives authority must identify the same principal selected by the security check.
|
||||
- In recovery flows, compare the submitted address with the selected account's stored canonical address. A fuzzy lookup followed by delivery to the submitted value is high risk; an exact lookup plus delivery to the stored value is the primary negative control.
|
||||
- Use only an explicitly authorized attacker-controlled mailbox or test domain. Derive a bounded variant set from the deployed collation and canonicalizers rather than spraying external domains or the Unicode range.
|
||||
|
||||
For a recovery, magic-link, or invitation finding, a generic success response is not proof. Demonstrate that the variant selects the protected principal, the issued authority reaches the controlled mailbox or identity, and redemption yields the protected principal's access. If delivery or redemption cannot be observed, keep the path `needs_follow_up`.
|
||||
|
||||
### Field and Type Overloading
|
||||
|
||||
- Identify shared fields reused for different concepts: path vs URL, content type vs handler, display name vs executable name, route vs filesystem location.
|
||||
@@ -170,6 +191,8 @@ A valid confusion finding should include:
|
||||
- Different error messages with identical final authorization and sink behavior
|
||||
- A parser accepts odd syntax but downstream consumers preserve the same safe meaning
|
||||
- A normalization difference visible only in logs, with no security decision between representations
|
||||
- A Unicode or Punycode lookalike that neither selects another principal nor receives authority intended for that principal
|
||||
- A permissive database comparison where recovery delivery still uses the selected account's stored canonical address
|
||||
- WAF bypass where the application itself rejects the request identically
|
||||
- Version-specific behavior claimed as universal without testing the relevant deployment
|
||||
- A search-path candidate that is attacker-named but cannot be created, claimed, loaded, or executed
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
"""In-memory bearer refresh for HTTP MCP connections."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import AsyncGenerator, Awaitable, Callable
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RefreshingBearerAuth(httpx.Auth):
|
||||
def __init__(
|
||||
self, token: str | None, refresh_token: Callable[[], Awaitable[str | None]]
|
||||
) -> None:
|
||||
self._token = token
|
||||
self._refresh_token = refresh_token
|
||||
self._lock = asyncio.Lock()
|
||||
self._refresh_generation = 0
|
||||
|
||||
def _authorize(self, request: httpx.Request, token: str | None) -> None:
|
||||
if token:
|
||||
request.headers["Authorization"] = f"Bearer {token}"
|
||||
else:
|
||||
request.headers.pop("Authorization", None)
|
||||
|
||||
async def async_auth_flow(
|
||||
self, request: httpx.Request
|
||||
) -> AsyncGenerator[httpx.Request, httpx.Response]:
|
||||
sent_token = self._token
|
||||
sent_generation = self._refresh_generation
|
||||
self._authorize(request, sent_token)
|
||||
response = yield request
|
||||
if response.status_code != 401:
|
||||
return
|
||||
async with self._lock:
|
||||
if self._token == sent_token and self._refresh_generation == sent_generation:
|
||||
try:
|
||||
token = await self._refresh_token()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("MCP bearer refresh failed (%s)", type(exc).__name__)
|
||||
token = None
|
||||
self._refresh_generation += 1
|
||||
if token and token != sent_token:
|
||||
self._token = token
|
||||
retry_token = self._token
|
||||
if retry_token and retry_token != sent_token:
|
||||
self._authorize(request, retry_token)
|
||||
yield request
|
||||
@@ -39,10 +39,11 @@ from strix.tools.mcp.session import McpConnectionUnavailableError, SupervisedMcp
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Awaitable, Callable, Mapping
|
||||
|
||||
import httpx
|
||||
|
||||
from strix.tools.mcp.auth import RefreshingBearerAuth
|
||||
from strix.tools.mcp.config import McpConnectionConfig
|
||||
from strix.tools.mcp.registry import McpConnectionRequest, McpRegistry
|
||||
|
||||
@@ -121,7 +122,9 @@ class _QuietMCPServerStdio(MCPServerStdio):
|
||||
return _quiet_stdio_streams(self.params)
|
||||
|
||||
|
||||
def _build_server(config: McpConnectionConfig) -> BuiltMcpServer:
|
||||
def _build_server(
|
||||
config: McpConnectionConfig, bearer_auth: RefreshingBearerAuth | None = None
|
||||
) -> BuiltMcpServer:
|
||||
"""Construct (but do not connect) the SDK server for one connection.
|
||||
|
||||
The returned tuple carries the server and, for HTTP connections, a recorder
|
||||
@@ -161,13 +164,13 @@ def _build_server(config: McpConnectionConfig) -> BuiltMcpServer:
|
||||
timeout: httpx.Timeout | None = None,
|
||||
auth: httpx.Auth | None = None,
|
||||
) -> httpx.AsyncClient:
|
||||
client = create_mcp_http_client(headers=headers, timeout=timeout, auth=auth)
|
||||
client = create_mcp_http_client(headers=headers, timeout=timeout, auth=bearer_auth or auth)
|
||||
client.event_hooks.setdefault("response", []).append(recorder)
|
||||
return client
|
||||
|
||||
http_params: MCPServerStreamableHttpParams = {
|
||||
"url": cast("str", config.url),
|
||||
"headers": _auth_headers(config),
|
||||
"headers": {} if bearer_auth is not None else _auth_headers(config),
|
||||
"timeout": config.http_timeout_seconds,
|
||||
"sse_read_timeout": config.sse_read_timeout_seconds,
|
||||
"httpx_client_factory": httpx_client_factory,
|
||||
@@ -300,9 +303,12 @@ async def connect_mcp_servers(
|
||||
configs: list[McpConnectionConfig],
|
||||
*,
|
||||
max_concurrency: int = 6,
|
||||
token_refreshers: Mapping[str, Callable[[], Awaitable[str | None]]] | None = None,
|
||||
) -> list[ConnectedMcpServer]:
|
||||
"""Connect MCP configs concurrently under a fixed bound.
|
||||
|
||||
``token_refreshers`` optionally supplies a bearer refresher by connection name.
|
||||
|
||||
Each connection becomes a :class:`~strix.tools.mcp.session.SupervisedMcpSession`
|
||||
that owns ``connect()``, the held-open session, and ``cleanup()`` on one
|
||||
dedicated task, so a later background failure in one session is contained to
|
||||
@@ -325,7 +331,10 @@ async def connect_mcp_servers(
|
||||
|
||||
async def connect_one(config: McpConnectionConfig) -> ConnectedMcpServer | None:
|
||||
async with semaphore:
|
||||
session = SupervisedMcpSession(config)
|
||||
session = SupervisedMcpSession(
|
||||
config,
|
||||
refresh_token=token_refreshers.get(config.name) if token_refreshers else None,
|
||||
)
|
||||
sessions.append(session)
|
||||
try:
|
||||
if not await session.start():
|
||||
@@ -381,7 +390,14 @@ async def attach_mcp_requests(
|
||||
ends).
|
||||
"""
|
||||
request_by_name = {request.config.name: request for request in requests}
|
||||
connections = await connect_mcp_servers([request.config for request in requests])
|
||||
connections = await connect_mcp_servers(
|
||||
[request.config for request in requests],
|
||||
token_refreshers={
|
||||
request.config.name: request.refresh_token
|
||||
for request in requests
|
||||
if request.refresh_token is not None
|
||||
},
|
||||
)
|
||||
for connection in connections:
|
||||
request = request_by_name[connection.name]
|
||||
registry.add(
|
||||
@@ -391,5 +407,6 @@ async def attach_mcp_requests(
|
||||
purpose=request.purpose or connection.notes,
|
||||
provider=request.provider,
|
||||
result_transform=request.result_transform,
|
||||
refresh_token=request.refresh_token,
|
||||
)
|
||||
return connections
|
||||
@@ -129,12 +129,20 @@ def classify(exc: BaseException) -> FailureInfo:
|
||||
|
||||
|
||||
class HttpStatusRecorder:
|
||||
"""Capture the last non-success response from one HTTP connection."""
|
||||
"""Capture the last non-success response from one HTTP connection.
|
||||
|
||||
A later success clears a recorded 401, so a bearer refreshed and retried
|
||||
in the auth flow does not leave a stale auth failure behind.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._failure: FailureInfo | None = None
|
||||
|
||||
async def __call__(self, response: httpx.Response) -> None:
|
||||
if 200 <= response.status_code < 300:
|
||||
if self._failure is not None and self._failure.kind == "auth":
|
||||
self._failure = None
|
||||
return
|
||||
if not 200 <= response.status_code < 300:
|
||||
request = response.request
|
||||
self._failure = _from_status(
|
||||
|
||||
@@ -32,7 +32,7 @@ from strix.tools.mcp.session import McpConnectionUnavailableError, SupervisedMcp
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from agents.mcp import MCPServer
|
||||
|
||||
@@ -96,6 +96,9 @@ class McpConnectionEntry:
|
||||
tool_count: int = 0
|
||||
result_transform: ResultTransform | None = None
|
||||
provider: str | None = None
|
||||
refresh_token: Callable[[], Awaitable[str | None]] | None = dataclasses.field(
|
||||
default=None, repr=False
|
||||
)
|
||||
state: McpConnectionState = "configured"
|
||||
_catalog: list[Any] | None = dataclasses.field(default=None, repr=False)
|
||||
_connect_task: asyncio.Task[SupervisedMcpSession] | None = dataclasses.field(
|
||||
@@ -180,7 +183,7 @@ class McpConnectionEntry:
|
||||
f"MCP connection {self.name!r} has no connection configuration."
|
||||
)
|
||||
self._set_state("connecting")
|
||||
session = SupervisedMcpSession(config)
|
||||
session = SupervisedMcpSession(config, refresh_token=self.refresh_token)
|
||||
try:
|
||||
started = await session.start()
|
||||
except BaseException:
|
||||
@@ -298,12 +301,17 @@ class McpConnectionRequest:
|
||||
at the single dispatch point (strix-pro's sanitizer; empty for the
|
||||
command-line path). ``purpose`` is the human label ``list_mcps`` reports as the
|
||||
connection's description; when unset it falls back to ``config.notes``.
|
||||
``refresh_token`` optionally fetches a new bearer asynchronously after an HTTP
|
||||
401; returning no token or the current token leaves the failure unchanged.
|
||||
"""
|
||||
|
||||
config: McpConnectionConfig
|
||||
provider: str | None = None
|
||||
result_transform: ResultTransform | None = None
|
||||
purpose: str | None = None
|
||||
refresh_token: Callable[[], Awaitable[str | None]] | None = dataclasses.field(
|
||||
default=None, repr=False
|
||||
)
|
||||
|
||||
|
||||
class McpCallInfo(NamedTuple):
|
||||
@@ -337,6 +345,7 @@ class McpRegistry:
|
||||
purpose=request.purpose or request.config.notes,
|
||||
result_transform=request.result_transform,
|
||||
provider=request.provider,
|
||||
refresh_token=request.refresh_token,
|
||||
)
|
||||
entry.set_status_sink(self._status_sink)
|
||||
self._entries[entry.name] = entry
|
||||
@@ -353,6 +362,7 @@ class McpRegistry:
|
||||
tool_count: int = 0,
|
||||
result_transform: ResultTransform | None = None,
|
||||
provider: str | None = None,
|
||||
refresh_token: Callable[[], Awaitable[str | None]] | None = None,
|
||||
) -> McpConnectionEntry:
|
||||
"""Register one connection under ``name`` (last write wins).
|
||||
|
||||
@@ -374,6 +384,7 @@ class McpRegistry:
|
||||
tool_count=tool_count,
|
||||
result_transform=result_transform,
|
||||
provider=provider,
|
||||
refresh_token=refresh_token,
|
||||
state="connected",
|
||||
)
|
||||
entry.set_status_sink(self._status_sink)
|
||||
|
||||
@@ -54,6 +54,7 @@ import time
|
||||
import weakref
|
||||
from typing import TYPE_CHECKING, Any, Literal, cast
|
||||
|
||||
from strix.tools.mcp.auth import RefreshingBearerAuth
|
||||
from strix.tools.mcp.config import DEFAULT_MAX_CONCURRENT_CALLS
|
||||
from strix.tools.mcp.failures import FailureInfo, HttpStatusRecorder, classify
|
||||
|
||||
@@ -157,9 +158,19 @@ class SupervisedMcpSession:
|
||||
:attr:`name`, :attr:`server`, :attr:`config`, :attr:`is_dead`.
|
||||
"""
|
||||
|
||||
def __init__(self, config: McpConnectionConfig) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
config: McpConnectionConfig,
|
||||
*,
|
||||
refresh_token: Callable[[], Awaitable[str | None]] | None = None,
|
||||
) -> None:
|
||||
self._name = config.name
|
||||
self._config: McpConnectionConfig | None = config
|
||||
self._bearer_auth = (
|
||||
RefreshingBearerAuth(config.auth.token if config.auth else None, refresh_token)
|
||||
if refresh_token is not None and config.transport == "http"
|
||||
else None
|
||||
)
|
||||
self._server: MCPServer | None = None
|
||||
self._supervised = True
|
||||
self._task: asyncio.Task[None] | None = None
|
||||
@@ -193,6 +204,7 @@ class SupervisedMcpSession:
|
||||
self = cls.__new__(cls)
|
||||
self._name = name
|
||||
self._config = config
|
||||
self._bearer_auth = None
|
||||
self._server = server
|
||||
self._supervised = False
|
||||
self._task = None
|
||||
@@ -701,7 +713,11 @@ class SupervisedMcpSession:
|
||||
|
||||
if self._config is None:
|
||||
raise RuntimeError(f"MCP connection {self._name!r} has no config to connect")
|
||||
built = _build_server(self._config)
|
||||
built = (
|
||||
_build_server(self._config, self._bearer_auth)
|
||||
if self._bearer_auth is not None
|
||||
else _build_server(self._config)
|
||||
)
|
||||
server = built.server
|
||||
self._recorder = built.recorder
|
||||
try:
|
||||
|
||||
+51
-110
@@ -16,8 +16,6 @@ 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
|
||||
|
||||
@@ -464,20 +462,6 @@ 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)
|
||||
@@ -819,8 +803,6 @@ 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]:
|
||||
@@ -863,10 +845,6 @@ 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
|
||||
)
|
||||
@@ -929,8 +907,6 @@ 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,
|
||||
@@ -1024,8 +1000,6 @@ 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,
|
||||
@@ -1322,18 +1296,6 @@ 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``.
|
||||
@@ -1488,8 +1450,6 @@ 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,
|
||||
@@ -1525,8 +1485,6 @@ 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,
|
||||
@@ -1600,9 +1558,6 @@ 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.
|
||||
@@ -1649,8 +1604,6 @@ 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,
|
||||
@@ -1736,15 +1689,7 @@ def _dependency_severity(advisory_cvss: float | None) -> tuple[float, str]:
|
||||
return score, "none"
|
||||
|
||||
|
||||
_VALID_REACHABILITY = frozenset(
|
||||
{
|
||||
"not_imported",
|
||||
"imported",
|
||||
"vulnerable_symbol_used",
|
||||
"reachable_call_path",
|
||||
"unknown",
|
||||
}
|
||||
)
|
||||
_VALID_REACHABILITY = frozenset({"reachable", "not_reachable", "not_applicable"})
|
||||
|
||||
|
||||
def _validate_manifest_path(manifest_path: str | None) -> str | None:
|
||||
@@ -1885,11 +1830,13 @@ def _build_dependency_metadata(
|
||||
|
||||
|
||||
_REACHABILITY_EVIDENCE_LABELS = {
|
||||
"not_imported": "not imported by application code",
|
||||
"imported": "imported by application code; affected API usage unconfirmed",
|
||||
"vulnerable_symbol_used": "the advisory's affected API is used in application code",
|
||||
"reachable_call_path": (
|
||||
"a call path from application code to the vulnerable function was proven"
|
||||
"reachable": "the advisory's vulnerable code runs in this codebase",
|
||||
"not_reachable": (
|
||||
"the package is loaded, but this advisory's vulnerable code does not run here"
|
||||
),
|
||||
"not_applicable": (
|
||||
"the advisory does not apply here: the vulnerable package is not loaded, "
|
||||
"or the installed copy is not affected"
|
||||
),
|
||||
}
|
||||
|
||||
@@ -1946,13 +1893,12 @@ async def _do_create_dependency( # noqa: PLR0912, PLR0915
|
||||
introduced_by: str | None = None,
|
||||
dependency_path: str | None = None,
|
||||
manifest_path: str | None = None,
|
||||
reachability: str = "unknown",
|
||||
reachability: str | None = None,
|
||||
reachability_evidence: str | None = None,
|
||||
code_locations: list[dict[str, Any]] | None = None,
|
||||
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]:
|
||||
@@ -1983,9 +1929,6 @@ 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(
|
||||
@@ -2006,17 +1949,22 @@ async def _do_create_dependency( # noqa: PLR0912, PLR0915
|
||||
)
|
||||
errors.extend(_validate_fix_verification(parsed_locations, fix_verification))
|
||||
|
||||
reachability = (reachability or "unknown").strip().lower()
|
||||
if reachability not in _VALID_REACHABILITY:
|
||||
reachability = (reachability or "").strip().lower()
|
||||
if not reachability:
|
||||
errors.append(
|
||||
"reachability is required: 'reachable', 'not_reachable' or 'not_applicable'. "
|
||||
"There is no 'unknown': finish the usage analysis for this CVE before filing it."
|
||||
)
|
||||
elif reachability not in _VALID_REACHABILITY:
|
||||
errors.append(
|
||||
f"Invalid reachability: {reachability!r}. Must be one of: {sorted(_VALID_REACHABILITY)}"
|
||||
)
|
||||
elif not (reachability_evidence or "").strip():
|
||||
errors.append(
|
||||
"reachability_evidence is required: cite the concrete proof (import "
|
||||
"file:line, matched symbol usage, or govulncheck call path), or, for "
|
||||
"'unknown', say what you searched and why the result is inconclusive. "
|
||||
"Never claim a reachability level without evidence."
|
||||
"reachability_evidence is required: cite the concrete proof (the call site "
|
||||
"and its source-to-sink trace for 'reachable'; the complete search or the "
|
||||
"condition that keeps the vulnerable path from running for 'not_reachable'; "
|
||||
"the search scope for 'not_applicable'). Never claim a verdict without evidence."
|
||||
)
|
||||
|
||||
errors.extend(_validate_contextual_cvss(contextual_cvss_breakdown, contextual_cvss_reasoning))
|
||||
@@ -2112,7 +2060,6 @@ 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",
|
||||
@@ -2167,13 +2114,12 @@ async def create_dependency_report(
|
||||
fix_effort: str = "low",
|
||||
introduced_by: str | None = None,
|
||||
dependency_path: str | None = None,
|
||||
reachability: str = "unknown",
|
||||
reachability: str | None = None,
|
||||
reachability_evidence: str | None = None,
|
||||
code_locations: list[dict[str, Any]] | None = None,
|
||||
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.
|
||||
|
||||
@@ -2198,23 +2144,26 @@ async def create_dependency_report(
|
||||
- Re-reporting the same CVE/package already filed.
|
||||
|
||||
**Reachability**: do NOT silently downgrade or suppress a finding
|
||||
because the vulnerable code path may be unreachable — report it, and
|
||||
record what the usage analysis showed via the structured
|
||||
``reachability`` + ``reachability_evidence`` fields (see the
|
||||
dependency-cve-scanning skill for the analysis procedure). The level
|
||||
is an evidence ladder, never an exploitability verdict:
|
||||
because the vulnerable code path is not reached — report it, and record
|
||||
the verdict of the usage analysis in the structured ``reachability`` +
|
||||
``reachability_evidence`` fields (see the dependency-cve-scanning skill
|
||||
for the procedure). The verdict answers one question for this advisory
|
||||
in this codebase, "does the vulnerable code run here?", and the tool
|
||||
rejects a report without it or without evidence:
|
||||
|
||||
- ``not_imported`` — the package is never imported/required by
|
||||
application code (strongest de-prioritization signal; still not
|
||||
proof of safety — dynamic loading, reflection, or framework wiring
|
||||
can evade static search).
|
||||
- ``imported`` — application code imports the package, but usage of
|
||||
the advisory's affected API was not confirmed.
|
||||
- ``vulnerable_symbol_used`` — the advisory's affected
|
||||
function/class/API appears in application code.
|
||||
- ``reachable_call_path`` — a call-graph tool (e.g. ``govulncheck``)
|
||||
proved a path from application code to the vulnerable function.
|
||||
- ``unknown`` — usage analysis was not performed or was inconclusive.
|
||||
- ``reachable`` — the advisory's vulnerable code executes here:
|
||||
application code, or a package it calls on the application's behalf,
|
||||
calls the affected function, mode or feature.
|
||||
- ``not_reachable`` — the package is loaded, but this advisory's
|
||||
vulnerable path does not execute here (no caller of the affected API
|
||||
after a complete search, a call behind a condition that is false here,
|
||||
a parent package that never references the affected API, a mode this
|
||||
code never uses, a triggering argument that is a constant here).
|
||||
- ``not_applicable`` — the advisory does not apply to this codebase:
|
||||
nothing the shipped code loads imports the package, it is build or
|
||||
test tooling the artifact never runs, the installed copy is a patched
|
||||
or unaffected build, or the flaw is specific to a platform this code
|
||||
does not run on.
|
||||
|
||||
Severity comes from ``contextual_cvss_breakdown`` when you provide one
|
||||
(computed exactly like a normal finding's ``cvss_breakdown``), otherwise
|
||||
@@ -2238,10 +2187,6 @@ 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
|
||||
@@ -2266,26 +2211,24 @@ async def create_dependency_report(
|
||||
checkout directory prefix so the path is relative to the
|
||||
repository root. This binds the finding to its exact file so
|
||||
remediation can target the right repository.
|
||||
reachability: Usage-evidence level from static analysis — one of
|
||||
``not_imported`` / ``imported`` / ``vulnerable_symbol_used`` /
|
||||
``reachable_call_path`` / ``unknown``. Claim only what the
|
||||
evidence proves; when in doubt use ``unknown``.
|
||||
reachability: **Required.** The verdict of the usage analysis for
|
||||
this CVE — ``reachable`` / ``not_reachable`` / ``not_applicable``.
|
||||
Claim only what the evidence proves; there is no ``unknown``,
|
||||
finish the analysis before filing.
|
||||
reachability_evidence: **Required.** The concrete proof for the
|
||||
claimed level, or, for ``unknown``, what you searched and why
|
||||
the result is inconclusive: 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
|
||||
verdict: repo-relative ``file:line`` of the import and of the
|
||||
call that reaches the affected symbol, the search scope and
|
||||
negative result behind ``not_applicable``, or the condition that
|
||||
keeps the vulnerable path from running behind ``not_reachable``.
|
||||
Whenever the affected symbol is called, 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
|
||||
that carries the 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.
|
||||
is off in production), and say who controls the input.
|
||||
code_locations: Where application code imports or calls the package —
|
||||
the evidence behind your ``reachability`` claim. List of dicts,
|
||||
same shape as ``create_vulnerability_report``::
|
||||
@@ -2367,7 +2310,6 @@ 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,
|
||||
)
|
||||
@@ -2396,7 +2338,6 @@ _REPORT_SUMMARY_FIELDS = (
|
||||
"finding_class",
|
||||
"cve",
|
||||
"cwe",
|
||||
"attack_techniques",
|
||||
"target",
|
||||
"endpoint",
|
||||
"method",
|
||||
|
||||
@@ -1,19 +0,0 @@
|
||||
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"
|
||||
@@ -0,0 +1,159 @@
|
||||
"""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
|
||||
+41
-1
@@ -2,11 +2,14 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING, 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,
|
||||
@@ -15,6 +18,11 @@ 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:
|
||||
@@ -85,6 +93,38 @@ 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,
|
||||
|
||||
@@ -0,0 +1,223 @@
|
||||
"""HTTP bearer refresh regressions without network access."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Any, cast
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from mcp.types import CallToolResult, TextContent, Tool
|
||||
|
||||
from strix.tools.mcp import BearerAuth, McpConnectionConfig, McpConnectionRequest, McpRegistry
|
||||
from strix.tools.mcp import client as mcp_client
|
||||
from strix.tools.mcp import session as mcp_session
|
||||
from strix.tools.mcp.auth import RefreshingBearerAuth
|
||||
|
||||
|
||||
def _config() -> McpConnectionConfig:
|
||||
return McpConnectionConfig(
|
||||
name="refresh",
|
||||
url="https://example.test/mcp",
|
||||
auth=BearerAuth(token="old"), # noqa: S106 # nosec B106
|
||||
)
|
||||
|
||||
|
||||
def _client(auth: RefreshingBearerAuth | None, handler: Any) -> httpx.AsyncClient:
|
||||
built = mcp_client._build_server(_config(), auth)
|
||||
params = cast("Any", built.server).params
|
||||
assert params["headers"] == ({} if auth else {"Authorization": "Bearer old"})
|
||||
client = cast("httpx.AsyncClient", params["httpx_client_factory"](headers=params["headers"]))
|
||||
client._transport = httpx.MockTransport(handler)
|
||||
return client
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_retries_once_and_persists_across_server_builds() -> None:
|
||||
refresh = AsyncMock(return_value="new")
|
||||
session = mcp_session.SupervisedMcpSession(_config(), refresh_token=refresh)
|
||||
seen: list[str] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
seen.append(request.headers["Authorization"])
|
||||
return httpx.Response(200 if seen[-1] == "Bearer new" else 401)
|
||||
|
||||
for _ in range(2):
|
||||
async with _client(session._bearer_auth, handler) as client:
|
||||
response = await client.post("https://example.test/mcp", content=b"payload")
|
||||
assert response.status_code == 200
|
||||
assert seen == ["Bearer old", "Bearer new", "Bearer new"]
|
||||
refresh.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("replacement", ["new", None, "old", "raises"])
|
||||
async def test_concurrent_unauthorized_requests_share_refresh(replacement: str | None) -> None:
|
||||
refresh = AsyncMock(return_value=replacement)
|
||||
if replacement == "raises":
|
||||
refresh.side_effect = ValueError("refresh unavailable")
|
||||
auth = RefreshingBearerAuth("old", refresh)
|
||||
both_sent = asyncio.Event()
|
||||
old_requests = 0
|
||||
|
||||
async def handler(request: httpx.Request) -> httpx.Response:
|
||||
nonlocal old_requests
|
||||
if request.headers["Authorization"] == "Bearer old":
|
||||
old_requests += 1
|
||||
if old_requests == 5:
|
||||
both_sent.set()
|
||||
await both_sent.wait()
|
||||
return httpx.Response(401)
|
||||
return httpx.Response(200)
|
||||
|
||||
async with _client(auth, handler) as client:
|
||||
responses = await asyncio.wait_for(
|
||||
asyncio.gather(*(client.get("https://example.test/mcp") for _ in range(5))),
|
||||
timeout=2,
|
||||
)
|
||||
assert [response.status_code for response in responses] == [
|
||||
200 if replacement == "new" else 401
|
||||
] * 5
|
||||
assert old_requests == 5
|
||||
refresh.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("replacement", [None, "", "old", "new", "raises", "disabled"])
|
||||
@pytest.mark.asyncio
|
||||
async def test_unrecovered_401_still_retires_connection(
|
||||
replacement: str | None, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
refresh = AsyncMock(return_value=replacement)
|
||||
if replacement == "raises":
|
||||
refresh.side_effect = ValueError("private-token-value")
|
||||
auth = None if replacement == "disabled" else RefreshingBearerAuth("old", refresh)
|
||||
seen: list[str] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
seen.append(request.headers["Authorization"])
|
||||
return httpx.Response(401)
|
||||
|
||||
async with _client(auth, handler) as client:
|
||||
server = AsyncMock()
|
||||
|
||||
async def call_tool(*_args: Any, **_kwargs: Any) -> None:
|
||||
response = await client.get("https://example.test/mcp")
|
||||
response.raise_for_status()
|
||||
|
||||
server.call_tool = call_tool
|
||||
session = mcp_session.SupervisedMcpSession.adopt(server, name="refresh")
|
||||
result = await session.dispatch("read", {}, label="refresh_read")
|
||||
assert result["success"] is False
|
||||
assert session.is_dead
|
||||
await session.aclose()
|
||||
assert seen == (["Bearer old", "Bearer new"] if replacement == "new" else ["Bearer old"])
|
||||
assert refresh.await_count == (0 if replacement == "disabled" else 1)
|
||||
assert "private-token-value" not in caplog.text
|
||||
if replacement == "raises":
|
||||
assert "ValueError" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_401_does_not_refresh() -> None:
|
||||
refresh = AsyncMock(return_value="new")
|
||||
auth = RefreshingBearerAuth("old", refresh)
|
||||
async with _client(auth, lambda _: httpx.Response(403)) as client:
|
||||
assert (await client.get("https://example.test/mcp")).status_code == 403
|
||||
refresh.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_registry_passes_refresher_to_session(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
refresh = AsyncMock(return_value="new")
|
||||
registry = McpRegistry()
|
||||
entry = registry.register(McpConnectionRequest(_config(), refresh_token=refresh))
|
||||
server = AsyncMock()
|
||||
captured: list[RefreshingBearerAuth] = []
|
||||
|
||||
def build(_config: McpConnectionConfig, auth: RefreshingBearerAuth) -> Any:
|
||||
captured.append(auth)
|
||||
return mcp_client.BuiltMcpServer(server, None)
|
||||
|
||||
monkeypatch.setattr(mcp_client, "_build_server", build)
|
||||
session = await entry.ensure_connected()
|
||||
assert entry.refresh_token is refresh
|
||||
assert captured == [session._bearer_auth]
|
||||
await session._open()
|
||||
assert captured == [session._bearer_auth, session._bearer_auth]
|
||||
await session.aclose()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refreshed_token_can_refresh_again_after_expiry() -> None:
|
||||
refresh = AsyncMock(side_effect=["new", "newer"])
|
||||
auth = RefreshingBearerAuth("old", refresh)
|
||||
valid = "new"
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200 if request.headers["Authorization"] == f"Bearer {valid}" else 401)
|
||||
|
||||
async with _client(auth, handler) as client:
|
||||
assert (await client.get("https://example.test/mcp")).status_code == 200
|
||||
valid = "newer"
|
||||
assert (await client.get("https://example.test/mcp")).status_code == 200
|
||||
assert refresh.await_count == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize("with_refresher", [True, False])
|
||||
@pytest.mark.asyncio
|
||||
async def test_eager_attachment_preserves_bearer_refresh(
|
||||
monkeypatch: pytest.MonkeyPatch, with_refresher: bool
|
||||
) -> None:
|
||||
refresh = AsyncMock(return_value="new")
|
||||
original_build = mcp_client._build_server
|
||||
valid = "old"
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200 if request.headers["Authorization"] == f"Bearer {valid}" else 401)
|
||||
|
||||
def build(config: McpConnectionConfig, auth: RefreshingBearerAuth | None = None) -> Any:
|
||||
built = original_build(config, auth)
|
||||
params = cast("Any", built.server).params
|
||||
client = params["httpx_client_factory"](headers=params["headers"])
|
||||
client._transport = httpx.MockTransport(handler)
|
||||
server = AsyncMock()
|
||||
server.use_structured_content = False
|
||||
server.list_tools.return_value = [Tool(name="read", inputSchema={"type": "object"})]
|
||||
server.cleanup.side_effect = client.aclose
|
||||
|
||||
async def call_tool(*_args: Any, **_kwargs: Any) -> CallToolResult:
|
||||
response = await client.post("https://example.test/mcp")
|
||||
response.raise_for_status()
|
||||
return CallToolResult(content=[TextContent(type="text", text="ok")])
|
||||
|
||||
server.call_tool = call_tool
|
||||
return mcp_client.BuiltMcpServer(server, built.recorder)
|
||||
|
||||
monkeypatch.setattr(mcp_client, "_build_server", build)
|
||||
registry = McpRegistry()
|
||||
try:
|
||||
connections = await mcp_client.attach_mcp_requests(
|
||||
[McpConnectionRequest(_config(), refresh_token=refresh if with_refresher else None)],
|
||||
registry,
|
||||
)
|
||||
assert len(connections) == 1
|
||||
entry = registry.get("refresh")
|
||||
assert entry is not None
|
||||
assert entry.refresh_token is (refresh if with_refresher else None)
|
||||
session = await entry.ensure_connected()
|
||||
assert await session.dispatch("read", {}, label="refresh_read") == {
|
||||
"type": "text",
|
||||
"text": "ok",
|
||||
}
|
||||
valid = "new"
|
||||
result = await session.dispatch("read", {}, label="refresh_read")
|
||||
if with_refresher:
|
||||
assert result == {"type": "text", "text": "ok"}
|
||||
assert not session.is_dead
|
||||
refresh.assert_awaited_once()
|
||||
else:
|
||||
assert result["success"] is False
|
||||
assert session.is_dead
|
||||
finally:
|
||||
await registry.close()
|
||||
@@ -111,6 +111,15 @@ async def test_retry_after_parses_seconds_and_http_date() -> None:
|
||||
assert 0 <= retry_after <= 20
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_successful_response_clears_previous_auth_failure() -> None:
|
||||
recorder = HttpStatusRecorder()
|
||||
await recorder(_http_error(401).response)
|
||||
request = httpx.Request("POST", "https://provider.example/tools")
|
||||
await recorder(httpx.Response(200, request=request))
|
||||
assert recorder.take() is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_recorder_only_keeps_non_sensitive_request_metadata() -> None:
|
||||
recorder = HttpStatusRecorder()
|
||||
|
||||
@@ -1,14 +1,50 @@
|
||||
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"
|
||||
@@ -60,6 +96,112 @@ 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
|
||||
|
||||
@@ -336,7 +336,7 @@ async def test_dependency_report_sets_class_and_metadata(report_state: ReportSta
|
||||
advisory_cvss=7.2,
|
||||
technical_analysis=None,
|
||||
fix_effort="trivial",
|
||||
reachability="imported",
|
||||
reachability="not_reachable",
|
||||
reachability_evidence=_DEP_EVIDENCE,
|
||||
contextual_cvss_breakdown=_DEP_CONTEXT,
|
||||
contextual_cvss_reasoning=_DEP_REASONING,
|
||||
@@ -357,7 +357,7 @@ async def test_dependency_report_sets_class_and_metadata(report_state: ReportSta
|
||||
"package_ecosystem": "npm",
|
||||
"manifest_path": "package-lock.json",
|
||||
"fixed_version": "4.17.21",
|
||||
"reachability": "imported",
|
||||
"reachability": "not_reachable",
|
||||
"reachability_evidence": _DEP_EVIDENCE,
|
||||
"contextual_cvss_breakdown": _DEP_CONTEXT,
|
||||
"contextual_cvss_score": pytest.approx(7.5, abs=0.05),
|
||||
@@ -386,7 +386,7 @@ async def test_dependency_report_records_transitive_chain(report_state: ReportSt
|
||||
fix_effort="trivial",
|
||||
introduced_by="[email protected]",
|
||||
dependency_path="[email protected] > [email protected] > [email protected]",
|
||||
reachability="imported",
|
||||
reachability="not_reachable",
|
||||
reachability_evidence=_DEP_EVIDENCE,
|
||||
contextual_cvss_breakdown=_DEP_CONTEXT,
|
||||
contextual_cvss_reasoning=_DEP_REASONING,
|
||||
@@ -427,7 +427,7 @@ async def test_dependency_report_persists_code_locations(report_state: ReportSta
|
||||
advisory_cvss=7.2,
|
||||
technical_analysis=None,
|
||||
fix_effort="trivial",
|
||||
reachability="vulnerable_symbol_used",
|
||||
reachability="reachable",
|
||||
reachability_evidence=_DEP_EVIDENCE,
|
||||
code_locations=[
|
||||
{
|
||||
@@ -468,7 +468,7 @@ async def test_dependency_report_fix_locations_require_verification(
|
||||
advisory_cvss=7.2,
|
||||
technical_analysis=None,
|
||||
fix_effort="trivial",
|
||||
reachability="imported",
|
||||
reachability="not_reachable",
|
||||
reachability_evidence=_DEP_EVIDENCE,
|
||||
code_locations=[
|
||||
{
|
||||
@@ -509,7 +509,7 @@ async def test_dependency_report_fix_locations_persist_with_verification(
|
||||
advisory_cvss=7.2,
|
||||
technical_analysis=None,
|
||||
fix_effort="trivial",
|
||||
reachability="imported",
|
||||
reachability="not_reachable",
|
||||
reachability_evidence=_DEP_EVIDENCE,
|
||||
code_locations=[
|
||||
{
|
||||
@@ -556,7 +556,7 @@ async def test_dependency_report_rejects_unusable_locations(
|
||||
advisory_cvss=7.2,
|
||||
technical_analysis=None,
|
||||
fix_effort="trivial",
|
||||
reachability="imported",
|
||||
reachability="not_reachable",
|
||||
reachability_evidence=_DEP_EVIDENCE,
|
||||
code_locations=[{"file": "/abs/path.ts", "snippet": "require('lodash')"}],
|
||||
contextual_cvss_breakdown=_DEP_CONTEXT,
|
||||
@@ -587,7 +587,7 @@ async def test_dependency_report_omits_blank_chain_fields(report_state: ReportSt
|
||||
fix_effort="trivial",
|
||||
introduced_by=" ",
|
||||
dependency_path=None,
|
||||
reachability="imported",
|
||||
reachability="not_reachable",
|
||||
reachability_evidence=_DEP_EVIDENCE,
|
||||
contextual_cvss_breakdown=_DEP_CONTEXT,
|
||||
contextual_cvss_reasoning=_DEP_REASONING,
|
||||
@@ -618,7 +618,7 @@ async def test_dependency_report_with_no_contextual_impact_is_info(
|
||||
advisory_cvss=0.0,
|
||||
technical_analysis=None,
|
||||
fix_effort="low",
|
||||
reachability="not_imported",
|
||||
reachability="not_applicable",
|
||||
reachability_evidence="No file imports the package.",
|
||||
contextual_cvss_breakdown={**_DEP_CONTEXT, "availability": "N"},
|
||||
contextual_cvss_reasoning="No application code imports the package.",
|
||||
@@ -649,7 +649,7 @@ async def test_dependency_report_records_reachability(report_state: ReportState)
|
||||
advisory_cvss=7.2,
|
||||
technical_analysis=None,
|
||||
fix_effort="low",
|
||||
reachability="vulnerable_symbol_used",
|
||||
reachability="reachable",
|
||||
reachability_evidence="src/render.ts:14 calls `_.template()`.",
|
||||
contextual_cvss_breakdown=_DEP_CONTEXT,
|
||||
contextual_cvss_reasoning=_DEP_REASONING,
|
||||
@@ -657,7 +657,7 @@ async def test_dependency_report_records_reachability(report_state: ReportState)
|
||||
|
||||
assert result["success"] is True
|
||||
report = report_state.vulnerability_reports[0]
|
||||
assert report["dependency_metadata"]["reachability"] == "vulnerable_symbol_used"
|
||||
assert report["dependency_metadata"]["reachability"] == "reachable"
|
||||
assert (
|
||||
report["dependency_metadata"]["reachability_evidence"]
|
||||
== "src/render.ts:14 calls `_.template()`."
|
||||
@@ -688,7 +688,7 @@ async def test_dependency_report_rejects_reachability_without_evidence(
|
||||
advisory_cvss=5.0,
|
||||
technical_analysis=None,
|
||||
fix_effort="low",
|
||||
reachability="not_imported",
|
||||
reachability="not_applicable",
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
@@ -725,7 +725,7 @@ async def test_dependency_report_rejects_unknown_reachability_level(
|
||||
assert not report_state.vulnerability_reports
|
||||
|
||||
|
||||
async def test_dependency_report_records_unknown_reachability(report_state: ReportState) -> None:
|
||||
async def test_dependency_report_requires_a_reachability_verdict(report_state: ReportState) -> None:
|
||||
result = await _do_create_dependency(
|
||||
title="CVE-2024-0001 in sample 1.0.0",
|
||||
description="Published advisory affects the pinned version.",
|
||||
@@ -748,10 +748,9 @@ async def test_dependency_report_records_unknown_reachability(report_state: Repo
|
||||
contextual_cvss_reasoning=_DEP_REASONING,
|
||||
)
|
||||
|
||||
assert result["success"] is True, result
|
||||
metadata = report_state.vulnerability_reports[0]["dependency_metadata"]
|
||||
assert metadata["reachability"] == "unknown"
|
||||
assert metadata["reachability_evidence"] == "Grep for the package found no import."
|
||||
assert result["success"] is False, result
|
||||
assert any("reachability is required" in e for e in result["errors"])
|
||||
assert not report_state.vulnerability_reports
|
||||
|
||||
|
||||
async def test_dependency_report_requires_advisory_cvss(report_state: ReportState) -> None:
|
||||
@@ -828,7 +827,7 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
|
||||
advisory_cvss=0.0,
|
||||
technical_analysis=None,
|
||||
fix_effort="low",
|
||||
reachability="imported",
|
||||
reachability="not_reachable",
|
||||
reachability_evidence=_DEP_EVIDENCE,
|
||||
contextual_cvss_breakdown=_DEP_CONTEXT,
|
||||
contextual_cvss_reasoning=_DEP_REASONING,
|
||||
@@ -847,7 +846,7 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
|
||||
"package_ecosystem": "npm",
|
||||
"manifest_path": "package-lock.json",
|
||||
"fixed_version": "1.0.1",
|
||||
"reachability": "imported",
|
||||
"reachability": "not_reachable",
|
||||
"reachability_evidence": _DEP_EVIDENCE,
|
||||
"contextual_cvss_breakdown": _DEP_CONTEXT,
|
||||
"contextual_cvss_score": pytest.approx(7.5, abs=0.05),
|
||||
@@ -1381,7 +1380,7 @@ async def test_dependency_report_computes_contextual_cvss(
|
||||
cwe="CWE-94",
|
||||
fix_effort="trivial",
|
||||
manifest_path="package-lock.json",
|
||||
reachability="vulnerable_symbol_used",
|
||||
reachability="reachable",
|
||||
reachability_evidence="scripts/import.py:88 calls `_.template()`.",
|
||||
contextual_cvss_breakdown=_CONTEXTUAL_BREAKDOWN,
|
||||
contextual_cvss_reasoning="Only scripts/import.py reaches the sink.",
|
||||
@@ -1421,7 +1420,7 @@ async def test_dependency_report_requires_contextual_breakdown(
|
||||
cwe="CWE-94",
|
||||
fix_effort="trivial",
|
||||
manifest_path="package-lock.json",
|
||||
reachability="imported",
|
||||
reachability="not_reachable",
|
||||
reachability_evidence=_DEP_EVIDENCE,
|
||||
)
|
||||
assert result["success"] is False
|
||||
@@ -2408,57 +2407,3 @@ 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,10 +371,3 @@ 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