mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 09:26:39 +02:00
Compare commits
37
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4c6df52871 | ||
|
|
f8bab65552 | ||
|
|
d44ca88a18 | ||
|
|
91d9a84716 | ||
|
|
e69c8f6633 | ||
|
|
81a8b2139b | ||
|
|
b959d528a2 | ||
|
|
daf39a2305 | ||
|
|
5304baa424 | ||
|
|
993fd41f32 | ||
|
|
b7a1259593 | ||
|
|
48521deb62 | ||
|
|
d6cefc176a | ||
|
|
a87bfb4881 | ||
|
|
24279e3279 | ||
|
|
4537f33f11 | ||
|
|
ee779987d3 | ||
|
|
a5f28c6b4b | ||
|
|
4b46a748e4 | ||
|
|
205e0b3707 | ||
|
|
c13960ae01 | ||
|
|
22d327d21f | ||
|
|
f528a6d265 | ||
|
|
054725ccb6 | ||
|
|
882664f70b | ||
|
|
e1abac0f0f | ||
|
|
df4bfafcd3 | ||
|
|
5c6cbe0884 | ||
|
|
dd29d99b85 | ||
|
|
9f6d0b106b | ||
|
|
e53b0bd11f | ||
|
|
0bf992ecbf | ||
|
|
f7fa54c12d | ||
|
|
b9994e2e0e | ||
|
|
0fb005c73f | ||
|
|
e1940769de | ||
|
|
9f278b9a5c |
@@ -16,7 +16,7 @@ jobs:
|
||||
target: macos-arm64
|
||||
- os: macos-15-intel
|
||||
target: macos-x86_64
|
||||
- os: ubuntu-latest
|
||||
- os: ubuntu-22.04
|
||||
target: linux-x86_64
|
||||
- os: windows-latest
|
||||
target: windows-x86_64
|
||||
|
||||
@@ -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" 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&utm_medium=badge&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>
|
||||
|
||||
</div>
|
||||
|
||||
@@ -169,6 +169,9 @@ strix --target https://your-app.com --instruction "Perform authenticated testing
|
||||
# Multi-target testing (source code + deployed app)
|
||||
strix -t https://github.com/org/app -t https://your-app.com
|
||||
|
||||
# Targets from a file, one target per non-empty, non-comment line
|
||||
strix --target-list ./targets.txt
|
||||
|
||||
# White-box source-aware scan (local repository)
|
||||
strix --target ./app-directory --scan-mode standard
|
||||
|
||||
|
||||
@@ -9,10 +9,24 @@ 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 &
|
||||
|
||||
|
||||
@@ -3,6 +3,14 @@ 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
|
||||
|
||||
@@ -62,6 +62,9 @@ strix --target https://your-app.com
|
||||
|
||||
# Multiple targets (white-box testing)
|
||||
strix -t https://github.com/org/repo -t https://your-app.com
|
||||
|
||||
# Targets from a file, one target per non-empty, non-comment line
|
||||
strix --target-list ./targets.txt
|
||||
```
|
||||
|
||||
## Next Steps
|
||||
|
||||
+10
-3
@@ -6,13 +6,17 @@ description: "Command-line options for Strix"
|
||||
## Basic Usage
|
||||
|
||||
```bash
|
||||
strix --target <target> [options]
|
||||
strix (--target <target> | --target-list <path> | --mount <path>) [options]
|
||||
```
|
||||
|
||||
## Options
|
||||
|
||||
<ParamField path="--target, -t" type="string" required>
|
||||
Target to test. Accepts URLs, repositories, local directories, domains, or IP addresses. Can be specified multiple times.
|
||||
<ParamField path="--target, -t" type="string">
|
||||
Target to test. Accepts URLs, repositories, local directories, domains, or IP addresses. Can be specified multiple times. Fresh runs require at least one target source: `--target`, `--target-list`, or `--mount`.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="--target-list" type="string">
|
||||
Path to a file containing targets, one per non-empty, non-comment line. Lines starting with `#` are ignored. Can be specified multiple times and combined with `--target`.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="--mount" type="string">
|
||||
@@ -101,6 +105,9 @@ strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main
|
||||
# Multi-target white-box testing
|
||||
strix -t https://github.com/org/app -t https://staging.example.com
|
||||
|
||||
# Targets from a file
|
||||
strix --target-list ./targets.txt
|
||||
|
||||
# Large local repository — bind-mount instead of copying it in
|
||||
strix --mount ./huge-monorepo
|
||||
```
|
||||
|
||||
+7
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "strix-agent"
|
||||
version = "1.0.4"
|
||||
version = "1.1.0"
|
||||
description = "Open-source AI Hackers for your apps"
|
||||
readme = "README.md"
|
||||
license = "Apache-2.0"
|
||||
@@ -34,6 +34,8 @@ 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",
|
||||
@@ -44,6 +46,10 @@ 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"
|
||||
|
||||
|
||||
+75
-14
@@ -41,7 +41,7 @@ from strix.tools.proxy.tools import (
|
||||
view_request,
|
||||
view_sitemap_entry,
|
||||
)
|
||||
from strix.tools.reporting.tool import create_vulnerability_report
|
||||
from strix.tools.reporting.tool import create_dependency_report, 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
|
||||
from collections.abc import Awaitable, Callable, Sequence
|
||||
|
||||
from agents import RunContextWrapper
|
||||
from agents.tool import FunctionToolResult
|
||||
@@ -209,6 +209,13 @@ 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:
|
||||
@@ -335,6 +342,7 @@ _BASE_TOOLS: tuple[Tool, ...] = (
|
||||
delete_note,
|
||||
web_search,
|
||||
create_vulnerability_report,
|
||||
create_dependency_report,
|
||||
list_requests,
|
||||
view_request,
|
||||
repeat_request,
|
||||
@@ -349,6 +357,48 @@ _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",
|
||||
@@ -359,26 +409,37 @@ 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.
|
||||
"""
|
||||
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,
|
||||
)
|
||||
|
||||
if is_root:
|
||||
tools: list[Tool] = [*_BASE_TOOLS, finish_scan]
|
||||
if instructions_override is not None:
|
||||
instructions = instructions_override
|
||||
else:
|
||||
tools = [*_BASE_TOOLS, agent_finish]
|
||||
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]
|
||||
else:
|
||||
tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
|
||||
_ensure_unique_tool_names(tools)
|
||||
|
||||
logger.info(
|
||||
"Built %s agent '%s' (skills=%d, tools=%d, scan_mode=%s, whitebox=%s)",
|
||||
|
||||
@@ -7,7 +7,7 @@ from typing import Any
|
||||
|
||||
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
||||
|
||||
from strix.skills import get_available_skills, load_skills
|
||||
from strix.skills import get_available_skills, load_skills, skill_search_dirs
|
||||
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)
|
||||
skills_dir = get_strix_resource_path("skills")
|
||||
loader_dirs = [prompt_dir, *skill_search_dirs()]
|
||||
env = Environment(
|
||||
loader=FileSystemLoader([prompt_dir, skills_dir]),
|
||||
loader=FileSystemLoader(loader_dirs),
|
||||
autoescape=select_autoescape(
|
||||
enabled_extensions=(),
|
||||
default_for_string=False,
|
||||
|
||||
@@ -190,7 +190,7 @@ VALIDATION REQUIREMENTS:
|
||||
- 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 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 (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
|
||||
- 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>
|
||||
|
||||
@@ -59,12 +59,49 @@ DEFAULT_MODEL_RETRY = ModelRetrySettings(
|
||||
),
|
||||
)
|
||||
|
||||
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)
|
||||
@@ -111,6 +148,29 @@ 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
|
||||
|
||||
@@ -156,6 +216,78 @@ 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
|
||||
|
||||
|
||||
@@ -36,6 +36,10 @@ 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")
|
||||
|
||||
|
||||
|
||||
+4
-1
@@ -28,7 +28,10 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
|
||||
|
||||
def __init__(self, *, model: str, max_budget_usd: float | None = None) -> None:
|
||||
import math
|
||||
if max_budget_usd is not None and (not math.isfinite(max_budget_usd) or max_budget_usd <= 0):
|
||||
|
||||
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
|
||||
|
||||
+17
-1
@@ -8,7 +8,11 @@ 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, model_supports_reasoning
|
||||
from strix.config.models import (
|
||||
DEFAULT_MODEL_RETRY,
|
||||
is_known_openai_bare_model,
|
||||
model_supports_reasoning,
|
||||
)
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -18,6 +22,15 @@ 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 {}
|
||||
@@ -111,6 +124,7 @@ def make_model_settings(
|
||||
reasoning_effort: ReasoningEffort | None,
|
||||
*,
|
||||
model_name: str,
|
||||
force_required_tool_choice: bool = False,
|
||||
) -> ModelSettings:
|
||||
model_settings = ModelSettings(
|
||||
parallel_tool_calls=False,
|
||||
@@ -125,6 +139,8 @@ 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
|
||||
|
||||
|
||||
|
||||
+69
-2
@@ -14,6 +14,7 @@ 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,
|
||||
@@ -51,6 +52,52 @@ 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],
|
||||
@@ -64,8 +111,17 @@ 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."""
|
||||
"""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.
|
||||
"""
|
||||
if scan_id is None:
|
||||
scan_id = f"scan-{uuid.uuid4().hex[:8]}"
|
||||
|
||||
@@ -158,6 +214,7 @@ 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,
|
||||
)
|
||||
run_config = RunConfig(
|
||||
model=resolved_model,
|
||||
@@ -169,6 +226,15 @@ 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",
|
||||
@@ -178,7 +244,8 @@ async def run_strix_scan(
|
||||
is_whitebox=is_whitebox,
|
||||
interactive=interactive,
|
||||
chat_completions_tools=chat_completions_tools,
|
||||
system_prompt_context=scope_context,
|
||||
system_prompt_context=root_context,
|
||||
instructions_override=root_instructions,
|
||||
)
|
||||
|
||||
if not is_resume:
|
||||
|
||||
+112
-9
@@ -23,9 +23,11 @@ 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
|
||||
@@ -44,6 +46,7 @@ from strix.interface.utils import (
|
||||
infer_target_type,
|
||||
is_whitebox_scan,
|
||||
process_pull_line,
|
||||
read_target_list_file,
|
||||
resolve_diff_scope_context,
|
||||
rewrite_localhost_targets,
|
||||
validate_config_file,
|
||||
@@ -55,6 +58,16 @@ 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
|
||||
@@ -213,10 +226,51 @@ def check_docker_installed() -> None:
|
||||
logger.debug("Docker CLI present")
|
||||
|
||||
|
||||
async def warm_up_llm() -> None:
|
||||
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:
|
||||
console = Console()
|
||||
logger.info("Warming up LLM connection")
|
||||
|
||||
raw_model = ""
|
||||
try:
|
||||
settings = load_settings()
|
||||
configure_sdk_model_defaults(settings)
|
||||
@@ -254,6 +308,32 @@ async def warm_up_llm() -> 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(
|
||||
@@ -279,6 +359,9 @@ async def warm_up_llm() -> 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(
|
||||
@@ -310,6 +393,7 @@ 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
|
||||
@@ -344,6 +428,9 @@ Examples:
|
||||
strix --target https://github.com/user/repo --target https://example.com
|
||||
strix --target ./my-project --target https://staging.example.com --target https://prod.example.com
|
||||
|
||||
# Targets from a file, one target per non-empty, non-comment line
|
||||
strix --target-list ./targets.txt
|
||||
|
||||
# Custom instructions (inline)
|
||||
strix --target example.com --instruction "Focus on authentication vulnerabilities"
|
||||
|
||||
@@ -367,7 +454,15 @@ Examples:
|
||||
action="append",
|
||||
help="Target to test (URL, repository, local directory path, domain name, or IP address). "
|
||||
"Can be specified multiple times for multi-target scans. "
|
||||
"Required for fresh runs; loaded from disk when ``--resume`` is set.",
|
||||
"Fresh runs require at least one of --target, --target-list, or --mount.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--target-list",
|
||||
type=str,
|
||||
action="append",
|
||||
metavar="PATH",
|
||||
help="Path to a file containing targets, one per non-empty, non-comment line. "
|
||||
"Can be specified multiple times and combined with --target.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--mount",
|
||||
@@ -488,10 +583,11 @@ Examples:
|
||||
args.user_explicit_instruction = args.instruction if args.resume else None
|
||||
|
||||
if args.resume:
|
||||
if args.target or args.mount:
|
||||
if args.target or args.target_list or args.mount:
|
||||
parser.error(
|
||||
"Cannot combine --resume with --target/--mount. --resume picks up where "
|
||||
"the prior run left off, including the original target list."
|
||||
"Cannot combine --resume with --target/--target-list/--mount. "
|
||||
"--resume picks up where the prior run left off, including the "
|
||||
"original target list."
|
||||
)
|
||||
_load_resume_state(args, parser)
|
||||
agents_path = runtime_state_dir(run_dir_for(args.resume)) / "agents.json"
|
||||
@@ -503,13 +599,20 @@ Examples:
|
||||
f"or remove --resume to start over with the same targets."
|
||||
)
|
||||
else:
|
||||
if not args.target and not args.mount:
|
||||
if not args.target and not args.target_list and not args.mount:
|
||||
parser.error(
|
||||
"the following arguments are required: -t/--target or --mount "
|
||||
"the following arguments are required: -t/--target, --target-list, or --mount "
|
||||
"(or use --resume <run_name> to continue a prior scan)"
|
||||
)
|
||||
args.targets_info = []
|
||||
for target in args.target or []:
|
||||
targets = list(args.target or [])
|
||||
for target_list_path in args.target_list or []:
|
||||
try:
|
||||
targets.extend(read_target_list_file(target_list_path))
|
||||
except ValueError as e:
|
||||
parser.error(str(e))
|
||||
|
||||
for target in targets:
|
||||
try:
|
||||
target_type, target_dict = infer_target_type(target)
|
||||
|
||||
@@ -752,7 +855,7 @@ def main() -> None:
|
||||
pull_docker_image()
|
||||
|
||||
validate_environment()
|
||||
asyncio.run(warm_up_llm())
|
||||
asyncio.run(warm_up_llm(show_model_warning=args.non_interactive))
|
||||
|
||||
persist_current()
|
||||
|
||||
|
||||
@@ -31,6 +31,7 @@ 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
|
||||
@@ -116,9 +117,16 @@ 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)
|
||||
@@ -128,7 +136,7 @@ class SplashScreen(Static): # type: ignore[misc]
|
||||
yield panel_static
|
||||
|
||||
def on_mount(self) -> None:
|
||||
self._animation_timer = self.set_interval(0.05, self._animate_start_line)
|
||||
self._animation_timer = self.set_interval(0.1, self._animate_start_line)
|
||||
|
||||
def on_unmount(self) -> None:
|
||||
if self._animation_timer is not None:
|
||||
@@ -145,7 +153,7 @@ class SplashScreen(Static): # type: ignore[misc]
|
||||
self._panel_static.update(panel)
|
||||
|
||||
def _build_panel(self, start_line: Text) -> Panel:
|
||||
content = Group(
|
||||
rows = [
|
||||
Align.center(Text(self.BANNER.strip("\n"), style=self.PRIMARY_GREEN, justify="center")),
|
||||
Align.center(Text(" ")),
|
||||
Align.center(self._build_welcome_text()),
|
||||
@@ -155,9 +163,26 @@ 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(content, border_style=self.PRIMARY_GREEN, padding=(1, 6))
|
||||
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
|
||||
|
||||
def _build_url_text(self) -> Text:
|
||||
return Text("strix.ai", style=Style(color=self.PRIMARY_GREEN, bold=True))
|
||||
@@ -371,6 +396,19 @@ 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")
|
||||
@@ -389,6 +427,18 @@ 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 = []
|
||||
@@ -434,6 +484,13 @@ 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")
|
||||
@@ -455,6 +512,13 @@ 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:
|
||||
@@ -476,14 +540,27 @@ 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:
|
||||
@@ -514,6 +591,9 @@ 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"):
|
||||
@@ -552,6 +632,9 @@ 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)
|
||||
|
||||
@@ -729,6 +812,7 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
"#86efac", # Brightest
|
||||
]
|
||||
self._dot_animation_timer: Any | None = None
|
||||
self._pending_scroll_end = False
|
||||
|
||||
self._setup_cleanup_handlers()
|
||||
|
||||
@@ -872,7 +956,7 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
|
||||
self._start_scan_thread()
|
||||
|
||||
self.set_interval(0.35, self._update_ui)
|
||||
self.set_interval(0.5, self._update_ui)
|
||||
|
||||
def _update_ui(self) -> None:
|
||||
if self.show_splash:
|
||||
@@ -1018,8 +1102,16 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
self._safe_widget_operation(chat_display.update, content)
|
||||
chat_display.set_classes(css_class)
|
||||
|
||||
if is_at_bottom:
|
||||
self.call_later(chat_history.scroll_end, animate=False)
|
||||
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)
|
||||
|
||||
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
|
||||
from typing import Any, ClassVar
|
||||
|
||||
from pygments.lexers import get_lexer_by_name, guess_lexer
|
||||
from pygments.styles import get_style_by_name
|
||||
@@ -161,6 +161,8 @@ 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:
|
||||
@@ -168,4 +170,11 @@ class AgentMessageRenderer:
|
||||
cleaned = _BLANK_LINE_RUNS.sub("\n\n", content).strip()
|
||||
if not cleaned:
|
||||
return Text()
|
||||
return _apply_markdown_styles(cleaned)
|
||||
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()
|
||||
|
||||
@@ -256,3 +256,176 @@ 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 = Text.from_ansi(output).plain.translate(_CONTROL_BYTES_TO_DROP)
|
||||
cleaned: str = Text.from_ansi(output).plain.translate(_CONTROL_BYTES_TO_DROP)
|
||||
for pattern in STRIP_PATTERNS:
|
||||
cleaned = re.sub(pattern, "", cleaned, flags=re.MULTILINE)
|
||||
|
||||
|
||||
@@ -1131,6 +1131,34 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR09
|
||||
)
|
||||
|
||||
|
||||
def read_target_list_file(path_str: str) -> list[str]:
|
||||
"""Read scan targets from a file, one target per non-empty, non-comment line."""
|
||||
if not path_str or not path_str.strip():
|
||||
raise ValueError("--target-list path must not be empty.")
|
||||
|
||||
path = Path(path_str).expanduser()
|
||||
if not path.is_file():
|
||||
raise ValueError(f"Target list file '{path_str}' is not an existing file.")
|
||||
|
||||
try:
|
||||
targets = [
|
||||
target
|
||||
for line in path.read_text(encoding="utf-8").splitlines()
|
||||
if (target := line.strip()) and not target.startswith("#")
|
||||
]
|
||||
except UnicodeDecodeError as e:
|
||||
raise ValueError(
|
||||
f"Target list file '{path_str}' must be valid UTF-8 text: {e!s}"
|
||||
) from e
|
||||
except OSError as e:
|
||||
raise ValueError(f"Failed to read target list file '{path_str}': {e!s}") from e
|
||||
|
||||
targets = [target for target in targets if target]
|
||||
if not targets:
|
||||
raise ValueError(f"Target list file '{path_str}' is empty.")
|
||||
return targets
|
||||
|
||||
|
||||
def sanitize_name(name: str) -> str:
|
||||
sanitized = re.sub(r"[^A-Za-z0-9._-]", "-", name.strip())
|
||||
return sanitized or "target"
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from agents.model_settings import ModelSettings
|
||||
@@ -51,6 +52,11 @@ 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
|
||||
@@ -101,6 +107,8 @@ def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
|
||||
"poc_description",
|
||||
"endpoint",
|
||||
"method",
|
||||
"cve",
|
||||
"dependency_metadata",
|
||||
]
|
||||
|
||||
cleaned = {}
|
||||
@@ -114,6 +122,112 @@ 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("```"):
|
||||
@@ -165,6 +279,10 @@ 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
|
||||
|
||||
+53
-25
@@ -135,6 +135,10 @@ 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
|
||||
@@ -212,6 +216,9 @@ 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,
|
||||
@@ -219,6 +226,9 @@ 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:
|
||||
@@ -245,6 +255,12 @@ 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:
|
||||
@@ -259,6 +275,11 @@ 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:
|
||||
@@ -266,8 +287,8 @@ class ReportState:
|
||||
|
||||
self.vulnerability_reports.append(report)
|
||||
logger.info(f"Added vulnerability report: {report_id} - {title}")
|
||||
posthog.finding(severity)
|
||||
scarf.finding(severity)
|
||||
posthog.finding(severity, cwe=cwe, is_cve=bool(cve))
|
||||
scarf.finding(severity, cwe=cwe, is_cve=bool(cve))
|
||||
|
||||
if self.vulnerability_found_callback:
|
||||
self.vulnerability_found_callback(report)
|
||||
@@ -438,30 +459,37 @@ class ReportState:
|
||||
targets = self.run_record.get("targets_info") or []
|
||||
if not isinstance(targets, list):
|
||||
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
|
||||
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
|
||||
|
||||
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
|
||||
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
|
||||
|
||||
def _sync_llm_usage_record(self) -> None:
|
||||
self.run_record["llm_usage"] = self._build_llm_usage_record()
|
||||
|
||||
@@ -124,8 +124,13 @@ 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")),
|
||||
@@ -134,6 +139,8 @@ 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}")
|
||||
@@ -143,6 +150,11 @@ 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"]))
|
||||
@@ -194,4 +206,9 @@ 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)
|
||||
|
||||
+131
-3
@@ -2,7 +2,9 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
@@ -16,6 +18,128 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
SandboxBackend = Callable[..., Awaitable[tuple[Any, Any]]]
|
||||
|
||||
_DEFAULT_START_ATTEMPTS = 3
|
||||
_START_BACKOFF_SECONDS = 2.0
|
||||
_TRANSIENT_TIMEOUT_NAMES = {
|
||||
"ConnectTimeout",
|
||||
"PoolTimeout",
|
||||
"ReadTimeout",
|
||||
"TimeoutError",
|
||||
"TimeoutException",
|
||||
"WriteTimeout",
|
||||
}
|
||||
_TRANSIENT_CONNECTION_NAMES = {
|
||||
"ConnectError",
|
||||
"ConnectionError",
|
||||
"ConnectionResetError",
|
||||
"ReadError",
|
||||
"WriteError",
|
||||
}
|
||||
|
||||
|
||||
def _start_attempts() -> int:
|
||||
raw = os.environ.get("STRIX_SANDBOX_START_ATTEMPTS")
|
||||
if raw is None:
|
||||
return _DEFAULT_START_ATTEMPTS
|
||||
try:
|
||||
attempts = int(raw)
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
"Invalid STRIX_SANDBOX_START_ATTEMPTS=%r; using %d",
|
||||
raw,
|
||||
_DEFAULT_START_ATTEMPTS,
|
||||
)
|
||||
return _DEFAULT_START_ATTEMPTS
|
||||
if attempts < 1:
|
||||
logger.warning(
|
||||
"STRIX_SANDBOX_START_ATTEMPTS must be positive; using %d",
|
||||
_DEFAULT_START_ATTEMPTS,
|
||||
)
|
||||
return _DEFAULT_START_ATTEMPTS
|
||||
return attempts
|
||||
|
||||
|
||||
def _exception_chain(error: BaseException) -> list[BaseException]:
|
||||
chain: list[BaseException] = []
|
||||
pending: list[BaseException | None] = [error]
|
||||
seen: set[int] = set()
|
||||
while pending:
|
||||
current = pending.pop()
|
||||
if current is None or id(current) in seen:
|
||||
continue
|
||||
seen.add(id(current))
|
||||
chain.append(current)
|
||||
pending.extend(
|
||||
(
|
||||
current.__cause__,
|
||||
current.__context__,
|
||||
getattr(current, "cause", None),
|
||||
)
|
||||
)
|
||||
return chain
|
||||
|
||||
|
||||
def _is_transient_start_error(error: BaseException) -> bool:
|
||||
for cause in _exception_chain(error):
|
||||
name = type(cause).__name__
|
||||
module = type(cause).__module__
|
||||
if isinstance(cause, TimeoutError | ConnectionError | ConnectionResetError):
|
||||
return True
|
||||
if name in _TRANSIENT_TIMEOUT_NAMES:
|
||||
return True
|
||||
if name in _TRANSIENT_CONNECTION_NAMES and (
|
||||
module.startswith(("httpcore", "httpx", "agents"))
|
||||
or name in {"ConnectionError", "ConnectionResetError"}
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
async def start_session_with_retry(
|
||||
client: Any,
|
||||
create_session: Callable[[], Awaitable[Any]],
|
||||
*,
|
||||
attempts: int | None = None,
|
||||
) -> Any:
|
||||
"""Start a sandbox session, retrying transient transport failures.
|
||||
|
||||
Backend implementations should use this helper when they own both session
|
||||
creation and ``session.start()`` so failed starts can be torn down before a
|
||||
retry. The caller owns the manifest and any temporary source directories
|
||||
until this helper returns.
|
||||
"""
|
||||
max_attempts = attempts if attempts is not None else _start_attempts()
|
||||
for attempt in range(1, max_attempts + 1):
|
||||
session: Any | None = None
|
||||
try:
|
||||
session = await create_session()
|
||||
assert session is not None
|
||||
await session.start()
|
||||
except Exception as exc:
|
||||
if session is not None:
|
||||
try:
|
||||
await client.delete(session)
|
||||
except Exception as teardown_error:
|
||||
logger.warning(
|
||||
"Failed to tear down sandbox after start failure; aborting retry",
|
||||
exc_info=True,
|
||||
)
|
||||
raise exc from teardown_error
|
||||
transient = _is_transient_start_error(exc)
|
||||
if not transient or attempt == max_attempts:
|
||||
raise
|
||||
delay = _START_BACKOFF_SECONDS * (2 ** (attempt - 1))
|
||||
logger.warning(
|
||||
"Transient sandbox start failure; retrying attempt %d/%d in %.1fs",
|
||||
attempt + 1,
|
||||
max_attempts,
|
||||
delay,
|
||||
)
|
||||
await asyncio.sleep(delay)
|
||||
else:
|
||||
return session
|
||||
raise AssertionError("sandbox start retry loop completed without returning or raising")
|
||||
|
||||
|
||||
async def _docker_backend(
|
||||
*,
|
||||
@@ -50,8 +174,10 @@ async def _docker_backend(
|
||||
client = StrixDockerSandboxClient(docker.from_env())
|
||||
client.strix_bind_mounts = bind_mounts or []
|
||||
options = DockerSandboxClientOptions(image=image, exposed_ports=exposed_ports)
|
||||
session = await client.create(options=options, manifest=manifest)
|
||||
await session.start()
|
||||
session = await start_session_with_retry(
|
||||
client,
|
||||
lambda: client.create(options=options, manifest=manifest),
|
||||
)
|
||||
return client, session
|
||||
|
||||
|
||||
@@ -83,7 +209,9 @@ def register_backend(name: str, backend: SandboxBackend) -> None:
|
||||
|
||||
Intended for downstream users who ship their own runtime — register
|
||||
before any ``session_manager.create_or_reuse`` call. Re-registering
|
||||
an existing name overwrites the prior entry.
|
||||
an existing name overwrites the prior entry. Backends that own both
|
||||
session creation and ``session.start()`` should use
|
||||
:func:`start_session_with_retry`.
|
||||
"""
|
||||
_BACKENDS[name] = backend
|
||||
logger.info("Registered sandbox backend: %s", name)
|
||||
|
||||
@@ -40,6 +40,7 @@ from docker import errors as docker_errors # type: ignore[import-untyped, unuse
|
||||
from docker.models.containers import Container # 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__)
|
||||
@@ -48,7 +49,7 @@ logger = logging.getLogger(__name__)
|
||||
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]] = [] # overridden per-instance in backends.py
|
||||
strix_bind_mounts: list[dict[str, Any]] | None = None
|
||||
|
||||
async def _create_container(
|
||||
self,
|
||||
@@ -148,6 +149,15 @@ class StrixDockerSandboxClient(DockerSandboxClient):
|
||||
async def delete(self, session: SandboxSession) -> SandboxSession:
|
||||
container_id = getattr(getattr(session._inner, "state", None), "container_id", None)
|
||||
if container_id:
|
||||
with contextlib.suppress(docker_errors.NotFound, docker_errors.APIError):
|
||||
# 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
|
||||
):
|
||||
self.docker_client.containers.get(container_id).kill()
|
||||
return await super().delete(session)
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
"""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
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -12,6 +13,7 @@ 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__)
|
||||
@@ -29,16 +31,20 @@ _WORKSPACE_ROOT = "/workspace"
|
||||
|
||||
def build_session_entries(
|
||||
local_sources: list[dict[str, Any]],
|
||||
) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]]]:
|
||||
) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]], list[Path]]:
|
||||
"""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.
|
||||
``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.
|
||||
"""
|
||||
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 ""
|
||||
@@ -54,8 +60,11 @@ def build_session_entries(
|
||||
}
|
||||
)
|
||||
else:
|
||||
entries[ws_subdir] = LocalDir(src=resolved)
|
||||
return entries, bind_mounts
|
||||
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
|
||||
|
||||
|
||||
async def create_or_reuse(
|
||||
@@ -75,7 +84,7 @@ async def create_or_reuse(
|
||||
logger.info("Reusing existing sandbox session for scan %s", scan_id)
|
||||
return cached
|
||||
|
||||
entries, bind_mounts = build_session_entries(local_sources)
|
||||
entries, bind_mounts, staged_dirs = 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.)
|
||||
@@ -106,15 +115,20 @@ async def create_or_reuse(
|
||||
backend_name,
|
||||
image,
|
||||
)
|
||||
client, session = await backend(
|
||||
image=image,
|
||||
manifest=manifest,
|
||||
exposed_ports=(_CONTAINER_CAIDO_PORT,),
|
||||
bind_mounts=bind_mounts,
|
||||
)
|
||||
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)
|
||||
|
||||
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
|
||||
host_caido_url = f"http://{caido_endpoint.host}:{caido_endpoint.port}"
|
||||
scheme = "https" if caido_endpoint.tls else "http"
|
||||
host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
|
||||
logger.debug("Caido host endpoint resolved: %s", host_caido_url)
|
||||
|
||||
caido_client = await bootstrap_caido(
|
||||
|
||||
@@ -41,6 +41,7 @@ The skills are dynamically injected into the agent's system prompt, allowing it
|
||||
Notable source-aware skills:
|
||||
- `source_aware_whitebox` (coordination): white-box orchestration playbook
|
||||
- `source_aware_sast` (custom): semgrep/AST/secrets/supply-chain static triage workflow
|
||||
- `dependency_cve_scanning` (custom): trivy-based SCA workflow for reporting known dependency CVEs via `create_dependency_report`
|
||||
|
||||
---
|
||||
|
||||
|
||||
+165
-35
@@ -1,7 +1,11 @@
|
||||
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
|
||||
|
||||
|
||||
@@ -10,20 +14,82 @@ 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."""
|
||||
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
|
||||
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
|
||||
|
||||
|
||||
def get_all_skill_names() -> set[str]:
|
||||
@@ -31,6 +97,54 @@ 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():
|
||||
@@ -52,48 +166,63 @@ def validate_requested_skills(skill_list: list[str], max_skills: int = 5) -> str
|
||||
if not skill_list:
|
||||
return None
|
||||
available = get_all_skill_names()
|
||||
invalid = sorted({s for s in skill_list if s not in available})
|
||||
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})
|
||||
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``. 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`` (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.
|
||||
"""
|
||||
skills_dir = get_strix_resource_path("skills")
|
||||
if not skills_dir.exists():
|
||||
search_dirs = skill_search_dirs()
|
||||
if not search_dirs:
|
||||
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:
|
||||
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():
|
||||
candidates = _candidate_skill_files(skill_name)
|
||||
if not candidates:
|
||||
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 = (skills_dir / rel_path).read_text(encoding="utf-8")
|
||||
content = file_path.read_text(encoding="utf-8")
|
||||
except (OSError, ValueError) as e:
|
||||
logger.warning("Failed to load skill %s: %s", skill_name, e)
|
||||
continue
|
||||
@@ -101,6 +230,7 @@ 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
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
---
|
||||
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.
|
||||
@@ -0,0 +1,138 @@
|
||||
---
|
||||
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.0–10.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.
|
||||
@@ -121,6 +121,11 @@ 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
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
---
|
||||
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.
|
||||
@@ -130,6 +130,13 @@ 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
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
---
|
||||
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,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. 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.
|
||||
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.
|
||||
|
||||
### Telemetry Policy
|
||||
|
||||
@@ -16,7 +16,8 @@ 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)\
|
||||
**Aggregate Metrics:** Vulnerability counts by severity
|
||||
**Feature Usage:** Which built-in skills are loaded\
|
||||
**Aggregate Metrics:** Vulnerability counts by severity and weakness category (CWE)
|
||||
|
||||
### What We **Never** Collect
|
||||
|
||||
|
||||
@@ -63,6 +63,18 @@ _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."""
|
||||
@@ -119,6 +131,7 @@ 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]
|
||||
|
||||
@@ -26,10 +26,10 @@ def _is_enabled() -> bool:
|
||||
return load_settings().telemetry.enabled
|
||||
|
||||
|
||||
def _send(event: str, properties: dict[str, Any]) -> None:
|
||||
def _send(event: str, properties: dict[str, Any]) -> bool:
|
||||
if not _is_enabled():
|
||||
logger.debug("posthog disabled; skipping event %s", event)
|
||||
return
|
||||
return False
|
||||
try:
|
||||
payload = {
|
||||
"api_key": _POSTHOG_PUBLIC_API_KEY,
|
||||
@@ -46,8 +46,10 @@ def _send(event: str, properties: dict[str, Any]) -> None:
|
||||
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(
|
||||
@@ -71,17 +73,34 @@ def start(
|
||||
)
|
||||
|
||||
|
||||
def finding(severity: str) -> None:
|
||||
def finding(severity: str, cwe: str | None = None, is_cve: bool = False) -> 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()
|
||||
@@ -110,11 +129,11 @@ def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
|
||||
except (TypeError, ValueError, AttributeError):
|
||||
pass
|
||||
|
||||
_send(
|
||||
report_state.posthog_scan_ended_sent = _send(
|
||||
"scan_ended",
|
||||
{
|
||||
**base_props(),
|
||||
"exit_reason": exit_reason,
|
||||
"exit_reason": report_state.scan_ended_exit_reason,
|
||||
"duration_seconds": round(duration),
|
||||
"vulnerabilities_total": len(report_state.vulnerability_reports),
|
||||
**{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},
|
||||
|
||||
@@ -28,10 +28,10 @@ def _is_enabled() -> bool:
|
||||
return load_settings().telemetry.enabled
|
||||
|
||||
|
||||
def _send(event: str, properties: dict[str, Any]) -> None:
|
||||
def _send(event: str, properties: dict[str, Any]) -> bool:
|
||||
if not _is_enabled():
|
||||
logger.debug("scarf disabled; skipping event %s", event)
|
||||
return
|
||||
return False
|
||||
try:
|
||||
props = dict(properties)
|
||||
version = str(props.pop("strix_version", get_version()) or "unknown")
|
||||
@@ -47,8 +47,10 @@ def _send(event: str, properties: dict[str, Any]) -> None:
|
||||
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(
|
||||
@@ -73,18 +75,36 @@ def start(
|
||||
)
|
||||
|
||||
|
||||
def finding(severity: str) -> None:
|
||||
def finding(severity: str, cwe: str | None = None, is_cve: bool = False) -> 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()
|
||||
@@ -115,12 +135,12 @@ def end(report_state: ReportState, exit_reason: str = "completed") -> None:
|
||||
except (TypeError, ValueError, AttributeError):
|
||||
pass
|
||||
|
||||
_send(
|
||||
report_state.scarf_scan_ended_sent = _send(
|
||||
"scan_ended",
|
||||
{
|
||||
**base_props(),
|
||||
"session": SESSION_ID,
|
||||
"exit_reason": exit_reason,
|
||||
"exit_reason": report_state.scan_ended_exit_reason,
|
||||
"duration_seconds": round(duration),
|
||||
"vulnerabilities_total": len(report_state.vulnerability_reports),
|
||||
**{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},
|
||||
|
||||
@@ -481,9 +481,10 @@ async def agent_finish(
|
||||
3. Stops this subagent's execution.
|
||||
|
||||
**Vulnerability findings must already be filed via
|
||||
``create_vulnerability_report`` before calling this.** The
|
||||
``findings`` field here is for narrative summary only — it does
|
||||
not register vulns in the scan report.
|
||||
``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.
|
||||
|
||||
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,
|
||||
@@ -494,8 +495,9 @@ 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; this is for
|
||||
narrative.
|
||||
``create_vulnerability_report`` first (or
|
||||
``create_dependency_report`` for dependency CVEs); this is
|
||||
for narrative.
|
||||
success: Whether the assigned subtask was completed
|
||||
successfully. Default ``True``.
|
||||
report_to_parent: Whether to deliver the completion report to
|
||||
|
||||
+101
-4
@@ -96,6 +96,15 @@ 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
|
||||
@@ -108,19 +117,39 @@ 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`` (un-reported findings are not
|
||||
tracked and not credited).
|
||||
``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.
|
||||
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.
|
||||
|
||||
**Customer-facing report rules** (this output is rendered into the
|
||||
final PDF the client sees):
|
||||
**Report output rules** (this content may be rendered into generated
|
||||
reports):
|
||||
|
||||
- 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:
|
||||
@@ -140,6 +169,74 @@ 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.
|
||||
|
||||
@@ -148,8 +148,12 @@ _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
|
||||
*,
|
||||
@@ -161,12 +165,16 @@ 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]:
|
||||
@@ -180,11 +188,19 @@ 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 = {}
|
||||
@@ -268,6 +284,9 @@ 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,
|
||||
@@ -275,6 +294,7 @@ 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,
|
||||
)
|
||||
@@ -309,12 +329,16 @@ 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.
|
||||
|
||||
@@ -327,24 +351,46 @@ 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.
|
||||
|
||||
**Customer-facing report rules** (the report is PDF-rendered for
|
||||
delivery):
|
||||
**Report output rules** (this content may be rendered into generated
|
||||
reports):
|
||||
|
||||
- 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.
|
||||
- Standard finding structure: Overview → Severity & CVSS →
|
||||
Affected assets → Technical details → PoC (steps + code) →
|
||||
Impact → Remediation → Evidence (in technical_analysis).
|
||||
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``).
|
||||
- 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
|
||||
@@ -407,13 +453,23 @@ 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: How the vuln was discovered + what it is.
|
||||
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.
|
||||
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.
|
||||
poc_description: Step-by-step reproduction (steps only, no code).
|
||||
poc_script_code: Working PoC (Python preferred).
|
||||
remediation_steps: Specific, actionable fix.
|
||||
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``.
|
||||
cvss_breakdown: 8-metric object per the format above.
|
||||
endpoint: API path / Git path (e.g. ``/api/login``).
|
||||
method: HTTP method when relevant.
|
||||
@@ -482,6 +538,47 @@ 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")
|
||||
@@ -503,12 +600,327 @@ 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,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
"""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
|
||||
@@ -0,0 +1,88 @@
|
||||
"""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 != ""
|
||||
@@ -0,0 +1,218 @@
|
||||
"""Tests for transient sandbox start retries."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from agents.sandbox.errors import (
|
||||
LocalDirReadError,
|
||||
WorkspaceArchiveWriteError,
|
||||
WorkspaceStartError,
|
||||
)
|
||||
|
||||
from strix.runtime import session_manager
|
||||
from strix.runtime.backends import start_session_with_retry
|
||||
|
||||
|
||||
class _FakeSession:
|
||||
def __init__(self, failures: list[BaseException]) -> None:
|
||||
self._failures = iter(failures)
|
||||
|
||||
async def start(self) -> None:
|
||||
try:
|
||||
raise next(self._failures)
|
||||
except StopIteration:
|
||||
return
|
||||
|
||||
async def resolve_exposed_port(self, _port: int) -> SimpleNamespace:
|
||||
return SimpleNamespace(tls=False, host="127.0.0.1", port=48080)
|
||||
|
||||
|
||||
class _FakeClient:
|
||||
def __init__(self, *, delete_error: BaseException | None = None) -> None:
|
||||
self.created = 0
|
||||
self.deleted: list[_FakeSession] = []
|
||||
self.delete_error = delete_error
|
||||
|
||||
async def create(self) -> _FakeSession:
|
||||
self.created += 1
|
||||
failures: list[BaseException] = []
|
||||
if self.created == 1:
|
||||
failures = [
|
||||
WorkspaceStartError(
|
||||
path=Path("/workspace"),
|
||||
cause=WorkspaceArchiveWriteError(
|
||||
path=Path("/workspace"),
|
||||
cause=TimeoutError("transient transport timeout"),
|
||||
),
|
||||
)
|
||||
]
|
||||
return _FakeSession(failures)
|
||||
|
||||
async def delete(self, session: _FakeSession) -> None:
|
||||
self.deleted.append(session)
|
||||
if self.delete_error is not None:
|
||||
raise self.delete_error
|
||||
|
||||
|
||||
async def test_transient_workspace_failure_retries_and_tears_down(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
client = _FakeClient()
|
||||
sleeps: list[float] = []
|
||||
|
||||
async def record_sleep(delay: float) -> None:
|
||||
sleeps.append(delay)
|
||||
|
||||
monkeypatch.setattr("strix.runtime.backends.asyncio.sleep", record_sleep)
|
||||
|
||||
session = await start_session_with_retry(client, client.create, attempts=3)
|
||||
|
||||
assert isinstance(session, _FakeSession)
|
||||
assert client.created == 2
|
||||
assert len(client.deleted) == 1
|
||||
assert sleeps == [2.0]
|
||||
|
||||
|
||||
async def test_non_transient_workspace_failure_does_not_retry() -> None:
|
||||
client = _FakeClient()
|
||||
session = _FakeSession([LocalDirReadError(src=Path("/workspace/repo"))])
|
||||
|
||||
async def create_session() -> _FakeSession:
|
||||
client.created += 1
|
||||
return session
|
||||
|
||||
with pytest.raises(LocalDirReadError):
|
||||
await start_session_with_retry(client, create_session, attempts=3)
|
||||
|
||||
assert client.created == 1
|
||||
assert client.deleted == [session]
|
||||
|
||||
|
||||
async def test_teardown_failure_raises_original_error_without_retry() -> None:
|
||||
start_error = WorkspaceStartError(
|
||||
path=Path("/workspace"),
|
||||
cause=TimeoutError("transient transport timeout"),
|
||||
)
|
||||
teardown_error = RuntimeError("teardown failed")
|
||||
client = _FakeClient(delete_error=teardown_error)
|
||||
session = _FakeSession([start_error])
|
||||
|
||||
async def create_session() -> _FakeSession:
|
||||
client.created += 1
|
||||
return session
|
||||
|
||||
with pytest.raises(WorkspaceStartError) as caught:
|
||||
await start_session_with_retry(client, create_session, attempts=3)
|
||||
|
||||
assert caught.value is start_error
|
||||
assert caught.value.__cause__ is teardown_error
|
||||
assert client.created == 1
|
||||
assert client.deleted == [session]
|
||||
|
||||
|
||||
async def test_each_transient_attempt_is_torn_down(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
client = _FakeClient()
|
||||
client.created = 0
|
||||
sessions: list[_FakeSession] = []
|
||||
sleeps: list[float] = []
|
||||
|
||||
async def record_sleep(delay: float) -> None:
|
||||
sleeps.append(delay)
|
||||
|
||||
monkeypatch.setattr("strix.runtime.backends.asyncio.sleep", record_sleep)
|
||||
|
||||
async def create_session() -> _FakeSession:
|
||||
client.created += 1
|
||||
failures: list[BaseException] = []
|
||||
if client.created < 3:
|
||||
failures = [
|
||||
WorkspaceStartError(
|
||||
path=Path("/workspace"),
|
||||
cause=TimeoutError("transient transport timeout"),
|
||||
)
|
||||
]
|
||||
session = _FakeSession(failures)
|
||||
sessions.append(session)
|
||||
return session
|
||||
|
||||
result = await start_session_with_retry(client, create_session, attempts=3)
|
||||
|
||||
assert result is sessions[2]
|
||||
assert client.deleted == sessions[:2]
|
||||
assert sleeps == [2.0, 4.0]
|
||||
|
||||
|
||||
async def test_staged_dirs_survive_retries_and_cleanup_once(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "real.txt").write_text("content")
|
||||
(repo / "link.txt").symlink_to(repo / "real.txt")
|
||||
|
||||
client = _FakeClient()
|
||||
observed_paths: list[Path] = []
|
||||
sleeps: list[float] = []
|
||||
original_rmtree = shutil.rmtree # pyright: ignore[reportDeprecated]
|
||||
removed_paths: list[Path] = []
|
||||
|
||||
async def record_sleep(delay: float) -> None:
|
||||
sleeps.append(delay)
|
||||
|
||||
monkeypatch.setattr("strix.runtime.backends.asyncio.sleep", record_sleep)
|
||||
|
||||
def record_rmtree(path: str | Path, **kwargs: Any) -> None:
|
||||
removed_paths.append(Path(path))
|
||||
original_rmtree(path, **kwargs) # pyright: ignore[reportDeprecated]
|
||||
|
||||
monkeypatch.setattr("strix.runtime.session_manager.shutil.rmtree", record_rmtree)
|
||||
monkeypatch.setattr(
|
||||
session_manager,
|
||||
"load_settings",
|
||||
lambda: SimpleNamespace(runtime=SimpleNamespace(backend="fake")),
|
||||
)
|
||||
monkeypatch.setattr(session_manager, "bootstrap_caido", _bootstrap_caido)
|
||||
|
||||
async def fake_backend(**kwargs: Any) -> tuple[_FakeClient, _FakeSession]:
|
||||
staged_path = kwargs["manifest"].entries["repo"].src
|
||||
|
||||
async def create_session() -> _FakeSession:
|
||||
observed_paths.append(Path(staged_path))
|
||||
return await client.create()
|
||||
|
||||
session = await start_session_with_retry(client, create_session, attempts=3)
|
||||
return client, session
|
||||
|
||||
def fake_get_backend(_name: str) -> Any:
|
||||
return fake_backend
|
||||
|
||||
monkeypatch.setattr(session_manager, "get_backend", fake_get_backend)
|
||||
|
||||
try:
|
||||
await session_manager.create_or_reuse(
|
||||
"retry-test",
|
||||
image="test-image",
|
||||
local_sources=[
|
||||
{
|
||||
"source_path": str(repo),
|
||||
"workspace_subdir": "repo",
|
||||
}
|
||||
],
|
||||
)
|
||||
finally:
|
||||
await session_manager.cleanup("retry-test")
|
||||
|
||||
assert len(observed_paths) == 2
|
||||
assert observed_paths[0] == observed_paths[1]
|
||||
assert observed_paths[0] in removed_paths
|
||||
assert not observed_paths[0].exists()
|
||||
|
||||
|
||||
async def _bootstrap_caido(*_args: Any, **_kwargs: Any) -> object:
|
||||
return object()
|
||||
@@ -0,0 +1,90 @@
|
||||
"""Tests for CLI target-list argument parsing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
cli_main: Any = importlib.import_module("strix.interface.main")
|
||||
|
||||
|
||||
def _stub_settings(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(
|
||||
cli_main,
|
||||
"load_settings",
|
||||
lambda: SimpleNamespace(runtime=SimpleNamespace(max_local_copy_mb=1024)),
|
||||
)
|
||||
|
||||
|
||||
def test_parse_arguments_accepts_target_list_file(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
target_list = tmp_path / "targets.txt"
|
||||
target_list.write_text(
|
||||
"https://test1.com/\n"
|
||||
"\n"
|
||||
"http://test2.com:5789/\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
_stub_settings(monkeypatch)
|
||||
monkeypatch.setattr(sys, "argv", ["strix", "--target-list", str(target_list), "-n"])
|
||||
|
||||
args = cli_main.parse_arguments()
|
||||
|
||||
assert [target["original"] for target in args.targets_info] == [
|
||||
"https://test1.com/",
|
||||
"http://test2.com:5789/",
|
||||
]
|
||||
assert [target["type"] for target in args.targets_info] == [
|
||||
"web_application",
|
||||
"web_application",
|
||||
]
|
||||
|
||||
|
||||
def test_parse_arguments_combines_target_and_target_list(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
target_list = tmp_path / "targets.txt"
|
||||
target_list.write_text("http://test2.com:5789/\n", encoding="utf-8")
|
||||
_stub_settings(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
["strix", "-t", "https://test1.com/", "--target-list", str(target_list)],
|
||||
)
|
||||
|
||||
args = cli_main.parse_arguments()
|
||||
|
||||
assert [target["original"] for target in args.targets_info] == [
|
||||
"https://test1.com/",
|
||||
"http://test2.com:5789/",
|
||||
]
|
||||
|
||||
|
||||
def test_parse_arguments_rejects_resume_with_target_list(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
target_list = tmp_path / "targets.txt"
|
||||
target_list.write_text("https://test1.com/\n", encoding="utf-8")
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
["strix", "--resume", "old-run", "--target-list", str(target_list)],
|
||||
)
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
cli_main.parse_arguments()
|
||||
|
||||
assert (
|
||||
"Cannot combine --resume with --target/--target-list/--mount"
|
||||
in capsys.readouterr().err
|
||||
)
|
||||
@@ -26,6 +26,7 @@ _LLM_ENV_KEYS = [
|
||||
"LITELLM_BASE_URL",
|
||||
"OLLAMA_API_BASE",
|
||||
"STRIX_REASONING_EFFORT",
|
||||
"STRIX_FORCE_REQUIRED_TOOL_CHOICE",
|
||||
"LLM_TIMEOUT",
|
||||
"PERPLEXITY_API_KEY",
|
||||
# RuntimeSettings
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
"""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()
|
||||
+44
-1
@@ -7,7 +7,7 @@ from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from strix.core.inputs import build_root_task, child_initial_input
|
||||
from strix.core.inputs import build_root_task, child_initial_input, make_model_settings
|
||||
|
||||
|
||||
def _child_kwargs(parent_history: list[Any]) -> dict[str, Any]:
|
||||
@@ -112,3 +112,46 @@ 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"
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
"""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()
|
||||
@@ -19,6 +19,7 @@ from strix.interface.utils import (
|
||||
dedupe_local_targets,
|
||||
directory_size_bytes,
|
||||
find_oversized_local_targets,
|
||||
read_target_list_file,
|
||||
)
|
||||
|
||||
|
||||
@@ -157,6 +158,66 @@ def test_build_mount_targets_info_rejects_empty_path(empty: str) -> None:
|
||||
build_mount_targets_info([empty])
|
||||
|
||||
|
||||
def test_read_target_list_file_strips_blank_lines(tmp_path: Path) -> None:
|
||||
target_list = tmp_path / "targets.txt"
|
||||
target_list.write_text(
|
||||
"\n"
|
||||
" https://test1.com/ \n"
|
||||
"\n"
|
||||
"http://test2.com:5789/\n"
|
||||
" \n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
assert read_target_list_file(str(target_list)) == [
|
||||
"https://test1.com/",
|
||||
"http://test2.com:5789/",
|
||||
]
|
||||
|
||||
|
||||
def test_read_target_list_file_ignores_comment_lines(tmp_path: Path) -> None:
|
||||
target_list = tmp_path / "targets.txt"
|
||||
target_list.write_text(
|
||||
"# production targets\n"
|
||||
"https://test1.com/\n"
|
||||
" # staging targets\n"
|
||||
"http://test2.com:5789/\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
assert read_target_list_file(str(target_list)) == [
|
||||
"https://test1.com/",
|
||||
"http://test2.com:5789/",
|
||||
]
|
||||
|
||||
|
||||
def test_read_target_list_file_rejects_empty_file(tmp_path: Path) -> None:
|
||||
target_list = tmp_path / "targets.txt"
|
||||
target_list.write_text(" \n# no targets yet\n\n", encoding="utf-8")
|
||||
|
||||
with pytest.raises(ValueError, match="is empty"):
|
||||
read_target_list_file(str(target_list))
|
||||
|
||||
|
||||
def test_read_target_list_file_rejects_missing_path(tmp_path: Path) -> None:
|
||||
with pytest.raises(ValueError, match="not an existing file"):
|
||||
read_target_list_file(str(tmp_path / "missing.txt"))
|
||||
|
||||
|
||||
def test_read_target_list_file_rejects_non_utf8_file(tmp_path: Path) -> None:
|
||||
target_list = tmp_path / "targets.txt"
|
||||
target_list.write_bytes(b"https://test1.com/\xff\n")
|
||||
|
||||
with pytest.raises(ValueError, match="must be valid UTF-8 text"):
|
||||
read_target_list_file(str(target_list))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("empty", ["", " "])
|
||||
def test_read_target_list_file_rejects_empty_path(empty: str) -> None:
|
||||
with pytest.raises(ValueError, match="must not be empty"):
|
||||
read_target_list_file(empty)
|
||||
|
||||
|
||||
def test_dedupe_keeps_distinct_targets_in_order() -> None:
|
||||
targets = [
|
||||
_local_target("/a"),
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
"""Tests for LLM model recommendation helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from strix.config.models import RECOMMENDED_MODEL_NAMES, is_recommended_or_frontier_model
|
||||
|
||||
|
||||
@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_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)
|
||||
@@ -0,0 +1,26 @@
|
||||
"""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"])
|
||||
@@ -0,0 +1,74 @@
|
||||
"""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
|
||||
@@ -79,6 +79,40 @@ 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_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"),
|
||||
|
||||
@@ -0,0 +1,564 @@
|
||||
"""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
|
||||
@@ -33,7 +33,11 @@ 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")
|
||||
llm=types.SimpleNamespace(
|
||||
model="openai/gpt-4o",
|
||||
reasoning_effort="high",
|
||||
force_required_tool_choice=False,
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(runner, "load_settings", lambda: settings)
|
||||
monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
"""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,
|
||||
)
|
||||
)
|
||||
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"}
|
||||
@@ -18,15 +18,18 @@ 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 = build_session_entries([_source("repo", str(tmp_path))])
|
||||
entries, bind_mounts, staged_dirs = 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 = build_session_entries([_source("repo", str(tmp_path), mount=True)])
|
||||
entries, bind_mounts, _staged = build_session_entries(
|
||||
[_source("repo", str(tmp_path), mount=True)]
|
||||
)
|
||||
|
||||
assert entries == {}
|
||||
assert bind_mounts == [
|
||||
@@ -44,7 +47,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
|
||||
copied.mkdir()
|
||||
mounted.mkdir()
|
||||
|
||||
entries, bind_mounts = build_session_entries(
|
||||
entries, bind_mounts, _staged = build_session_entries(
|
||||
[
|
||||
_source("copied", str(copied)),
|
||||
_source("mounted", str(mounted), mount=True),
|
||||
@@ -57,7 +60,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
|
||||
|
||||
|
||||
def test_incomplete_sources_are_skipped() -> None:
|
||||
entries, bind_mounts = build_session_entries(
|
||||
entries, bind_mounts, staged_dirs = build_session_entries(
|
||||
[
|
||||
{"source_path": "", "workspace_subdir": "x"},
|
||||
{"source_path": "/p", "workspace_subdir": ""},
|
||||
@@ -65,3 +68,20 @@ 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"
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
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"]) == {}
|
||||
@@ -41,6 +41,15 @@ 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()
|
||||
|
||||
Reference in New Issue
Block a user