From dd659d536fcd98f47d0c706d1cd4ead26b138998 Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Wed, 5 Aug 2026 09:25:07 -0700 Subject: [PATCH 1/2] docs(ai-red-teaming): document devops/support ATLAS scenarios + envs generate_atlas_attack's scenario_name now lists devops (RCE) and support (exfil) and notes the environment-derived names (devops-rce, support-exfil) resolve to the shipped argument-aware policy. environments doc lists the new devops-rce-mesh / support-exfil-mesh targets. --- capabilities/ai-red-teaming/tools/attacks.py | 5 ++++- capabilities/ai-red-teaming/tools/environments.py | 6 ++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/capabilities/ai-red-teaming/tools/attacks.py b/capabilities/ai-red-teaming/tools/attacks.py index c0b5f98..671cd3b 100644 --- a/capabilities/ai-red-teaming/tools/attacks.py +++ b/capabilities/ai-red-teaming/tools/attacks.py @@ -309,7 +309,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)"] = "", diff --git a/capabilities/ai-red-teaming/tools/environments.py b/capabilities/ai-red-teaming/tools/environments.py index 9abda46..21c71ff 100644 --- a/capabilities/ai-red-teaming/tools/environments.py +++ b/capabilities/ai-red-teaming/tools/environments.py @@ -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 From c315fff3f7c9ebf6e264b78c844b987931c4f741 Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Wed, 5 Aug 2026 10:56:01 -0700 Subject: [PATCH 2/2] =?UTF-8?q?feat(ai-red-teaming):=20generate=5Fagentic?= =?UTF-8?q?=5Fsuite=5Fattack=20=E2=80=94=20run=20all=20attacks=20on=20an?= =?UTF-8?q?=20agent?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a turnkey capability tool + workflow generator that runs the FULL agentic suite (every OWASP-ASI category, auto-selecting the mapped attacks, family transforms, and detection scorers via the SDK's run_agentic_suite) against an agent HTTP endpoint — the "run all possible attacks / red team my agent" path, no need to name individual attacks. Wires it into the agent decision tree + tool list so the TUI agent picks it for broad/complete agent audits. Generate-verified + capability test; live TUI run pending SDK release. --- .../agents/ai-red-teaming-agent.md | 6 +- .../ai-red-teaming/scripts/attack_runner.py | 163 ++++++++++++++++++ .../tests/test_attack_runner.py | 14 ++ capabilities/ai-red-teaming/tools/attacks.py | 56 ++++++ 4 files changed, 237 insertions(+), 2 deletions(-) diff --git a/capabilities/ai-red-teaming/agents/ai-red-teaming-agent.md b/capabilities/ai-red-teaming/agents/ai-red-teaming-agent.md index 41e032a..6861c94 100644 --- a/capabilities/ai-red-teaming/agents/ai-red-teaming-agent.md +++ b/capabilities/ai-red-teaming/agents/ai-red-teaming-agent.md @@ -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). @@ -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 diff --git a/capabilities/ai-red-teaming/scripts/attack_runner.py b/capabilities/ai-red-teaming/scripts/attack_runner.py index 8bd9d4a..ee27519 100644 --- a/capabilities/ai-red-teaming/scripts/attack_runner.py +++ b/capabilities/ai-red-teaming/scripts/attack_runner.py @@ -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, "", "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, diff --git a/capabilities/ai-red-teaming/tests/test_attack_runner.py b/capabilities/ai-red-teaming/tests/test_attack_runner.py index db95d00..b5cca7a 100644 --- a/capabilities/ai-red-teaming/tests/test_attack_runner.py +++ b/capabilities/ai-red-teaming/tests/test_attack_runner.py @@ -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( { diff --git a/capabilities/ai-red-teaming/tools/attacks.py b/capabilities/ai-red-teaming/tools/attacks.py index 671cd3b..3c3c4e3 100644 --- a/capabilities/ai-red-teaming/tools/attacks.py +++ b/capabilities/ai-red-teaming/tools/attacks.py @@ -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"],