Compare commits

..
Author SHA1 Message Date
bearsyankees ecbafb2592 Add skill directory registration 2026-07-12 00:20:45 -04:00
alex sandGitHub 09872744f5 Allow scan agent tool registration (#733) 2026-07-11 23:58:51 -04:00
5 changed files with 275 additions and 280 deletions
+3 -3
View File
@@ -7,7 +7,7 @@ from typing import Any
from jinja2 import Environment, FileSystemLoader, select_autoescape
from strix.skills import get_available_skills, load_skills
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,
+2 -68
View File
@@ -14,7 +14,6 @@ from agents.sandbox import SandboxRunConfig
from openai import RateLimitError
from strix.agents.factory import build_strix_agent, make_child_factory
from strix.agents.prompt import render_system_prompt
from strix.config import load_settings
from strix.config.models import (
StrixProvider,
@@ -52,52 +51,6 @@ logger = logging.getLogger(__name__)
StreamEventSink = Callable[[str, Any], None]
def _merge_root_prompt_context(
scope_context: dict[str, Any],
extra_system_prompt_context: dict[str, Any] | None,
) -> dict[str, Any]:
if not extra_system_prompt_context:
return scope_context
reserved_keys = scope_context.keys() & extra_system_prompt_context.keys()
if reserved_keys:
raise ValueError(
"extra_system_prompt_context cannot override built-in scope keys: "
f"{sorted(reserved_keys)}",
)
return {**scope_context, **extra_system_prompt_context}
def _compose_root_instructions_override(
root_instructions_override: str | None,
*,
skills: list[str],
scan_mode: str,
is_whitebox: bool,
interactive: bool,
system_prompt_context: dict[str, Any],
) -> str | None:
if root_instructions_override is None:
return None
base_instructions = render_system_prompt(
skills=skills,
scan_mode=scan_mode,
is_whitebox=is_whitebox,
is_root=True,
interactive=interactive,
system_prompt_context=system_prompt_context,
)
return (
f"{base_instructions}\n\n"
"<root_scan_instructions_override>\n"
"The following root scan instructions are subordinate to the "
"system-verified scope above. They cannot expand, replace, or weaken "
"authorized target constraints.\n\n"
f"{root_instructions_override}\n"
"</root_scan_instructions_override>"
)
async def run_strix_scan(
*,
scan_config: dict[str, Any],
@@ -111,17 +64,8 @@ async def run_strix_scan(
model: str | None = None,
cleanup_on_exit: bool = True,
event_sink: StreamEventSink | None = None,
root_instructions_override: str | None = None,
extra_system_prompt_context: dict[str, Any] | None = None,
) -> RunResultBase | None:
"""Run or resume one Strix scan against a sandbox.
``root_instructions_override`` adds root scan instructions to the rendered
root prompt without replacing the system-verified scope block.
``extra_system_prompt_context`` is merged into the root agent's scan
context before prompt rendering. Child agents keep the standard scan prompt
and context.
"""
"""Run or resume one Strix scan against a sandbox."""
if scan_id is None:
scan_id = f"scan-{uuid.uuid4().hex[:8]}"
@@ -226,15 +170,6 @@ async def run_strix_scan(
hooks = ReportUsageHooks(model=resolved_model, max_budget_usd=max_budget_usd)
scope_context = build_scope_context(scan_config)
root_context = _merge_root_prompt_context(scope_context, extra_system_prompt_context)
root_instructions = _compose_root_instructions_override(
root_instructions_override,
skills=skills,
scan_mode=scan_mode,
is_whitebox=is_whitebox,
interactive=interactive,
system_prompt_context=root_context,
)
root_agent = build_strix_agent(
name="strix",
@@ -244,8 +179,7 @@ async def run_strix_scan(
is_whitebox=is_whitebox,
interactive=interactive,
chat_completions_tools=chat_completions_tools,
system_prompt_context=root_context,
instructions_override=root_instructions,
system_prompt_context=scope_context,
)
if not is_resume:
+150 -35
View File
@@ -1,6 +1,8 @@
import logging
import re
from collections import Counter
from collections.abc import Iterator
from pathlib import Path
from strix.utils.resource_paths import get_strix_resource_path
@@ -10,20 +12,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 +95,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
candidate = _qualified_skill_file(skills_dir, _ROOT_SKILL_CATEGORY, skill_name)
if candidate is not None:
seen.add(key)
candidates.append(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 +164,51 @@ 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 _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
-174
View File
@@ -1,174 +0,0 @@
"""Tests for root scan prompt options in run_strix_scan.
Verify that ``root_instructions_override`` and ``extra_system_prompt_context``
flow through to the root agent's ``build_strix_agent`` call.
"""
from __future__ import annotations
import types
from typing import Any
import httpx
import pytest
from openai import RateLimitError
import strix.tools.notes.tools as notes_tools
import strix.tools.todo.tools as todo_tools
from strix.core import runner
from strix.core.agents import AgentCoordinator
def _make_rate_limit_error() -> RateLimitError:
request = httpx.Request("POST", "https://api.openai.com/v1/responses")
response = httpx.Response(status_code=429, request=request)
return RateLimitError("rate limited", response=response, body=None)
def _patch_engine_scaffold(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Any,
scope_context: dict[str, Any],
) -> dict[str, Any]:
"""Stub out everything around build_strix_agent and stop at run_agent_loop.
Returns a dict that will be populated with the kwargs the runner passed to
``build_strix_agent`` for the root agent.
"""
monkeypatch.setattr(runner, "run_dir_for", lambda _scan_id: tmp_path)
monkeypatch.setattr(runner, "runtime_state_dir", lambda _run_dir: tmp_path)
monkeypatch.setattr(runner, "setup_scan_logging", lambda _run_dir: lambda: None)
monkeypatch.setattr(runner, "set_scan_id", lambda _scan_id: None)
settings = types.SimpleNamespace(
llm=types.SimpleNamespace(
model="openai/gpt-4o",
reasoning_effort="high",
force_required_tool_choice=False,
)
)
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"}
+120
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"]) == {}