Merge branch 'main' into fix/report-generation-fallback

This commit is contained in:
Kundan Krishna 2026-07-15 18:56:32 +05:30 • committed by GitHub
commit 506858bc32
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
43 changed files with 5144 additions and 1564 deletions

View file

@ -27,6 +27,7 @@
<a href="https://x.com/strix_ai"><img src="https://github.com/usestrix/.github/raw/main/imgs/X.png" height="40" alt="Follow on X"></a>
<a href="https://trendshift.io/repositories/15362?utm_source=trendshift-badge&amp;utm_medium=badge&amp;utm_campaign=badge-trendshift-15362" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/trendshift/repositories/15362/weekly" alt="usestrix%2Fstrix | Trendshift" width="250" height="55"/></a>
<a href="https://trendshift.io/repositories/15362" target="_blank"><img src="https://trendshift.io/api/badge/repositories/15362" alt="usestrix/strix | Trendshift" width="250" height="55"/></a>
<a href="https://trendshift.io/repositories/15362?utm_source=trendshift-badge&amp;utm_medium=badge&amp;utm_campaign=badge-trendshift-15362" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/trendshift/repositories/15362/weekly" alt="usestrix%2Fstrix | Trendshift" width="250" height="55"/></a>

View file

@ -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",

View file

@ -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)",

View file

@ -7,7 +7,7 @@ from typing import Any
from jinja2 import Environment, FileSystemLoader, select_autoescape
from strix.skills import get_available_skills, load_skills
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,

View file

@ -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>

View file

@ -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

View file

@ -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

View file

@ -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]}"
@ -170,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",
@ -179,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:

View file

@ -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
@ -265,6 +267,7 @@ def _provider_import_hint(exc: BaseException, model: str) -> str | None:
async def warm_up_llm() -> None:
async def warm_up_llm(show_model_warning: bool = True) -> None:
console = Console()
logger.info("Warming up LLM connection")
@ -306,6 +309,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(
@ -827,7 +856,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()

View file

@ -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)
@ -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)

View file

@ -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)

View file

@ -1148,6 +1148,9 @@ def read_target_list_file(path_str: str) -> list[str]:
]
except UnicodeDecodeError as e:
raise ValueError(f"Target list file '{path_str}' must be valid UTF-8 text: {e!s}") from 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

View file

@ -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

View file

@ -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)

View file

@ -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)

View file

@ -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.__class__.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)

View file

@ -50,6 +50,7 @@ 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: ClassVar[list[dict[str, Any]]] = [] # overridden per-instance in backends.py
strix_bind_mounts: list[dict[str, Any]] | None = None
async def _create_container(
self,

View file

@ -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

View file

@ -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,12 +115,16 @@ 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)
scheme = "https" if caido_endpoint.tls else "http"

View file

@ -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`
---

View file

@ -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

View file

@ -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.

View file

@ -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

View file

@ -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

View file

@ -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.

View file

@ -2,7 +2,7 @@
To help make Strix better for everyone, we collect anonymized data that helps us understand how to better improve our AI security agent for our users, guide the addition of new features, and fix common errors and bugs. This feedback loop is crucial for improving Strix's capabilities and user experience.
We use [PostHog](https://posthog.com), an open-source analytics platform, for data collection and analysis. 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

View file

@ -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()},

View file

@ -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()},

View file

@ -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

View file

@ -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.

View file

@ -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,
)

View file

@ -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

View file

@ -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 != ""

View file

@ -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()

View file

@ -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()

View file

@ -162,6 +162,11 @@ 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\nhttp://test2.com:5789/\n \n",
"\n"
" https://test1.com/ \n"
"\n"
"http://test2.com:5789/\n"
" \n",
encoding="utf-8",
)
@ -175,6 +180,10 @@ 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\nhttps://test1.com/\n # staging targets\nhttp://test2.com:5789/\n",
"# production targets\n"
"https://test1.com/\n"
" # staging targets\n"
"http://test2.com:5789/\n",
encoding="utf-8",
)

66
tests/test_models.py Normal file
View file

@ -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)

View file

@ -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"),

View file

@ -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

View file

@ -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"}

View file

@ -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"

View file

@ -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"]) == {}

3021
uv.lock generated

File diff suppressed because it is too large Load diff