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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions capabilities/ai-red-teaming/agents/ai-red-teaming-agent.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
Expand Down
6 changes: 4 additions & 2 deletions capabilities/ai-red-teaming/capability.yaml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
schema: 1
name: ai-red-teaming
version: "1.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,
Expand All @@ -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.

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 @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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,
}


Expand Down
78 changes: 78 additions & 0 deletions capabilities/ai-red-teaming/tests/test_attack_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
63 changes: 63 additions & 0 deletions capabilities/ai-red-teaming/tools/attacks.py
Original file line number Diff line number Diff line change
Expand Up @@ -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[
Expand Down
Loading
Loading