Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion capabilities/ai-red-teaming/capability.yaml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
schema: 1
name: ai-red-teaming
version: "1.17.0"
version: "1.17.1"
description: >
Probe the security and safety of AI applications, agents, and foundation models.
Orchestrates adversarial attack workflows to discover vulnerabilities in LLMs,
Expand Down
31 changes: 31 additions & 0 deletions capabilities/ai-red-teaming/scripts/attack_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -4330,6 +4330,33 @@ def generate_attack(params: dict) -> dict:

# Agentic attack generation — targets HTTP agent APIs

# Emitted into generated workflow scripts. Normalizes agent tool_calls to the
# {name, arguments:str} shape the SDK scorers expect, regardless of whether the
# agent returns {"tool": ..., "arguments": {dict}} or OpenAI-style nested
# {"function": {"name": ..., "arguments": "<str>"}}. Without this, dict arguments
# and the "tool" alias make _extract_tool_calls return [] and every agentic
# tool scorer silently reads no tool calls.
_NORMALIZE_TOOL_CALLS_SRC = [
"def _normalize_tool_calls(raw):",
' """Coerce agent tool_calls to [{name, arguments:str}] for SDK scorers."""',
" out = []",
" for tc in raw or []:",
" if not isinstance(tc, dict):",
" continue",
' fn = tc.get("function") if isinstance(tc.get("function"), dict) else tc',
' name = fn.get("name") or fn.get("tool") or ""',
' args = fn.get("arguments", fn.get("args", ""))',
" if not isinstance(args, str):",
" try:",
" args = json.dumps(args)",
" except Exception:",
" args = str(args)",
' out.append({"name": name, "arguments": args})',
" return out",
"",
"",
]

# Response extraction presets for common agent API formats
_AGENT_PRESETS: dict[str, dict[str, str]] = {
"openai_assistants": {
Expand Down Expand Up @@ -4377,6 +4404,7 @@ def _build_agent_target_code(agent_config: dict) -> str:
escaped_tc_path = _safe_str(tool_calls_path)

lines = [
*_NORMALIZE_TOOL_CALLS_SRC,
"@task",
"async def target(prompt: str) -> dict:",
' """Call external agent API and extract text + tool_calls."""',
Expand Down Expand Up @@ -4406,6 +4434,7 @@ def _build_agent_target_code(agent_config: dict) -> str:
" tool_calls = tc_matches[0] if tc_matches else []",
" if not isinstance(tool_calls, list):",
" tool_calls = [tool_calls] if tool_calls else []",
" tool_calls = _normalize_tool_calls(tool_calls)",
"",
' return {"content": content, "tool_calls": tool_calls}',
"",
Expand Down Expand Up @@ -4836,6 +4865,7 @@ def _build_atlas_target_code(agent_config: dict) -> str:

escaped_url = _safe_str(agent_url)
lines = [
*_NORMALIZE_TOOL_CALLS_SRC,
"async def target(prompt: str, *, surface: str = \"direct\", injection: str | None = None) -> dict:",
' """POST to the ATLAS multi-agent environment and return content + tool calls."""',
" import httpx",
Expand All @@ -4855,6 +4885,7 @@ def _build_atlas_target_code(agent_config: dict) -> str:
' tool_calls = data.get("tool_calls") or []',
" if not isinstance(tool_calls, list):",
" tool_calls = [tool_calls] if tool_calls else []",
" tool_calls = _normalize_tool_calls(tool_calls)",
" return {",
' "content": content,',
' "tool_calls": tool_calls,',
Expand Down
49 changes: 49 additions & 0 deletions capabilities/ai-red-teaming/tests/test_attack_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -1261,3 +1261,52 @@ def test_inversion_unknown_attack_errors(self, tmp_path, monkeypatch) -> None:
{"attack_type": "bogus", "api_url": "http://t/predict", "generate_only": True}
)
assert "error" in result and "Unknown inversion" in result["error"]


class TestAgentToolCallNormalization:
"""Generated agent targets must normalize tool_calls to {name, arguments:str}.

Regression for the agentic scoring blind spot: agents returning
{"tool": ..., "arguments": {dict}} produced empty tool_calls, so every
agentic tool scorer silently read no evidence.
"""

def _normalizer(self):
ns: dict = {"json": json}
exec("\n".join(runner._NORMALIZE_TOOL_CALLS_SRC), ns) # noqa: S102
return ns["_normalize_tool_calls"]

def test_target_code_calls_normalizer(self) -> None:
code = runner._build_agent_target_code(
{
"agent_url": "http://t/chat",
"agent_auth_type": "none",
"agent_response_text_path": "$.content",
"agent_response_tool_calls_path": "$.tool_calls",
}
)
compile(code, "<gen>", "exec")
assert "def _normalize_tool_calls" in code
assert "tool_calls = _normalize_tool_calls(tool_calls)" in code

def test_normalizes_tool_alias_and_dict_args(self) -> None:
norm = self._normalizer()
out = norm([{"tool": "execute_command", "arguments": {"command": "cat /etc/passwd"}}])
assert len(out) == 1
assert out[0]["name"] == "execute_command"
assert isinstance(out[0]["arguments"], str)
assert "cat /etc/passwd" in out[0]["arguments"]

def test_normalizes_openai_nested_and_passes_strings_through(self) -> None:
norm = self._normalizer()
out = norm(
[
{"function": {"name": "send_email", "arguments": '{"to":"x"}'}},
{"tool": "noop"},
"junk",
]
)
assert out[0]["name"] == "send_email"
assert out[0]["arguments"] == '{"to":"x"}'
assert out[1]["name"] == "noop"
assert len(out) == 2 # non-dict entries dropped
Loading