From 4d5d5dfdb11225e7b8a32490590c3fb1c161bba9 Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Wed, 15 Jul 2026 18:58:42 -0700 Subject: [PATCH 1/7] fix(ai-red-teaming): import get_generator in generated agentic script Same missing-import bug as the ATLAS template (#100), in the separate `_build_agentic_imports`: generated agentic-attack scripts call get_generator (proxy-routing block) but never imported it, raising NameError at runtime. Add the import and a TestAgenticGeneration regression test (the agentic generator was previously untested). --- .../ai-red-teaming/scripts/attack_runner.py | 2 ++ .../tests/test_attack_runner.py | 27 +++++++++++++++++++ 2 files changed, 29 insertions(+) diff --git a/capabilities/ai-red-teaming/scripts/attack_runner.py b/capabilities/ai-red-teaming/scripts/attack_runner.py index 7ede610..9dcb463 100644 --- a/capabilities/ai-red-teaming/scripts/attack_runner.py +++ b/capabilities/ai-red-teaming/scripts/attack_runner.py @@ -4324,6 +4324,8 @@ def _build_agentic_imports(attacks: list[dict], transforms: list[dict], has_scor "", "import dreadnode as dn", "from dreadnode import task", + # Required by the proxy-routing block (get_generator/GenerateParams). + "from dreadnode.generators.generator import get_generator, GenerateParams", ] attack_funcs = set() diff --git a/capabilities/ai-red-teaming/tests/test_attack_runner.py b/capabilities/ai-red-teaming/tests/test_attack_runner.py index c30f923..0786b9b 100644 --- a/capabilities/ai-red-teaming/tests/test_attack_runner.py +++ b/capabilities/ai-red-teaming/tests/test_attack_runner.py @@ -1039,6 +1039,33 @@ def test_bearer_auth_injects_header(self): assert "Authorization" in script +_AGENTIC_BASE = { + "attack_type": "goat", + "goal": "get the agent to misuse a privileged tool", + "agent_url": "http://localhost:8000/attack", + "agent_preset": "custom", + "attacker_model": "groq scout", + "n_iterations": 4, + "generate_only": True, +} + + +class TestAgenticGeneration: + def test_generated_script_compiles_and_imports_get_generator(self): + result = _generate_method("generate_agentic_attack", _AGENTIC_BASE) + assert "error" not in result, result.get("error") + script = Path(result["filepath"]).read_text() + compile(script, result["filepath"], "exec") + # Regression: the proxy-routing block calls get_generator(...), so the + # generated script MUST import it (compile() only checks syntax, so a + # missing import is a runtime NameError). + assert "get_generator(" in script + assert ( + "from dreadnode.generators.generator import get_generator, GenerateParams" + in script + ) + + class TestAtlasValidation: def test_missing_agent_url_errors(self): params = {k: v for k, v in _ATLAS_BASE.items() if k != "agent_url"} From 331d1c4b98443728115f4012be5e4055255fc01c Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Sun, 19 Jul 2026 21:42:36 -0700 Subject: [PATCH 2/7] feat(airt): add model-extraction & membership-inference attack generators Adds generate_extraction_attack and generate_membership_attack to attack_runner. They build workflow scripts that construct a PredictionTargetSpec from a target's predict endpoint (or an environment challenge_url) and run the SDK's extraction (equation_solving/jacobian/copycat/knockoff) or membership (threshold/label_only) attacks under an Assessment, exporting airt spans to the platform. Verified: a generated knockoff-extraction workflow ran against a live fraud target (fidelity 0.93, 500 queries) and completed its assessment on the local platform. --- .../ai-red-teaming/scripts/attack_runner.py | 378 ++++++++++++++++++ 1 file changed, 378 insertions(+) diff --git a/capabilities/ai-red-teaming/scripts/attack_runner.py b/capabilities/ai-red-teaming/scripts/attack_runner.py index 9dcb463..bf10db7 100644 --- a/capabilities/ai-red-teaming/scripts/attack_runner.py +++ b/capabilities/ai-red-teaming/scripts/attack_runner.py @@ -6633,6 +6633,382 @@ async def main(): return {"result": "\n".join(result_lines), "filename": filename, "filepath": str(filepath)} +# Model-extraction & membership-inference attacks (traditional-ML privacy). +# These query a classifier's predict API (PredictionTargetSpec) rather than perturbing +# a single input, so they run the attack object directly (it carries airt_assessment_id) +# inside a dn.run context — no Study/objective. +_EXTRACTION_ATTACK_MAP = { + "equation_solving": "equation_solving_extraction", + "jacobian": "jacobian_extraction", + "copycat": "copycat_extraction", + "knockoff": "knockoff_extraction", +} +_MEMBERSHIP_ATTACK_MAP = { + "threshold": "threshold_membership", + "label_only": "label_only_membership", +} + + +def _build_prediction_imports(func_names: list[str]) -> str: + lines = [ + "import asyncio", + "import os", + "import sys", + "import traceback", + "", + "import httpx", + "import dreadnode as dn", + "from dreadnode.airt import PredictionTargetSpec, TargetAuth", + "from dreadnode.airt import {}".format(", ".join(func_names)), + "from dreadnode.airt.assessment import Assessment", + ] + return "\n".join(lines) + + +def generate_extraction_attack(params: dict) -> dict: + """Generate a workflow that steals a classifier via black-box queries. + + Requires: api_url (predict endpoint) and a query pool (pool_url or query_pool). + """ + attack_type = params.get("attack_type", "knockoff") + api_url = params.get("api_url", "") + api_key = params.get("api_key", "") + pool_url = params.get("pool_url", "") + query_pool = params.get("query_pool", []) + request_template = params.get("request_template", '{"features": {input}}') + probabilities_path = params.get("probabilities_path", "$.probabilities") + input_format = params.get("input_format", "json_array") + num_classes = int(params.get("num_classes", 2)) + query_budget = int(params.get("query_budget", 1000)) + modality = params.get("modality", "tabular") + goal = params.get("goal", "Steal model functionality via API queries") + goal_category = params.get("goal_category", "model_extraction") + assessment_name = params.get("assessment_name", "") + + if not api_url: + return {"error": "api_url is required (target classifier predict endpoint)"} + if not pool_url and not query_pool: + return {"error": "pool_url or query_pool is required (extraction query inputs)"} + + key = attack_type.strip().lower().replace("-", "_").replace(" ", "_") + func = _EXTRACTION_ATTACK_MAP.get(key) + if not func: + return { + "error": "Unknown extraction attack '{}'. Available: {}".format( + attack_type, ", ".join(sorted(_EXTRACTION_ATTACK_MAP)) + ) + } + + timestamp = time.strftime("%Y%m%d_%H%M%S") + filename = "extraction_{}_{}.py".format(key, timestamp) + assessment_name = assessment_name or "Model Extraction ({})".format(key) + imports = _build_prediction_imports([func]) + configure = _build_configure() + analytics_writer = _build_analytics_writer() + + script = '''{imports} + +{configure} + +{analytics_writer} + +API_URL = "{api_url}" +API_KEY = "{api_key}" +POOL_URL = "{pool_url}" +QUERY_POOL = {query_pool} +NUM_CLASSES = {num_classes} +QUERY_BUDGET = {query_budget} + +if API_KEY: + os.environ["TARGET_API_KEY"] = API_KEY + + +async def main(): + async with httpx.AsyncClient(timeout=60) as _c: + pool = (await _c.get(POOL_URL)).json()["inputs"] if POOL_URL else QUERY_POOL + print("Query pool: {{}} inputs".format(len(pool))) + sys.stdout.flush() + + auth = ( + TargetAuth(type="api_key", header="x-api-key", env_var="TARGET_API_KEY") + if API_KEY + else TargetAuth() + ) + spec = PredictionTargetSpec( + endpoint=API_URL, + auth=auth, + request_template={request_template!r}, + probabilities_path={probabilities_path!r}, + input_format={input_format!r}, + num_classes=NUM_CLASSES, + name="ml_classifier", + ) + + assessment = Assessment( + name="{assessment_name}", + description="Model extraction: {func} on {{}}".format(API_URL), + workflow_run_id="{filename}", + target_config={{"url": API_URL, "type": "ml_classifier"}}, + attacker_config={{"attack": "{func}"}}, + attack_manifest=[{{"attack": "{func}", "domain": "model_extraction", "input_modality": "{modality}"}}], + ) + await assessment.register() + try: + with dn.run("{assessment_name}"): + attack = {func}( + spec, + query_pool=pool, + query_budget=QUERY_BUDGET, + num_classes=NUM_CLASSES, + modality="{modality}", + measure_transfer=False, + airt_assessment_id=assessment.assessment_id, + airt_target_model="ml_classifier", + ) + result = await attack.run() + print("--- RESULTS ---") + print(" Strategy: {{}}".format(result.strategy)) + print(" Fidelity: {{:.4f}}".format(result.fidelity)) + print(" Agreement: {{:.4f}}".format(result.agreement_rate)) + print(" Queries: {{}}".format(result.query_count)) + print("--- end ---") + sys.stdout.flush() + await assessment.complete() + except Exception as e: + await assessment.fail(str(e)) + raise + + _write_local_analytics(assessment) + print("Assessment complete.") + sys.stdout.flush() + + +asyncio.run(main()) + +try: + dn.shutdown() +except Exception: + pass +'''.format( + imports=imports, + configure=configure, + analytics_writer=analytics_writer, + api_url=_safe_str(api_url), + api_key=_safe_str(api_key), + pool_url=_safe_str(pool_url), + query_pool=repr(query_pool), + num_classes=num_classes, + query_budget=query_budget, + request_template=request_template, + probabilities_path=probabilities_path, + input_format=input_format, + modality=_safe_str(modality), + func=func, + assessment_name=_safe_str(assessment_name), + filename=_safe_str(filename), + ) + + return _finalize_prediction_workflow( + script, filename, params, "Model Extraction: {} vs {}".format(func, api_url) + ) + + +def generate_membership_attack(params: dict) -> dict: + """Generate a workflow that infers training-set membership from a classifier. + + Requires: api_url and member/non-member records (via *_url or inline). + """ + attack_type = params.get("attack_type", "threshold") + api_url = params.get("api_url", "") + api_key = params.get("api_key", "") + members_url = params.get("members_url", "") + nonmembers_url = params.get("nonmembers_url", "") + members = params.get("members", []) + nonmembers = params.get("nonmembers", []) + request_template = params.get("request_template", '{"features": {input}}') + probabilities_path = params.get("probabilities_path", "$.probabilities") + input_format = params.get("input_format", "json_array") + num_classes = int(params.get("num_classes", 2)) + signal = params.get("signal", "confidence") + modality = params.get("modality", "tabular") + goal_category = params.get("goal_category", "membership_inference") + assessment_name = params.get("assessment_name", "") + + if not api_url: + return {"error": "api_url is required (target classifier predict endpoint)"} + if not (members_url or members) or not (nonmembers_url or nonmembers): + return {"error": "members/nonmembers (or *_url) are required for membership scoring"} + + key = attack_type.strip().lower().replace("-", "_").replace(" ", "_") + func = _MEMBERSHIP_ATTACK_MAP.get(key) + if not func: + return { + "error": "Unknown membership attack '{}'. Available: {}".format( + attack_type, ", ".join(sorted(_MEMBERSHIP_ATTACK_MAP)) + ) + } + + timestamp = time.strftime("%Y%m%d_%H%M%S") + filename = "membership_{}_{}.py".format(key, timestamp) + assessment_name = assessment_name or "Membership Inference ({})".format(key) + imports = _build_prediction_imports([func]) + configure = _build_configure() + analytics_writer = _build_analytics_writer() + + script = '''{imports} + +{configure} + +{analytics_writer} + +API_URL = "{api_url}" +API_KEY = "{api_key}" +MEMBERS_URL = "{members_url}" +NONMEMBERS_URL = "{nonmembers_url}" +MEMBERS = {members} +NONMEMBERS = {nonmembers} +NUM_CLASSES = {num_classes} + +if API_KEY: + os.environ["TARGET_API_KEY"] = API_KEY + + +async def main(): + async with httpx.AsyncClient(timeout=60) as _c: + m = (await _c.get(MEMBERS_URL)).json() if MEMBERS_URL else {{"records": MEMBERS, "labels": None}} + nm = (await _c.get(NONMEMBERS_URL)).json() if NONMEMBERS_URL else {{"records": NONMEMBERS, "labels": None}} + print("Members: {{}} Non-members: {{}}".format(len(m["records"]), len(nm["records"]))) + sys.stdout.flush() + + auth = ( + TargetAuth(type="api_key", header="x-api-key", env_var="TARGET_API_KEY") + if API_KEY + else TargetAuth() + ) + spec = PredictionTargetSpec( + endpoint=API_URL, + auth=auth, + request_template={request_template!r}, + probabilities_path={probabilities_path!r}, + input_format={input_format!r}, + num_classes=NUM_CLASSES, + name="ml_classifier", + ) + + assessment = Assessment( + name="{assessment_name}", + description="Membership inference: {func} on {{}}".format(API_URL), + workflow_run_id="{filename}", + target_config={{"url": API_URL, "type": "ml_classifier"}}, + attacker_config={{"attack": "{func}", "signal": "{signal}"}}, + attack_manifest=[{{"attack": "{func}", "domain": "membership_inference", "input_modality": "{modality}"}}], + ) + await assessment.register() + try: + with dn.run("{assessment_name}"): + attack = {func}( + spec, + members=m["records"], + nonmembers=nm["records"], + member_labels=m.get("labels"), + nonmember_labels=nm.get("labels"), + signal="{signal}", + modality="{modality}", + airt_assessment_id=assessment.assessment_id, + airt_target_model="ml_classifier", + ) + result = await attack.run() + print("--- RESULTS ---") + print(" Method: {{}}".format(result.method)) + print(" AUC: {{:.4f}}".format(result.auc)) + print(" TPR@1%FPR: {{:.4f}}".format(result.tpr_at_1pct_fpr)) + print(" Advantage: {{:.4f}}".format(result.advantage)) + print(" Re-identified: {{}}".format(result.records_reidentified)) + print("--- end ---") + sys.stdout.flush() + await assessment.complete() + except Exception as e: + await assessment.fail(str(e)) + raise + + _write_local_analytics(assessment) + print("Assessment complete.") + sys.stdout.flush() + + +asyncio.run(main()) + +try: + dn.shutdown() +except Exception: + pass +'''.format( + imports=imports, + configure=configure, + analytics_writer=analytics_writer, + api_url=_safe_str(api_url), + api_key=_safe_str(api_key), + members_url=_safe_str(members_url), + nonmembers_url=_safe_str(nonmembers_url), + members=repr(members), + nonmembers=repr(nonmembers), + num_classes=num_classes, + request_template=request_template, + probabilities_path=probabilities_path, + input_format=input_format, + signal=_safe_str(signal), + modality=_safe_str(modality), + func=func, + assessment_name=_safe_str(assessment_name), + filename=_safe_str(filename), + ) + + return _finalize_prediction_workflow( + script, filename, params, "Membership Inference: {} vs {}".format(func, api_url) + ) + + +def _finalize_prediction_workflow(script: str, filename: str, params: dict, description: str) -> dict: + """Syntax-check, persist, and (unless generate_only) execute a generated workflow.""" + try: + compile(script, filename, "exec") + except SyntaxError as e: + return { + "error": "Generated script has syntax error: {} (line {}). This is a bug in the tool.".format( + e.msg, e.lineno + ) + } + + filepath, filename = _unique_workflow_path(filename) + filepath.write_text(script) + + metadata = {} + if METADATA_FILE.exists(): + try: + metadata = json.loads(METADATA_FILE.read_text()) + except Exception: + pass + metadata[filename] = { + "description": description, + "saved_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "size_bytes": len(script.encode()), + } + METADATA_FILE.write_text(json.dumps(metadata, indent=2)) + + result_lines = [ + "{} workflow generated and saved.".format(description), + "", + "File: {}".format(filepath), + "Workflow filename: {}".format(filename), + "", + '>>> NEXT STEP: call execute_workflow(filename="{}") to run this attack <<<'.format(filename), + ] + if not params.get("generate_only"): + result_lines.append(_auto_execute_workflow(filename)) + + return {"result": "\n".join(result_lines), "filename": filename, "filepath": str(filepath)} + + # stdin/stdout JSON dispatch METHODS = { @@ -6644,6 +7020,8 @@ async def main(): "generate_tabular_attack": generate_tabular_attack, "generate_multimodal_attack": generate_multimodal_attack, "generate_multimodal_category_attack": generate_multimodal_category_attack, + "generate_extraction_attack": generate_extraction_attack, + "generate_membership_attack": generate_membership_attack, } From e04ccd1d22a3d7c0329da497d79897a29784f337 Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Mon, 20 Jul 2026 08:46:19 -0700 Subject: [PATCH 3/7] feat(airt): expose extraction/membership attacks as agent tools Add generate_extraction_attack and generate_membership_attack @safe_tool wrappers so the AIRT agent (and TUI) can run the new traditional-ML privacy attacks, mirroring the existing generate_image_attack wrapper over the attack_runner generators. --- capabilities/ai-red-teaming/tools/attacks.py | 106 +++++++++++++++++++ 1 file changed, 106 insertions(+) diff --git a/capabilities/ai-red-teaming/tools/attacks.py b/capabilities/ai-red-teaming/tools/attacks.py index 249a985..3cca71e 100644 --- a/capabilities/ai-red-teaming/tools/attacks.py +++ b/capabilities/ai-red-teaming/tools/attacks.py @@ -448,6 +448,112 @@ def generate_image_attack( return _call_runner("generate_image_attack", params) +@safe_tool +def generate_extraction_attack( + attack_type: t.Annotated[ + str, + "Model-extraction attack: equation_solving (exact for linear models), " + "jacobian, copycat (hard-label), or knockoff (soft-label, highest fidelity).", + ] = "knockoff", + api_url: t.Annotated[str, "Target classifier predict endpoint (POST)."] = "", + api_key: t.Annotated[str, "API key for the x-api-key header (optional)."] = "", + pool_url: t.Annotated[ + str, "GET endpoint returning {inputs: [...]} — the unlabeled query pool." + ] = "", + query_pool: t.Annotated[ + list | None, "Inline query pool (used if pool_url is not given)." + ] = None, + request_template: t.Annotated[ + str, "Request body with a single {input} placeholder." + ] = '{"features": {input}}', + probabilities_path: t.Annotated[str, "JSONPath to the probability vector."] = "$.probabilities", + input_format: t.Annotated[str, "json_array | image_b64 | text."] = "json_array", + num_classes: t.Annotated[int, "Number of classes."] = 2, + query_budget: t.Annotated[int, "Max target queries."] = 1000, + modality: t.Annotated[str, "tabular | image | text."] = "tabular", + assessment_name: t.Annotated[str, "Human-readable assessment name."] = "", +) -> str: + """Steal a classifier's decision boundary via black-box queries. + + Trains a surrogate on the target's predictions and reports fidelity/agreement. + Provide api_url and a query pool (pool_url or query_pool). Results appear in the + platform under AI Red Teaming with model-extraction metrics. + """ + params: dict[str, t.Any] = { + "attack_type": attack_type, + "api_url": api_url, + "num_classes": num_classes, + "query_budget": query_budget, + "modality": modality, + "request_template": request_template, + "probabilities_path": probabilities_path, + "input_format": input_format, + } + if api_key: + params["api_key"] = api_key + if pool_url: + params["pool_url"] = pool_url + if query_pool: + params["query_pool"] = query_pool + if assessment_name: + params["assessment_name"] = assessment_name + return _call_runner("generate_extraction_attack", params) + + +@safe_tool +def generate_membership_attack( + attack_type: t.Annotated[ + str, "Membership-inference attack: threshold (confidence/loss) or label_only." + ] = "threshold", + api_url: t.Annotated[str, "Target classifier predict endpoint (POST)."] = "", + api_key: t.Annotated[str, "API key for the x-api-key header (optional)."] = "", + members_url: t.Annotated[ + str, "GET endpoint returning {records: [...], labels: [...]} for training members." + ] = "", + nonmembers_url: t.Annotated[str, "GET endpoint for held-out non-members."] = "", + members: t.Annotated[list | None, "Inline member records (if no members_url)."] = None, + nonmembers: t.Annotated[list | None, "Inline non-member records (if no nonmembers_url)."] = None, + request_template: t.Annotated[ + str, "Request body with a single {input} placeholder." + ] = '{"features": {input}}', + probabilities_path: t.Annotated[str, "JSONPath to the probability vector."] = "$.probabilities", + input_format: t.Annotated[str, "json_array | image_b64 | text."] = "json_array", + num_classes: t.Annotated[int, "Number of classes."] = 2, + signal: t.Annotated[str, "Threshold signal: confidence | entropy | loss."] = "confidence", + modality: t.Annotated[str, "tabular | image | text."] = "tabular", + assessment_name: t.Annotated[str, "Human-readable assessment name."] = "", +) -> str: + """Determine whether records were in the target's training set. + + Reports AUC, TPR@1%FPR, advantage, and count re-identified. Provide api_url and + member/non-member records (via *_url or inline). Results appear in the platform + under AI Red Teaming with membership-inference metrics. + """ + params: dict[str, t.Any] = { + "attack_type": attack_type, + "api_url": api_url, + "num_classes": num_classes, + "signal": signal, + "modality": modality, + "request_template": request_template, + "probabilities_path": probabilities_path, + "input_format": input_format, + } + if api_key: + params["api_key"] = api_key + if members_url: + params["members_url"] = members_url + if nonmembers_url: + params["nonmembers_url"] = nonmembers_url + if members: + params["members"] = members + if nonmembers: + params["nonmembers"] = nonmembers + if assessment_name: + params["assessment_name"] = assessment_name + return _call_runner("generate_membership_attack", params) + + @safe_tool def generate_multimodal_attack( goal: t.Annotated[ From 69a424c77885d4c9a71de80cb8fc7ba37f9ddf77 Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Mon, 20 Jul 2026 21:52:59 -0700 Subject: [PATCH 4/7] feat(airt): add model-evasion attack generator and bump to 1.9.0 Wire the boundary (numeric L2/Linf decision search) and text (token-flip) evasion attacks into the workflow runner and expose generate_evasion_attack as an agent tool. Bump capability to 1.9.0 and note traditional black-box ML coverage (extraction, membership inference, evasion) in the description. --- capabilities/ai-red-teaming/capability.yaml | 10 +- .../ai-red-teaming/scripts/attack_runner.py | 145 ++++++++++++++++++ capabilities/ai-red-teaming/tools/attacks.py | 53 +++++++ 3 files changed, 204 insertions(+), 4 deletions(-) diff --git a/capabilities/ai-red-teaming/capability.yaml b/capabilities/ai-red-teaming/capability.yaml index 0d76d3c..968c2aa 100644 --- a/capabilities/ai-red-teaming/capability.yaml +++ b/capabilities/ai-red-teaming/capability.yaml @@ -1,6 +1,6 @@ schema: 1 name: ai-red-teaming -version: "1.8.0" +version: "1.9.0" description: > Probe the security and safety of AI applications, agents, and foundation models. Orchestrates adversarial attack workflows to discover vulnerabilities in LLMs, @@ -8,9 +8,11 @@ description: > agents, and custom AI endpoints before they are exploited. Covers jailbreaking, prompt injection, data exfiltration, tool manipulation, reasoning attacks, guardrail bypass, and more — mapped to OWASP LLM Top 10, OWASP ASI01-ASI10, MITRE ATLAS, - and NIST AI RMF compliance frameworks. 45 attack algorithms (41 LLM + 4 adversarial - ML samplers), 500+ transforms, an extensive scorer catalog, and 260 bundled harm - goals across 25 sub-categories in safety, security, and agentic tiers. + and NIST AI RMF compliance frameworks. Also probes traditional black-box ML + classifiers: model extraction (model stealing), membership inference (training-data + leakage), and model evasion (adversarial examples) across tabular, image, and text. + 500+ transforms, an extensive scorer catalog, and 260 bundled harm goals across 25 + sub-categories in safety, security, and agentic tiers. agents: - agents/ diff --git a/capabilities/ai-red-teaming/scripts/attack_runner.py b/capabilities/ai-red-teaming/scripts/attack_runner.py index bf10db7..f5c3fc7 100644 --- a/capabilities/ai-red-teaming/scripts/attack_runner.py +++ b/capabilities/ai-red-teaming/scripts/attack_runner.py @@ -6647,6 +6647,10 @@ async def main(): "threshold": "threshold_membership", "label_only": "label_only_membership", } +_EVASION_ATTACK_MAP = { + "boundary": "boundary_evasion", + "text": "text_evasion", +} def _build_prediction_imports(func_names: list[str]) -> str: @@ -6968,6 +6972,146 @@ async def main(): ) +def generate_evasion_attack(params: dict) -> dict: + """Generate a workflow that crafts an adversarial example to flip a classifier. + + Requires: api_url and an original input (via `original` or sample_url). + """ + attack_type = params.get("attack_type", "boundary") + api_url = params.get("api_url", "") + api_key = params.get("api_key", "") + sample_url = params.get("sample_url", "") + original = params.get("original") + request_template = params.get("request_template", '{"features": {input}}') + probabilities_path = params.get("probabilities_path", "$.probabilities") + input_format = params.get("input_format", "json_array") + num_classes = int(params.get("num_classes", 2)) + max_queries = int(params.get("max_queries", 500)) + norm = params.get("norm", "l2") + modality = params.get("modality", "tabular") + assessment_name = params.get("assessment_name", "") + + if not api_url: + return {"error": "api_url is required (target classifier predict endpoint)"} + if original is None and not sample_url: + return {"error": "an original input (original or sample_url) is required to perturb"} + + key = attack_type.strip().lower().replace("-", "_").replace(" ", "_") + func = _EVASION_ATTACK_MAP.get(key) + if not func: + return { + "error": "Unknown evasion attack '{}'. Available: {}".format( + attack_type, ", ".join(sorted(_EVASION_ATTACK_MAP)) + ) + } + + timestamp = time.strftime("%Y%m%d_%H%M%S") + filename = "evasion_{}_{}.py".format(key, timestamp) + assessment_name = assessment_name or "Model Evasion ({})".format(key) + imports = _build_prediction_imports([func]) + configure = _build_configure() + analytics_writer = _build_analytics_writer() + + script = '''{imports} + +{configure} + +{analytics_writer} + +API_URL = "{api_url}" +API_KEY = "{api_key}" +SAMPLE_URL = "{sample_url}" +ORIGINAL = {original} +NUM_CLASSES = {num_classes} +MAX_QUERIES = {max_queries} + +if API_KEY: + os.environ["TARGET_API_KEY"] = API_KEY + + +async def main(): + original = ORIGINAL + if SAMPLE_URL: + async with httpx.AsyncClient(timeout=60) as _c: + original = (await _c.get(SAMPLE_URL)).json()["inputs"][0] + print("Perturbing one input against {{}}".format(API_URL)) + sys.stdout.flush() + + auth = ( + TargetAuth(type="api_key", header="x-api-key", env_var="TARGET_API_KEY") + if API_KEY + else TargetAuth() + ) + spec = PredictionTargetSpec( + endpoint=API_URL, + auth=auth, + request_template={request_template!r}, + probabilities_path={probabilities_path!r}, + input_format={input_format!r}, + num_classes=NUM_CLASSES, + name="ml_classifier", + ) + + async with Assessment( + name="{assessment_name}", + description="Model evasion: {func} on {{}}".format(API_URL), + workflow_run_id="{filename}", + target_config={{"url": API_URL, "type": "ml_classifier"}}, + attacker_config={{"attack": "{func}"}}, + attack_manifest=[{{"attack": "{func}", "domain": "adversarial_ml", "input_modality": "{modality}"}}], + ) as assessment: + attack = {func}( + spec, + original, + num_classes=NUM_CLASSES, + max_queries=MAX_QUERIES, + {norm_kw}modality="{modality}", + airt_target_model="ml_classifier", + ) + result = await attack.run() + print("--- RESULTS ---") + print(" Success: {{}}".format(result.success)) + print(" Distance: {{:.4f}} ({{}})".format(result.distance_value, result.distance_norm)) + print(" Class flip: {{}} -> {{}}".format(result.original_class, result.adversarial_class)) + print(" Queries: {{}}".format(result.query_count)) + print("--- end ---") + sys.stdout.flush() + _write_local_analytics(assessment) + print("Assessment complete.") + sys.stdout.flush() + + +asyncio.run(main()) + +try: + dn.shutdown() +except Exception: + pass +'''.format( + imports=imports, + configure=configure, + analytics_writer=analytics_writer, + api_url=_safe_str(api_url), + api_key=_safe_str(api_key), + sample_url=_safe_str(sample_url), + original=repr(original), + num_classes=num_classes, + max_queries=max_queries, + request_template=request_template, + probabilities_path=probabilities_path, + input_format=input_format, + norm_kw=('norm="{}", '.format(_safe_str(norm)) if key == "boundary" else ""), + modality=_safe_str(modality), + func=func, + assessment_name=_safe_str(assessment_name), + filename=_safe_str(filename), + ) + + return _finalize_prediction_workflow( + script, filename, params, "Model Evasion: {} vs {}".format(func, api_url) + ) + + def _finalize_prediction_workflow(script: str, filename: str, params: dict, description: str) -> dict: """Syntax-check, persist, and (unless generate_only) execute a generated workflow.""" try: @@ -7022,6 +7166,7 @@ def _finalize_prediction_workflow(script: str, filename: str, params: dict, desc "generate_multimodal_category_attack": generate_multimodal_category_attack, "generate_extraction_attack": generate_extraction_attack, "generate_membership_attack": generate_membership_attack, + "generate_evasion_attack": generate_evasion_attack, } diff --git a/capabilities/ai-red-teaming/tools/attacks.py b/capabilities/ai-red-teaming/tools/attacks.py index 3cca71e..8ad78d7 100644 --- a/capabilities/ai-red-teaming/tools/attacks.py +++ b/capabilities/ai-red-teaming/tools/attacks.py @@ -554,6 +554,59 @@ def generate_membership_attack( return _call_runner("generate_membership_attack", params) +@safe_tool +def generate_evasion_attack( + attack_type: t.Annotated[ + str, "Evasion attack: boundary (numeric L2/Linf decision search) or text (token flip)." + ] = "boundary", + api_url: t.Annotated[str, "Target classifier predict endpoint (POST)."] = "", + api_key: t.Annotated[str, "API key for the x-api-key header (optional)."] = "", + sample_url: t.Annotated[ + str, "GET endpoint returning {inputs: [...]} - the first input is perturbed." + ] = "", + original: t.Annotated[ + object, "Inline original input to perturb (if no sample_url)." + ] = None, + request_template: t.Annotated[ + str, "Request body with a single {input} placeholder." + ] = '{"features": {input}}', + probabilities_path: t.Annotated[str, "JSONPath to the probability vector."] = "$.probabilities", + input_format: t.Annotated[str, "json_array | image_b64 | text."] = "json_array", + num_classes: t.Annotated[int, "Number of classes."] = 2, + max_queries: t.Annotated[int, "Max target queries."] = 500, + norm: t.Annotated[str, "Distance norm for boundary evasion: l2 | linf."] = "l2", + modality: t.Annotated[str, "tabular | image | text."] = "tabular", + assessment_name: t.Annotated[str, "Human-readable assessment name."] = "", +) -> str: + """Craft an adversarial example that flips the target's prediction. + + Reports attack success, perturbation distance (L2/Linf or fraction of tokens), + original -> adversarial class, and query cost. Provide api_url and an original + input (original or sample_url). Results appear in the platform under AI Red + Teaming with model-evasion metrics. + """ + params: dict[str, t.Any] = { + "attack_type": attack_type, + "api_url": api_url, + "num_classes": num_classes, + "max_queries": max_queries, + "norm": norm, + "modality": modality, + "request_template": request_template, + "probabilities_path": probabilities_path, + "input_format": input_format, + } + if api_key: + params["api_key"] = api_key + if sample_url: + params["sample_url"] = sample_url + if original is not None: + params["original"] = original + if assessment_name: + params["assessment_name"] = assessment_name + return _call_runner("generate_evasion_attack", params) + + @safe_tool def generate_multimodal_attack( goal: t.Annotated[ From a1e487acc7ea0ec003fa740b952359b065cedcd5 Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Mon, 20 Jul 2026 22:43:20 -0700 Subject: [PATCH 5/7] feat(airt): add deepwordbug evasion strategy to generator --- capabilities/ai-red-teaming/scripts/attack_runner.py | 1 + 1 file changed, 1 insertion(+) diff --git a/capabilities/ai-red-teaming/scripts/attack_runner.py b/capabilities/ai-red-teaming/scripts/attack_runner.py index f5c3fc7..0182f0d 100644 --- a/capabilities/ai-red-teaming/scripts/attack_runner.py +++ b/capabilities/ai-red-teaming/scripts/attack_runner.py @@ -6650,6 +6650,7 @@ async def main(): _EVASION_ATTACK_MAP = { "boundary": "boundary_evasion", "text": "text_evasion", + "deepwordbug": "deepwordbug_evasion", } From 833ec2625681b3ee2fce2aec15290752bd764c82 Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Mon, 20 Jul 2026 23:39:49 -0700 Subject: [PATCH 6/7] feat(airt): expand traditional-ML attack roster to 1.10.0 Register the new named attacks in the generator dispatch maps: evasion (hopskipjump, simba, square, zoo, textfooler), extraction (activethief, distillation), and membership (lira, shadow_model, entropy, loss). --- capabilities/ai-red-teaming/capability.yaml | 2 +- capabilities/ai-red-teaming/scripts/attack_runner.py | 11 +++++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/capabilities/ai-red-teaming/capability.yaml b/capabilities/ai-red-teaming/capability.yaml index 968c2aa..5e373f7 100644 --- a/capabilities/ai-red-teaming/capability.yaml +++ b/capabilities/ai-red-teaming/capability.yaml @@ -1,6 +1,6 @@ schema: 1 name: ai-red-teaming -version: "1.9.0" +version: "1.10.0" description: > Probe the security and safety of AI applications, agents, and foundation models. Orchestrates adversarial attack workflows to discover vulnerabilities in LLMs, diff --git a/capabilities/ai-red-teaming/scripts/attack_runner.py b/capabilities/ai-red-teaming/scripts/attack_runner.py index 0182f0d..c625db3 100644 --- a/capabilities/ai-red-teaming/scripts/attack_runner.py +++ b/capabilities/ai-red-teaming/scripts/attack_runner.py @@ -6642,15 +6642,26 @@ async def main(): "jacobian": "jacobian_extraction", "copycat": "copycat_extraction", "knockoff": "knockoff_extraction", + "activethief": "activethief_extraction", + "distillation": "distillation_extraction", } _MEMBERSHIP_ATTACK_MAP = { "threshold": "threshold_membership", "label_only": "label_only_membership", + "lira": "lira_membership", + "shadow_model": "shadow_model_membership", + "entropy": "entropy_membership", + "loss": "loss_membership", } _EVASION_ATTACK_MAP = { "boundary": "boundary_evasion", + "hopskipjump": "hopskipjump_evasion", + "simba": "simba_evasion", + "square": "square_evasion", + "zoo": "zoo_evasion", "text": "text_evasion", "deepwordbug": "deepwordbug_evasion", + "textfooler": "textfooler_evasion", } From 67c146ed909a1f984d64f83fd7354ebcdbf4f3a5 Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Mon, 20 Jul 2026 23:53:05 -0700 Subject: [PATCH 7/7] docs(airt): advertise new attack types in generator tool descriptions Update the attack_type parameter descriptions so the agent surfaces the new evasion (hopskipjump/simba/square/zoo/textfooler), extraction (activethief/ distillation), and membership (entropy/loss/shadow_model/lira) options. --- capabilities/ai-red-teaming/tools/attacks.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/capabilities/ai-red-teaming/tools/attacks.py b/capabilities/ai-red-teaming/tools/attacks.py index 8ad78d7..c0b5f98 100644 --- a/capabilities/ai-red-teaming/tools/attacks.py +++ b/capabilities/ai-red-teaming/tools/attacks.py @@ -453,7 +453,8 @@ def generate_extraction_attack( attack_type: t.Annotated[ str, "Model-extraction attack: equation_solving (exact for linear models), " - "jacobian, copycat (hard-label), or knockoff (soft-label, highest fidelity).", + "jacobian, copycat (hard-label), knockoff (soft-label, highest fidelity), " + "activethief (active-learning query selection), or distillation.", ] = "knockoff", api_url: t.Annotated[str, "Target classifier predict endpoint (POST)."] = "", api_key: t.Annotated[str, "API key for the x-api-key header (optional)."] = "", @@ -503,7 +504,9 @@ def generate_extraction_attack( @safe_tool def generate_membership_attack( attack_type: t.Annotated[ - str, "Membership-inference attack: threshold (confidence/loss) or label_only." + str, + "Membership-inference attack: threshold (confidence/entropy/loss), entropy, " + "loss, label_only, shadow_model (Shokri), or lira (likelihood-ratio).", ] = "threshold", api_url: t.Annotated[str, "Target classifier predict endpoint (POST)."] = "", api_key: t.Annotated[str, "API key for the x-api-key header (optional)."] = "", @@ -557,7 +560,9 @@ def generate_membership_attack( @safe_tool def generate_evasion_attack( attack_type: t.Annotated[ - str, "Evasion attack: boundary (numeric L2/Linf decision search) or text (token flip)." + str, + "Evasion attack. Numeric: hopskipjump, boundary, simba, square, zoo. " + "Text: text (word flip), deepwordbug (char typos), textfooler (word swaps).", ] = "boundary", api_url: t.Annotated[str, "Target classifier predict endpoint (POST)."] = "", api_key: t.Annotated[str, "API key for the x-api-key header (optional)."] = "",