Skip to content
Draft
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
6 changes: 4 additions & 2 deletions capabilities/ai-red-teaming/agents/ai-red-teaming-agent.md
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,8 @@ Keep it to a single line; don't pad it.
1. Pick the right generator for the target type:
- LLM with a specific goal → `generate_attack`
- LLM by harm category / sweep → `generate_category_attack`
- Agent/MCP/HTTP endpoint with tools → `generate_agentic_attack`
- Agent/MCP/HTTP endpoint with tools, ONE specific attack → `generate_agentic_attack`
- **"Run all possible attacks" / "red team my agent" / comprehensive agent audit → `generate_agentic_suite_attack`**. This is the turnkey full-coverage path: it runs every OWASP-ASI category (auto-selecting the mapped attacks, family transforms, and detection scorers) against the agent endpoint — the user does NOT need to name attacks. Pass `agent_url` + `attacker_model` (+ preset/template + `agent_dangerous_tools`); omit `categories` to run everything. Use this whenever the user hands you an agent and asks for a broad/complete assessment.
- Multi-agent system (delegation chains, trust boundaries) → provision a hosted environment with `provision_environment` (or use a user-supplied URL), then `generate_atlas_attack`
- ML image classifier (perturb pixels to misclassify) → `generate_image_attack`
- **Multimodal LLM (vision/audio/video) with media inputs → `generate_multimodal_attack`**. Detect this when the user attaches or points to media and wants to probe a chat/vision model: "attack this vision model", "run these prompts with the images in `./imgs`", "apply an image transform on the images", "test this voice model with the audio in this folder", "visual prompt injection", "typographic jailbreak". Pass `image_dir`/`audio_dir`/`video_dir` for folders or `image_paths`/`audio_paths`/`video_paths` for explicit files. Do NOT confuse with `generate_image_attack` (classifier evasion, not chat).
Expand Down Expand Up @@ -147,7 +148,8 @@ The AI Red Teaming capability provides these tools:

- **generate_attack** — Generate + auto-execute an attack workflow (single, campaign, or transform study)
- **generate_category_attack** — Generate + auto-execute a category-based assessment from bundled goals
- **generate_agentic_attack** — Generate + auto-execute an attack against an HTTP agent API
- **generate_agentic_attack** — Generate + auto-execute a single attack against an HTTP agent API
- **generate_agentic_suite_attack** — Generate + auto-execute the FULL agentic suite against an HTTP agent API: every OWASP-ASI category, auto-selecting the mapped attacks + family transforms (MCP, multi-agent, reasoning, exfiltration, …) + detection scorers. The "run all possible attacks on my agent" path — no need to name individual attacks; omit `categories` for everything.
- **generate_atlas_attack** — Generate + auto-execute an ATLAS multi-agent campaign (Adaptive Topology-Level Attack Synthesis) against a deployed multi-agent environment. Runs a Probe → Route → Learn loop over a budget of episodes, driving GOAT/Crescendo through three injection surfaces (direct / tool_output / peer_message) and gating success on *real tool execution*. Use for multi-agent systems with delegation chains and trust boundaries.
- **generate_image_attack** — Generate + auto-execute a traditional ML adversarial attack (HopSkipJump, SimBA, NES, ZOO) against an image classifier endpoint
- **generate_multimodal_attack** — Generate + auto-execute a MULTIMODAL LLM red teaming attack: send text + image/audio/video to a vision/audio-capable model, apply modality-typed transforms, score the text response for jailbreak success
Expand Down
163 changes: 163 additions & 0 deletions capabilities/ai-red-teaming/scripts/attack_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -7170,12 +7170,175 @@ def _finalize_prediction_workflow(script: str, filename: str, params: dict, desc
return {"result": "\n".join(result_lines), "filename": filename, "filepath": str(filepath)}


_AGENTIC_SUITE_BODY = '''
async def main():
output_dir = Path.home() / "workspace" / "airt"
output_dir.mkdir(parents=True, exist_ok=True)

assessment = Assessment(
ASSESSMENT_NAME,
target_model=TARGET_MODEL,
model=ATTACKER_MODEL,
attacker_model=ATTACKER_MODEL,
judge_model=JUDGE_MODEL,
attack_manifest=[{"attack": "agentic_suite"}],
)
await assessment.register()
print(f"Assessment registered: {assessment.assessment_id or 'local-only'}")
print("Mode: Agentic Suite (all OWASP-ASI categories)")
print(f"Agent URL: {AGENT_URL}")
sys.stdout.flush()

categories = None
if CATEGORIES:
categories = [OWASPAgenticCategory(c) for c in CATEGORIES]

async with assessment.trace():
try:
results = await run_agentic_suite(
assessment, target=target, goal=GOAL, categories=categories
)
summary = {}
for cat, attacks in results.items():
ran = [name for name, entry in attacks.items() if "error" not in entry]
summary[cat] = {"attacks_run": ran, "errors": len(attacks) - len(ran)}
print("\\n--- AGENTIC SUITE RESULTS ---")
print(json.dumps(summary, indent=2))
sys.stdout.flush()
except Exception as e:
print(f"\\nERROR: {e}")
traceback.print_exc()
await assessment.fail(str(e))
sys.exit(1)

_write_local_analytics(
assessment, target_model=TARGET_MODEL, attacker_model=ATTACKER_MODEL, evaluator_model=JUDGE_MODEL
)
print("\\nAssessment complete.")
sys.stdout.flush()

asyncio.run(main())

try:
dn.shutdown()
except Exception:
pass
'''


def _generate_agentic_suite_script(config: dict, agent_config: dict) -> str:
"""Assemble a workflow that runs the full agentic suite against an agent API."""
imports = chr(10).join(
[
"import asyncio",
"import json",
"import os",
"import sys",
"import traceback",
"from pathlib import Path",
"",
"import dreadnode as dn",
"from dreadnode import task",
"from dreadnode.generators.generator import get_generator, GenerateParams",
"from dreadnode.airt import run_agentic_suite",
"from dreadnode.airt.assessment import Assessment",
"from dreadnode.airt.analytics import analyze",
"from dreadnode.airt.compliance.owasp_agentic import OWASPAgenticCategory",
]
)
constants = chr(10).join(
[
"GOAL = {!r}".format(config["goal"]),
"AGENT_URL = {!r}".format(agent_config["agent_url"]),
"TARGET_MODEL = {!r}".format(config["target_model"]),
"ATTACKER_MODEL = {!r}".format(config["attacker_model"]),
"JUDGE_MODEL = {!r}".format(config["evaluator_model"]),
"ASSESSMENT_NAME = {!r}".format(config["assessment_name"]),
"CATEGORIES = {!r}".format(config.get("categories") or None),
]
)
parts = [
imports,
_build_configure(),
_build_analytics_writer(),
constants,
"",
_build_agent_target_code(agent_config),
_AGENTIC_SUITE_BODY,
]
return chr(10).join(parts)


def generate_agentic_suite(params: dict) -> dict:
"""Generate a workflow that red-teams an agent with the FULL agentic suite.

Runs every OWASP-ASI category the SDK map covers (auto-selecting attacks,
family transforms, and detection scorers) against the agent endpoint — this is
the "run all possible attacks on my agent" path.
"""
goal = params.get("goal", "")
agent_url = params.get("agent_url", "")
attacker_model = params.get("attacker_model")
evaluator_model = params.get("evaluator_model")
if not goal:
return {"error": "goal is required"}
if not agent_url:
return {"error": "agent_url is required — the HTTP endpoint of the agent to red-team"}
if not attacker_model:
return {"error": "attacker_model is required (the LLM that generates adversarial prompts)"}

preset = _AGENT_PRESETS.get(params.get("agent_preset", "custom"), _AGENT_PRESETS["custom"])
agent_config = {
"agent_url": agent_url,
"agent_auth_type": params.get("agent_auth_type", "none"),
"agent_auth_env_var": params.get("agent_auth_env_var", "AGENT_API_KEY"),
"agent_request_template": params.get("agent_request_template") or preset["request_template"],
"agent_response_text_path": params.get("agent_response_text_path") or preset["response_text_path"],
"agent_response_tool_calls_path": params.get("agent_response_tool_calls_path")
or preset["response_tool_calls_path"],
"agent_dangerous_tools": params.get("agent_dangerous_tools", []),
"agent_safe_tools": params.get("agent_safe_tools", []),
}
resolved_attacker = _resolve_model(attacker_model)
resolved_eval = _resolve_model(evaluator_model) if evaluator_model else resolved_attacker

config = {
"goal": goal,
"target_model": "agent://{}".format(agent_url.split("//")[-1].split("/")[0]),
"attacker_model": resolved_attacker,
"evaluator_model": resolved_eval,
"assessment_name": _safe_str(params.get("assessment_name") or "Agentic Suite Assessment"),
"categories": params.get("categories"),
}

script = _generate_agentic_suite_script(config, agent_config)

try:
compile(script, "<agentic_suite>", "exec")
except SyntaxError as e:
return {"error": "generated script has a syntax error: {}".format(e)}

timestamp = time.strftime("%Y%m%d_%H%M%S")
filename = "agentic_suite_{}.py".format(timestamp)
WORKFLOWS_DIR.mkdir(parents=True, exist_ok=True)
filepath = WORKFLOWS_DIR / filename
filepath.write_text(script)
return {
"result": "Agentic suite workflow generated (all OWASP-ASI categories).\\n\\nFile: {}".format(
filepath
),
"filename": filename,
"filepath": str(filepath),
}


# stdin/stdout JSON dispatch

METHODS = {
"generate_attack": generate_attack,
"generate_category_attack": generate_category_attack,
"generate_agentic_attack": generate_agentic_attack,
"generate_agentic_suite": generate_agentic_suite,
"generate_atlas_attack": generate_atlas_attack,
"generate_image_attack": generate_image_attack,
"generate_tabular_attack": generate_tabular_attack,
Expand Down
14 changes: 14 additions & 0 deletions capabilities/ai-red-teaming/tests/test_attack_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -357,6 +357,20 @@ def test_with_scorer(self) -> None:
)
assert "error" not in result

def test_agentic_suite(self) -> None:
result = _generate_method(
"generate_agentic_suite",
{
"goal": "red team my agent",
"agent_url": "http://localhost:8100/attack",
"attacker_model": "groq",
},
)
assert "error" not in result
script = Path(result["filepath"]).read_text()
assert "run_agentic_suite(" in script
assert "def target(" in script

def test_with_goal_category(self) -> None:
result = _generate(
{
Expand Down
61 changes: 60 additions & 1 deletion capabilities/ai-red-teaming/tools/attacks.py
Original file line number Diff line number Diff line change
Expand Up @@ -299,6 +299,62 @@ def generate_agentic_attack(
return _call_runner("generate_agentic_attack", params)


def generate_agentic_suite_attack(
goal: t.Annotated[str, "Overall red-team goal for the agent"],
agent_url: t.Annotated[str, "HTTP endpoint of the target agent"],
attacker_model: t.Annotated[str, "LLM generating attack prompts"],
categories: t.Annotated[
list[str] | None,
"OWASP-ASI category values to run (e.g. 'agentic_asi02_tool_misuse'); "
"omit to run ALL categories — the 'run all possible attacks' path.",
] = None,
agent_auth_type: t.Annotated[str, "Auth scheme: 'none', 'bearer', or 'api_key'"] = "none",
agent_auth_env_var: t.Annotated[str, "Env var name for auth credential"] = "AGENT_API_KEY",
agent_request_template: t.Annotated[str, "JSON request template with {prompt} placeholder"] = "",
agent_response_text_path: t.Annotated[str, "JSONPath to extract response text"] = "",
agent_response_tool_calls_path: t.Annotated[str, "JSONPath for tool calls in response"] = "",
agent_dangerous_tools: t.Annotated[list[str] | None, "Dangerous tool names for scoring"] = None,
agent_safe_tools: t.Annotated[list[str] | None, "Safe tool whitelist for scoring"] = None,
agent_preset: t.Annotated[str, "Preset: 'openai_assistants', 'anthropic', or 'custom'"] = "custom",
evaluator_model: t.Annotated[str, "Judge LLM"] = "",
assessment_name: t.Annotated[str, "Assessment name"] = "",
) -> str:
"""Red-team an agent with the FULL agentic suite (all OWASP-ASI categories).

This is the "run all possible attacks on my agent" path: it drives every ASI
category the SDK map covers, auto-selecting the mapped attacks, family
transforms, and detection scorers — no need to name individual attacks. Point
it at the agent's HTTP endpoint (with a preset or custom request/response
template); omit ``categories`` to run everything.
"""
params: dict[str, t.Any] = {
"goal": goal,
"agent_url": agent_url,
"attacker_model": attacker_model,
"agent_auth_type": agent_auth_type,
"agent_auth_env_var": agent_auth_env_var,
"agent_preset": agent_preset,
}
if categories:
params["categories"] = categories
if agent_request_template:
params["agent_request_template"] = agent_request_template
if agent_response_text_path:
params["agent_response_text_path"] = agent_response_text_path
if agent_response_tool_calls_path:
params["agent_response_tool_calls_path"] = agent_response_tool_calls_path
if agent_dangerous_tools:
params["agent_dangerous_tools"] = agent_dangerous_tools
if agent_safe_tools:
params["agent_safe_tools"] = agent_safe_tools
if evaluator_model:
params["evaluator_model"] = evaluator_model
if assessment_name:
params["assessment_name"] = assessment_name

return _call_runner("generate_agentic_suite", params)


@safe_tool
def generate_atlas_attack(
agent_url: t.Annotated[str, "HTTP /attack endpoint of the deployed multi-agent environment"],
Expand All @@ -309,7 +365,10 @@ def generate_atlas_attack(
"CB, DE, GH, RP, MP. Defaults to one objective per category if omitted.",
] = None,
scenario_name: t.Annotated[
str, "Scenario for scenario-specific probes: finops, devsecops, healthcare, soc"
str,
"Scenario for scenario-specific probes and the shipped argument-aware "
"policy: finops, devsecops, healthcare, soc, devops (RCE), support (exfil). "
"The environment-derived name also works (e.g. 'devops-rce', 'support-exfil').",
] = "",
total_budget: t.Annotated[int, "Total attack episodes across the campaign"] = 64,
evaluator_model: t.Annotated[str, "Judge LLM (defaults to attacker_model)"] = "",
Expand Down
6 changes: 4 additions & 2 deletions capabilities/ai-red-teaming/tools/environments.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,10 @@

Tools that let the AI red-teaming agent provision a hosted **task environment**
(e.g. the ``finops-mesh`` / ``devsecops-mesh`` / ``healthcare-mesh`` /
``soc-mesh`` multi-agent systems) and target it with ATLAS — closing the loop
between the Environments the platform hosts and ``generate_atlas_attack``.
``soc-mesh`` tool-misuse pipelines, ``devops-rce-mesh`` for real code execution,
and ``support-exfil-mesh`` for data exfiltration) and target it with ATLAS —
closing the loop between the Environments the platform hosts and
``generate_atlas_attack``.

Provisioning uses the SDK's ``TaskEnvironment`` (platform Docker/E2B sandbox
provider). The model the environment's agents use is passed in via
Expand Down
Loading