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 8a21a22..24983ed 100644 --- a/capabilities/ai-red-teaming/agents/ai-red-teaming-agent.md +++ b/capabilities/ai-red-teaming/agents/ai-red-teaming-agent.md @@ -150,6 +150,10 @@ The AI Red Teaming capability provides these tools: - **generate_agentic_attack** — Generate + auto-execute an attack against an HTTP agent API - **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_evasion_attack** — Adversarial evasion against a `/predict` classifier across tabular/image/text (hopskipjump/boundary/simba/square/zoo; text/deepwordbug/textfooler). Needs `api_url` + an `original` input. +- **generate_extraction_attack** — Model extraction / stealing (knockoff/copycat/equation_solving/jacobian/activethief/distillation). Needs `api_url` + a query pool (`pool_url` or `query_pool`). +- **generate_membership_attack** — Membership inference / training-data leakage (threshold/shadow_model/lira/entropy/loss/label_only). Needs `api_url`. +- **generate_inversion_attack** — Model inversion: reconstruct a representative input per class (confidence/nes). Needs `api_url` + `num_classes`; for tabular pass `input_dim` (or `pool_url` to derive it), for image pass `input_shape`. - **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 - **build_media_manifest** — Inventory a folder/list of media into a byte-free reference manifest (kind, mime, size, dimensions) for planning a multimodal attack without loading raw media. Call this first when the user points at a folder of images/audio/video. - **generate_injection_images** — Render attack text (or a CSV of texts) into typographic/visual prompt-injection IMAGES, so you can probe a vision model without the user supplying media. You create the data (render text → images) and pass the paths to generate_multimodal_attack — never view the text. @@ -160,6 +164,13 @@ The AI Red Teaming capability provides these tools: - **list_environments** — List the deployable multi-agent environments (e.g. `finops-mesh`, `devsecops-mesh`, `healthcare-mesh`, `soc-mesh`) that ATLAS can target - **provision_environment** — Deploy a hosted multi-agent environment (passing the model its agents use) and return its `/attack` URL + execute token. Chain into `generate_atlas_attack` to probe it — closing the loop from Environment to ATLAS probe. +**Traditional-ML Targets:** + +- **list_ml_targets** — List Dreadnode-hosted traditional-ML classifiers (`ml-extraction-fraud-tabular`, `ml-extraction-mnist-image`, `ml-extraction-imdb-text`) with their modality/class count. +- **provision_ml_target** — Deploy a hosted classifier and return its `/predict` URL + a seed input + modality/classes, ready to chain into `generate_evasion_attack` / `generate_extraction_attack` / `generate_membership_attack` / `generate_inversion_attack`. + +**Traditional-ML targeting rule:** if the user wants to attack a **Dreadnode** classifier (or names none), call `list_ml_targets` then `provision_ml_target` to get the `/predict` URL + seed — never ask them for a URL. If the user has their **own** classifier, take their `api_url` (+ key + a sample input) and go straight to the attack tool. Pick the tool by goal: flip a prediction → evasion; steal the model → extraction; test training-data leakage → membership; reconstruct class inputs → inversion. + **Workflow Management:** - **execute_workflow** — Run a saved workflow script diff --git a/capabilities/ai-red-teaming/capability.yaml b/capabilities/ai-red-teaming/capability.yaml index ba7e960..2a33b4f 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.10.1" +version: "1.11.0" description: > Probe the security and safety of AI applications, agents, and foundation models. Orchestrates adversarial attack workflows to discover vulnerabilities in LLMs, @@ -10,7 +10,9 @@ description: > bypass, and more — mapped to OWASP LLM Top 10, OWASP ASI01-ASI10, MITRE ATLAS, 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. + leakage), model evasion (adversarial examples), and model inversion (per-class input + reconstruction) across tabular, image, and text — against Dreadnode-hosted targets + (deploy with provision_ml_target) or your own /predict endpoint. 590+ transforms, 140+ scorers, and 260 bundled harm goals across 25 sub-categories in safety, security, and agentic tiers. diff --git a/capabilities/ai-red-teaming/scripts/attack_runner.py b/capabilities/ai-red-teaming/scripts/attack_runner.py index 8bd9d4a..8cf9b6f 100644 --- a/capabilities/ai-red-teaming/scripts/attack_runner.py +++ b/capabilities/ai-red-teaming/scripts/attack_runner.py @@ -6668,6 +6668,10 @@ async def main(): "deepwordbug": "deepwordbug_evasion", "textfooler": "textfooler_evasion", } +_INVERSION_ATTACK_MAP = { + "confidence": "confidence_inversion", + "nes": "nes_inversion", +} def _build_prediction_imports(func_names: list[str]) -> str: @@ -7129,6 +7133,164 @@ async def main(): ) +def generate_inversion_attack(params: dict) -> dict: + """Generate a workflow that reconstructs a representative input per class (model inversion). + + Requires: api_url (predict endpoint) and num_classes. For tabular targets give + input_dim (or pool_url to derive it); for image targets give input_shape. + """ + attack_type = params.get("attack_type", "confidence") + api_url = params.get("api_url", "") + api_key = params.get("api_key", "") + 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)) + input_dim = params.get("input_dim") + input_shape = params.get("input_shape") + pool_url = params.get("pool_url", "") + target_classes = params.get("target_classes") + max_queries = int(params.get("max_queries", 1200)) + seed = int(params.get("seed", 0)) + 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 modality == "image" and not input_shape: + return {"error": "input_shape (e.g. [8, 8]) is required for image model inversion"} + if modality != "image" and input_dim is None and not pool_url: + return {"error": "input_dim (or pool_url to derive it) is required for tabular model inversion"} + + key = attack_type.strip().lower().replace("-", "_").replace(" ", "_") + func = _INVERSION_ATTACK_MAP.get(key) + if not func: + return { + "error": "Unknown inversion attack '{}'. Available: {}".format( + attack_type, ", ".join(sorted(_INVERSION_ATTACK_MAP)) + ) + } + + timestamp = time.strftime("%Y%m%d_%H%M%S") + filename = "inversion_{}_{}.py".format(key, timestamp) + assessment_name = assessment_name or "Model Inversion ({})".format(key) + imports = _build_prediction_imports([func]) + configure = _build_configure() + analytics_writer = _build_analytics_writer() + + if modality == "image": + dim_kw = "input_shape=tuple(INPUT_SHAPE), " + else: + dim_kw = "input_dim=input_dim, " # resolved local (INPUT_DIM or derived from POOL_URL) + tc_kw = "target_classes=TARGET_CLASSES, " if target_classes else "" + + script = '''{imports} + +{configure} + +{analytics_writer} + +API_URL = "{api_url}" +API_KEY = "{api_key}" +POOL_URL = "{pool_url}" +NUM_CLASSES = {num_classes} +INPUT_DIM = {input_dim} +INPUT_SHAPE = {input_shape} +TARGET_CLASSES = {target_classes} +MAX_QUERIES = {max_queries} +SEED = {seed} + +if API_KEY: + os.environ["TARGET_API_KEY"] = API_KEY + + +async def main(): + input_dim = INPUT_DIM + if input_dim is None and POOL_URL: + async with httpx.AsyncClient(timeout=60) as _c: + pool = (await _c.get(POOL_URL)).json()["inputs"] + input_dim = len(pool[0]) + print("Reconstructing {{}} class(es) from {{}}".format(NUM_CLASSES, 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 inversion: {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: + result = await {func}( + spec, + num_classes=NUM_CLASSES, + {dim_kw}modality="{modality}", + {tc_kw}max_queries=MAX_QUERIES, + seed=SEED, + airt_target_model="ml_classifier", + ).run() + print("--- RESULTS ---") + print(" Classes reconstructed: {{}}/{{}}".format(result.classes_reconstructed, result.num_classes)) + for c in result.per_class: + print(" class {{}}: confidence={{:.3f}} {{}}".format( + c.get("class"), c.get("achieved_confidence", 0.0), c.get("reconstruction_preview", ""))) + 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), + pool_url=_safe_str(pool_url), + num_classes=num_classes, + input_dim=repr(int(input_dim)) if input_dim is not None else "None", + input_shape=repr(list(input_shape)) if input_shape else "None", + target_classes=repr(list(target_classes)) if target_classes else "None", + max_queries=max_queries, + seed=seed, + request_template=request_template, + probabilities_path=probabilities_path, + input_format=input_format, + dim_kw=dim_kw, + tc_kw=tc_kw, + modality=_safe_str(modality), + func=func, + assessment_name=_safe_str(assessment_name), + filename=_safe_str(filename), + ) + + return _finalize_prediction_workflow( + script, filename, params, "Model Inversion: {} 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: @@ -7184,6 +7346,7 @@ def _finalize_prediction_workflow(script: str, filename: str, params: dict, desc "generate_extraction_attack": generate_extraction_attack, "generate_membership_attack": generate_membership_attack, "generate_evasion_attack": generate_evasion_attack, + "generate_inversion_attack": generate_inversion_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..9733b6a 100644 --- a/capabilities/ai-red-teaming/tests/test_attack_runner.py +++ b/capabilities/ai-red-teaming/tests/test_attack_runner.py @@ -1147,3 +1147,81 @@ def test_dataset_is_campaign_default(self): script = Path(result["filepath"]).read_text() # 25 objective ids from the dataset should be embedded. assert script.count("'category':") >= 25 or script.count('"category":') >= 25 + + +class TestTraditionalMLGeneration: + """Generated traditional-ML workflows for the Dreadnode /predict flow must + compile and target the provided endpoint (evasion, extraction, membership, + inversion). Covers the hosted-environment path a user drives by natural language.""" + + DN_URL = "https://8000-abc.sandbox.dreadnode.io/predict" + POOL = "https://8000-abc.sandbox.dreadnode.io/pool?n=50" + MEMBERS = "https://8000-abc.sandbox.dreadnode.io/members?n=200" + NONMEMBERS = "https://8000-abc.sandbox.dreadnode.io/nonmembers?n=200" + + def _gen_ok(self, method: str, params: dict) -> str: + result = _generate_method(method, {**params, "generate_only": True}) + assert "error" not in result, result.get("error") + src = Path(result["filepath"]).read_text() + assert self.DN_URL in src, "generated workflow must target the provided predict URL" + compile(src, result["filepath"], "exec") + return src + + def test_evasion_tabular_generates(self): + self._gen_ok( + "generate_evasion_attack", + {"attack_type": "hopskipjump", "api_url": self.DN_URL, + "modality": "tabular", "num_classes": 2, "original": [0.1, 0.2, 0.3]}, + ) + + def test_extraction_tabular_generates(self): + self._gen_ok( + "generate_extraction_attack", + {"attack_type": "knockoff", "api_url": self.DN_URL, + "pool_url": self.POOL, "modality": "tabular", "num_classes": 2}, + ) + + def test_membership_generates(self): + self._gen_ok( + "generate_membership_attack", + {"attack_type": "shadow_model", "api_url": self.DN_URL, "num_classes": 2, + "modality": "tabular", "members_url": self.MEMBERS, "nonmembers_url": self.NONMEMBERS}, + ) + + def test_inversion_tabular_generates(self): + src = self._gen_ok( + "generate_inversion_attack", + {"attack_type": "confidence", "api_url": self.DN_URL, "modality": "tabular", + "num_classes": 2, "pool_url": self.POOL, "target_classes": [0, 1]}, + ) + assert "confidence_inversion(" in src + # input_dim must reference the resolved local (derived from pool), not the None constant + assert "input_dim=input_dim" in src + + def test_inversion_image_generates(self): + src = self._gen_ok( + "generate_inversion_attack", + {"attack_type": "confidence", "api_url": self.DN_URL, "modality": "image", + "num_classes": 10, "input_shape": [8, 8]}, + ) + assert "input_shape=tuple(INPUT_SHAPE)" in src + + def test_inversion_registered(self): + assert "generate_inversion_attack" in runner.METHODS + assert set(runner._INVERSION_ATTACK_MAP) >= {"confidence", "nes"} + + def test_inversion_requires_dim_or_pool(self): + result = _generate_method( + "generate_inversion_attack", + {"attack_type": "confidence", "api_url": self.DN_URL, + "modality": "tabular", "num_classes": 2, "generate_only": True}, + ) + assert "error" in result # neither input_dim nor pool_url supplied + + def test_inversion_image_requires_shape(self): + result = _generate_method( + "generate_inversion_attack", + {"attack_type": "confidence", "api_url": self.DN_URL, + "modality": "image", "num_classes": 10, "generate_only": True}, + ) + assert "error" in result diff --git a/capabilities/ai-red-teaming/tools/attacks.py b/capabilities/ai-red-teaming/tools/attacks.py index c0b5f98..8911c41 100644 --- a/capabilities/ai-red-teaming/tools/attacks.py +++ b/capabilities/ai-red-teaming/tools/attacks.py @@ -612,6 +612,69 @@ def generate_evasion_attack( return _call_runner("generate_evasion_attack", params) +@safe_tool +def generate_inversion_attack( + attack_type: t.Annotated[ + str, "Model-inversion attack: confidence (confidence-guided) or nes (NES gradient-free)." + ] = "confidence", + api_url: t.Annotated[str, "Target classifier predict endpoint (POST)."] = "", + api_key: t.Annotated[str, "API key for the x-api-key header (optional)."] = "", + num_classes: t.Annotated[int, "Number of classes the target predicts."] = 2, + modality: t.Annotated[str, "tabular | image."] = "tabular", + input_dim: t.Annotated[ + int, "Feature count for tabular targets (or leave 0 and pass pool_url to derive it)." + ] = 0, + input_shape: t.Annotated[ + list, "Image shape as [H, W] (e.g. [8, 8]) - required for image targets." + ] = None, + pool_url: t.Annotated[ + str, "GET endpoint returning {inputs: [...]} - used to derive input_dim for tabular." + ] = "", + target_classes: t.Annotated[ + list, "Classes to reconstruct (default: all classes)." + ] = None, + max_queries: t.Annotated[int, "Max target queries."] = 1200, + seed: t.Annotated[int, "Random seed."] = 0, + 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", + assessment_name: t.Annotated[str, "Human-readable assessment name."] = "", +) -> str: + """Reconstruct a representative input per class (model inversion / training-data leakage). + + Queries the target's confidence scores to synthesize an input that maximizes each + class, exposing what a class 'looks like' to the model. For tabular targets pass + input_dim (or pool_url to derive it); for image targets pass input_shape. Results + appear in the platform under AI Red Teaming with per-class reconstruction metrics. + """ + params: dict[str, t.Any] = { + "attack_type": attack_type, + "api_url": api_url, + "num_classes": num_classes, + "modality": modality, + "max_queries": max_queries, + "seed": seed, + "request_template": request_template, + "probabilities_path": probabilities_path, + "input_format": input_format, + } + if api_key: + params["api_key"] = api_key + if input_dim: + params["input_dim"] = input_dim + if input_shape: + params["input_shape"] = input_shape + if pool_url: + params["pool_url"] = pool_url + if target_classes: + params["target_classes"] = target_classes + if assessment_name: + params["assessment_name"] = assessment_name + return _call_runner("generate_inversion_attack", params) + + @safe_tool def generate_multimodal_attack( goal: t.Annotated[ diff --git a/capabilities/ai-red-teaming/tools/environments.py b/capabilities/ai-red-teaming/tools/environments.py index 9abda46..c96d8a4 100644 --- a/capabilities/ai-red-teaming/tools/environments.py +++ b/capabilities/ai-red-teaming/tools/environments.py @@ -79,6 +79,129 @@ def list_environments() -> str: return "\n".join(lines) +# Dreadnode-hosted traditional-ML /predict classifiers, provisionable for +# evasion / extraction / membership / inversion attacks. +_ML_TARGETS: dict[str, dict] = { + "ml-extraction-fraud-tabular": { + "modality": "tabular", "num_classes": 2, "input_dim": 30, + "label": "Credit-card fraud (tabular)", + }, + "ml-extraction-mnist-image": { + "modality": "image", "num_classes": 10, "input_shape": [8, 8], + "label": "Handwritten digits (image)", + }, + "ml-extraction-imdb-text": { + "modality": "text", "num_classes": 2, + "label": "Movie-review sentiment (text)", + }, +} + + +def _fetch_seed(members_url: str) -> t.Any: + """Fetch one sample record from a hosted target's /members endpoint (best effort).""" + import json as _json + import urllib.request as _u + + try: + with _u.urlopen(members_url, timeout=30) as r: # noqa: S310 - platform sandbox URL + data = _json.loads(r.read().decode()) + recs = data.get("records") or data.get("inputs") or [] + return recs[0] if recs else None + except Exception: + return None + + +@safe_tool +def list_ml_targets() -> str: + """List Dreadnode-hosted traditional-ML classifier targets (fraud/tabular, + MNIST/image, IMDB/text) you can deploy with ``provision_ml_target``. + + For your own classifier, skip provisioning and pass its predict URL directly + to generate_evasion_attack / generate_extraction_attack / + generate_membership_attack / generate_inversion_attack via ``api_url``. + """ + lines = ["Hosted traditional-ML targets (deploy with provision_ml_target):"] + for ref, spec in _ML_TARGETS.items(): + shape = spec.get("input_shape") or spec.get("input_dim") + lines.append( + f" - {ref} [{spec['modality']}, {spec['num_classes']} classes" + f"{f', shape/dim={shape}' if shape else ''}] {spec['label']}" + ) + lines.append("\nOwn target? Pass its /predict URL to any trad-ML attack tool via api_url.") + return "\n".join(lines) + + +@safe_tool +def provision_ml_target( + task_ref: t.Annotated[ + str, "Hosted ML target to deploy (e.g. 'ml-extraction-fraud-tabular'); see list_ml_targets." + ], + timeout_sec: t.Annotated[int, "Provision budget in seconds"] = 600, +) -> str: + """Deploy a Dreadnode-hosted traditional-ML classifier and return its /predict URL + plus a seed input, ready to hand to the evasion / extraction / membership / inversion tools. + + Use this when the user wants to attack a **Dreadnode** target. For the user's **own** + classifier, skip this and pass their predict URL directly via ``api_url``. + """ + from dreadnode.core.environment import TaskEnvironment + + _inst, api, org, workspace = _configured() + if not org or not workspace: + return "Not configured for a platform org/workspace. Run `dreadnode login` first." + + env = TaskEnvironment(api, org=org, workspace=workspace, task_ref=task_ref, timeout_sec=timeout_sec) + ctx = _run(env.setup()) + svc = (ctx.get("service_urls") or {}).get("challenge") + url = (svc.get("url") if isinstance(svc, dict) else svc) or "" + if not url: + return f"Target '{task_ref}' provisioned but exposed no 'challenge' URL: {ctx.get('service_urls')}" + + predict_url = f"{url}/predict" + pool_url = f"{url}/pool?n=50" + members_url = f"{url}/members?n=1" + members_pool_url = f"{url}/members?n=200" + nonmembers_url = f"{url}/nonmembers?n=200" + spec = _ML_TARGETS.get(task_ref, {}) + modality = spec.get("modality", "tabular") + num_classes = spec.get("num_classes") + seed = _fetch_seed(members_url) + + lines = [ + f"Target '{task_ref}' is ready.", + f" Predict URL: {predict_url}", + f" Pool URL: {pool_url} (query inputs for extraction; derives input_dim)", + f" Members URL: {members_url} (labeled records for membership/evasion seeds)", + f" Modality: {modality}", + ] + if num_classes is not None: + lines.append(f" Classes: {num_classes}") + if spec.get("input_shape"): + lines.append(f" Input shape: {spec['input_shape']} (for inversion)") + if spec.get("input_dim"): + lines.append(f" Input dim: {spec['input_dim']} (for inversion)") + if seed is not None: + preview = repr(seed) + lines.append(f" Seed input: {preview[:160]}{'...' if len(preview) > 160 else ''}") + + nc = num_classes if num_classes is not None else 2 + lines += [ + "", + ">>> NEXT STEP: run an attack against it, e.g.:", + f' - Evasion: generate_evasion_attack(api_url="{predict_url}", modality="{modality}", ' + f'num_classes={nc}, original=)', + f' - Extraction: generate_extraction_attack(api_url="{predict_url}", pool_url="{pool_url}", ' + f'num_classes={nc}, modality="{modality}")', + f' - Membership: generate_membership_attack(api_url="{predict_url}", num_classes={nc}, ' + f'modality="{modality}", members_url="{members_pool_url}", nonmembers_url="{nonmembers_url}")', + f' - Inversion: generate_inversion_attack(api_url="{predict_url}", num_classes={nc}, ' + f'modality="{modality}", ' + + (f'input_shape={spec.get("input_shape")}' if spec.get("input_shape") else f'pool_url="{pool_url}"') + + ")", + ] + return "\n".join(lines) + + @safe_tool def provision_environment( task_ref: t.Annotated[str, "Environment/task to deploy, e.g. 'finops-mesh'"],