From ccfa053fca11810bc6d44a7412598ed0b052b0ec Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Tue, 14 Jul 2026 17:08:42 -0700 Subject: [PATCH 1/2] fix(ai-red-teaming): make multimodal SageMaker SigV4 targets work end-to-end (v1.7.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Probing an Amazon SageMaker /invocations endpoint from the TUI silently produced zero real trials. Three bugs in the generated multimodal-target workflow, each found via live probing against a real Gemma-3 vision endpoint: 1. SigV4 signing imported `boto3`, which isn't installed in the workflow-runner venv (only `botocore` is), so every trial died with `ModuleNotFoundError: No module named 'boto3'`. Sign via `botocore.session` directly — botocore is always present. Applies to both the image target and the multimodal custom target. 2. Injection-image paths were baked into the workflow as relative paths, which don't resolve from the workflow's execution CWD -> image never loaded (`image=False`). Resolve media paths to absolute in `_expand_media_paths`. 3. Request-template placeholders were never substituted: the generated `.replace()` used double-brace `{{prompt}}`/`{{image_b64}}` literals while the template uses single braces, so an undecodable image reached the container and it returned `424 Failed Dependency`. Emit single-brace replacements. Also add `aws_sigv4` as a first-class `custom_auth_type` for multimodal targets (with `custom_region`/`custom_service`), and print the real endpoint URL as the target instead of the never-invoked backing model id. Verified end-to-end: TUI -> SigV4-signed POST -> HTTP 200 real model response, confirmed by CloudWatch Invocations on the endpoint. --- capabilities/ai-red-teaming/capability.yaml | 2 +- .../ai-red-teaming/scripts/attack_runner.py | 77 +++++++++++++++---- capabilities/ai-red-teaming/tools/attacks.py | 16 +++- 3 files changed, 78 insertions(+), 17 deletions(-) diff --git a/capabilities/ai-red-teaming/capability.yaml b/capabilities/ai-red-teaming/capability.yaml index 560c920..19cece6 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.7.0" +version: "1.7.1" 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 aac37e1..7d610fe 100644 --- a/capabilities/ai-red-teaming/scripts/attack_runner.py +++ b/capabilities/ai-red-teaming/scripts/attack_runner.py @@ -4747,11 +4747,11 @@ def _build_image_target(target_config: dict) -> str: auth_code = ' _api_key = os.environ.get("{}", "")\n headers["X-API-Key"] = _api_key'.format(auth_env_var) elif auth_type == "aws_sigv4": auth_code = ( - " # AWS SigV4 auth — uses boto3 session\n" - " import boto3\n" + " # AWS SigV4 auth — uses the botocore credential chain\n" + " import botocore.session\n" " from botocore.auth import SigV4Auth\n" " from botocore.awsrequest import AWSRequest\n" - " _session = boto3.Session()\n" + " _session = botocore.session.get_session()\n" " _credentials = _session.get_credentials().get_frozen_credentials()" ) else: @@ -5262,14 +5262,18 @@ def _resolve_multimodal_prompts( def _expand_media_paths( paths: list[str] | None, directory: str | None, kind: str ) -> list[str]: - """Resolve explicit paths + a directory glob into a sorted list of media files.""" + """Resolve explicit paths + a directory glob into a sorted list of media files. + + Paths are resolved to absolute so they stay valid when the generated workflow + runs from a different working directory than the tool that produced them. + """ resolved: list[str] = [] for p in paths or []: if p: - resolved.append(str(p)) + resolved.append(str(Path(p).expanduser().resolve())) if directory: exts = _MULTIMODAL_MEDIA_EXTS.get(kind, set()) - d = Path(directory).expanduser() + d = Path(directory).expanduser().resolve() if d.is_dir(): for f in sorted(d.rglob("*")): if f.is_file() and f.suffix.lower() in exts: @@ -5315,6 +5319,30 @@ def _build_multimodal_imports(transforms: list[dict]) -> str: return "\n".join(lines) +def _sigv4_sign_block(auth_type: str, url: str, region: str, service: str) -> str: + """Generated code that SigV4-signs the assembled request body for AWS endpoints. + + Returns an empty string for non-SigV4 auth. For ``aws_sigv4`` (e.g. Amazon + SageMaker ``/invocations``), signs the exact ``_content`` bytes with the + botocore credential chain and replaces ``headers`` with the signed set. + Uses ``botocore`` directly (always present) rather than ``boto3`` so signing + works even in minimal runtimes where ``boto3`` isn't installed. + """ + if auth_type != "aws_sigv4": + return "" + return ( + " import botocore.session\n" + " from botocore.auth import SigV4Auth\n" + " from botocore.awsrequest import AWSRequest\n" + " _creds = botocore.session.get_session().get_credentials()\n" + " if _creds is None:\n" + ' raise RuntimeError("aws_sigv4 target requires AWS credentials (env vars, profile, or role)")\n' + ' _req = AWSRequest(method="POST", url="{url}", data=_content, headers=dict(headers))\n' + ' SigV4Auth(_creds.get_frozen_credentials(), "{service}", "{region}").add_auth(_req)\n' + " headers = dict(_req.headers)" + ).format(url=url, service=service, region=region) + + def _build_custom_multimodal_target(custom: dict) -> str: """Build a @task target that sends a multimodal Message to a custom HTTP endpoint. @@ -5328,6 +5356,8 @@ def _build_custom_multimodal_target(custom: dict) -> str: auth_env_var = custom.get("auth_env_var", "TARGET_API_KEY") request_template = custom.get("request_template", '{"prompt": "{prompt}", "image": "{image_b64}"}') text_path = _safe_str(custom.get("response_text_path", "$.response")) + region = _safe_str(custom.get("region") or "us-east-1") + service = _safe_str(custom.get("service") or "sagemaker") if auth_type == "bearer": auth_lines = ( @@ -5340,7 +5370,8 @@ def _build_custom_multimodal_target(custom: dict) -> str: ' headers["X-API-Key"] = api_key'.format(auth_env_var) ) else: - auth_lines = " pass # No auth configured" + # none / aws_sigv4 — SigV4 signs the assembled body below, not here. + auth_lines = " pass # No header auth (SigV4 signs the request below if configured)" lines = [ "@task", @@ -5384,14 +5415,16 @@ def _build_custom_multimodal_target(custom: dict) -> str: auth_lines, "", " body_str = {}".format(repr(request_template)), - " body_str = body_str.replace('{{prompt}}', json.dumps(prompt)[1:-1])", - " body_str = body_str.replace('{{image_b64}}', image_b64)", - " body_str = body_str.replace('{{audio_b64}}', audio_b64)", - " body_str = body_str.replace('{{video_b64}}', video_b64)", + " body_str = body_str.replace('{prompt}', json.dumps(prompt)[1:-1])", + " body_str = body_str.replace('{image_b64}', image_b64)", + " body_str = body_str.replace('{audio_b64}', audio_b64)", + " body_str = body_str.replace('{video_b64}', video_b64)", " body = json.loads(body_str)", "", + " _content = json.dumps(body).encode()", + _sigv4_sign_block(auth_type, url, region, service), " async with httpx.AsyncClient(timeout=120.0) as client:", - ' resp = await client.post("{}", json=body, headers=headers)'.format(url), + ' resp = await client.post("{}", content=_content, headers=headers)'.format(url), " resp.raise_for_status()", " data = resp.json()", "", @@ -5631,6 +5664,9 @@ def generate_multimodal_attack(params: dict) -> dict: "custom_request_template", '{"prompt": "{prompt}", "image": "{image_b64}"}' ), "response_text_path": params.get("custom_response_text_path", "$.response"), + # For auth_type="aws_sigv4" (e.g. Amazon SageMaker /invocations). + "region": params.get("custom_region") or "us-east-1", + "service": params.get("custom_service") or "sagemaker", } if not goal: @@ -5717,6 +5753,15 @@ def generate_multimodal_attack(params: dict) -> dict: 'GOAL = "{}"'.format(_safe_str(goal)), "GOAL_CATEGORY = GoalCategory.{}".format(goal_cat_enum), 'TARGET_MODEL = "{}"'.format(_safe_str(target_model)), + # Human-readable target for display/provenance: a custom endpoint's URL + # when one is configured, else the model id. TARGET_MODEL is only the + # never-invoked backing generator for custom targets, so printing it as + # "Target" is misleading. + 'TARGET_LABEL = "{}"'.format( + _safe_str(custom_target.get("url") or target_model) + if custom_target + else _safe_str(target_model) + ), # ATTACKER_MODEL is unused for multimodal but the proxy-routing block # expects the symbol; alias it to the judge model. 'ATTACKER_MODEL = "{}"'.format(_safe_str(judge_model)), @@ -5734,13 +5779,17 @@ def generate_multimodal_attack(params: dict) -> dict: 'MEDIA_OUTPUT_RUBRIC = "{}"'.format(_safe_str(media_output_rubric)), "TRANSFORMS = {}".format(transforms_expr), 'ASSESSMENT_NAME = "{}"'.format(_safe_str(assessment_name)), - 'ASSESSMENT_DESC = "Multimodal red teaming vs {}"'.format(_safe_str(target_model)), + 'ASSESSMENT_DESC = "Multimodal red teaming vs {}"'.format( + _safe_str(custom_target.get("url") or target_model) + if custom_target + else _safe_str(target_model) + ), 'WORKFLOW_RUN_ID = "{}"'.format(_safe_str(filename)), "", 'print("=" * 60)', 'print("MULTIMODAL RED TEAMING CONFIGURATION")', 'print("=" * 60)', - 'print(f" Target: {TARGET_MODEL}")', + 'print(f" Target: {TARGET_LABEL}")', 'print(f" Judge: {JUDGE_MODEL}")', 'print(f" Goal: {GOAL}")', 'print(f" Images: {len(IMAGE_PATHS)} Audio: {len(AUDIO_PATHS)} Video: {len(VIDEO_PATHS)}")', diff --git a/capabilities/ai-red-teaming/tools/attacks.py b/capabilities/ai-red-teaming/tools/attacks.py index 774cbfa..0eff907 100644 --- a/capabilities/ai-red-teaming/tools/attacks.py +++ b/capabilities/ai-red-teaming/tools/attacks.py @@ -466,7 +466,12 @@ def generate_multimodal_attack( "Target a custom multimodal HTTP endpoint instead of a litellm model. When " "set, target_model is optional; the endpoint receives the text + base64 media.", ] = "", - custom_auth_type: t.Annotated[str, "Auth scheme for custom_url: 'none', 'bearer', or 'api_key'"] = "none", + custom_auth_type: t.Annotated[ + str, + "Auth scheme for custom_url: 'none', 'bearer', 'api_key', or 'aws_sigv4' " + "(SigV4-signed, e.g. an Amazon SageMaker /invocations endpoint — set " + "custom_region and custom_service).", + ] = "none", custom_auth_env_var: t.Annotated[str, "Env var holding the custom endpoint credential"] = "TARGET_API_KEY", custom_request_template: t.Annotated[ str, @@ -484,7 +489,12 @@ def generate_multimodal_attack( "(Amazon Nova Sonic S2S over Bedrock). The input audio is spoken to the model and " "its spoken reply (audio + transcript) is scored.", ] = "", - custom_region: t.Annotated[str, "AWS region for a streaming target (default us-east-1)."] = "", + custom_region: t.Annotated[ + str, "AWS region for a streaming (Nova Sonic) or aws_sigv4 target (default us-east-1)." + ] = "", + custom_service: t.Annotated[ + str, "AWS service for an aws_sigv4 HTTP target (default 'sagemaker')." + ] = "", custom_voice: t.Annotated[str, "Voice id for a Nova Sonic streaming target (default matthew)."] = "", custom_system_prompt: t.Annotated[str, "System prompt for a streaming S2S target."] = "", custom_model_id: t.Annotated[ @@ -582,6 +592,8 @@ def generate_multimodal_attack( params["custom_protocol"] = custom_protocol if custom_region: params["custom_region"] = custom_region + if custom_service: + params["custom_service"] = custom_service if custom_voice: params["custom_voice"] = custom_voice if custom_system_prompt: From 67873437c28b0d2ee718c04e7491a792fc5e3783 Mon Sep 17 00:00:00 2001 From: Raja Sekhar Rao Dheekonda Date: Tue, 14 Jul 2026 19:08:45 -0700 Subject: [PATCH 2/2] fix(ai-red-teaming): support raw-audio (audio_bytes) multimodal targets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SageMaker ASR/audio endpoints (e.g. a JumpStart Whisper container) take a raw audio file as the request body, not JSON — the existing JSON request_template path can't drive them. Add custom_request_format='audio_bytes' (with custom_audio_content_type, default audio/wav): the generated target posts the first audio part's raw bytes with SigV4 signing and extracts the transcript via custom_response_text_path. Verified end-to-end from the TUI against a live SageMaker Whisper endpoint: time_stretch-transformed audio -> raw-bytes SigV4 POST -> HTTP 200 transcript. --- .../ai-red-teaming/scripts/attack_runner.py | 48 +++++++++++++++---- capabilities/ai-red-teaming/tools/attacks.py | 13 +++++ 2 files changed, 52 insertions(+), 9 deletions(-) diff --git a/capabilities/ai-red-teaming/scripts/attack_runner.py b/capabilities/ai-red-teaming/scripts/attack_runner.py index 7d610fe..60ac6e3 100644 --- a/capabilities/ai-red-teaming/scripts/attack_runner.py +++ b/capabilities/ai-red-teaming/scripts/attack_runner.py @@ -5343,6 +5343,33 @@ def _sigv4_sign_block(auth_type: str, url: str, region: str, service: str) -> st ).format(url=url, service=service, region=region) +def _multimodal_body_block(request_format: str, request_template: str, audio_content_type: str) -> str: + """Generated code that assembles the request body into ``_content`` (bytes). + + ``json`` (default) substitutes the media placeholders into the JSON template. + ``audio_bytes`` posts the raw audio bytes with ``audio_content_type`` (for SageMaker + ASR/audio endpoints that take an audio file body rather than a JSON payload). + """ + if request_format == "audio_bytes": + return "\n".join( + [ + " _content = audio_raw", + ' headers["Content-Type"] = "{}"'.format(audio_content_type), + ] + ) + return "\n".join( + [ + " body_str = {}".format(repr(request_template)), + " body_str = body_str.replace('{prompt}', json.dumps(prompt)[1:-1])", + " body_str = body_str.replace('{image_b64}', image_b64)", + " body_str = body_str.replace('{audio_b64}', audio_b64)", + " body_str = body_str.replace('{video_b64}', video_b64)", + " body = json.loads(body_str)", + " _content = json.dumps(body).encode()", + ] + ) + + def _build_custom_multimodal_target(custom: dict) -> str: """Build a @task target that sends a multimodal Message to a custom HTTP endpoint. @@ -5358,6 +5385,11 @@ def _build_custom_multimodal_target(custom: dict) -> str: text_path = _safe_str(custom.get("response_text_path", "$.response")) region = _safe_str(custom.get("region") or "us-east-1") service = _safe_str(custom.get("service") or "sagemaker") + # "json" (default) renders the request_template; "audio_bytes" posts the first audio + # part's raw bytes with audio_content_type — for SageMaker ASR/audio endpoints (e.g. + # Whisper) that take an audio file body instead of a JSON payload. + request_format = _safe_str(custom.get("request_format") or "json") + audio_content_type = _safe_str(custom.get("audio_content_type") or "audio/wav") if auth_type == "bearer": auth_lines = ( @@ -5386,6 +5418,7 @@ def _build_custom_multimodal_target(custom: dict) -> str: " prompt = ''", " image_b64 = ''", " audio_b64 = ''", + " audio_raw = b''", " video_b64 = ''", " for p in (getattr(message, 'content_parts', None) or []):", " ptype = getattr(p, 'type', None)", @@ -5401,7 +5434,8 @@ def _build_custom_multimodal_target(custom: dict) -> str: " image_b64 = ''", " elif ptype == 'input_audio' and not audio_b64:", " try:", - " audio_b64 = base64.b64encode(p.to_bytes()).decode('ascii')", + " audio_raw = p.to_bytes()", + " audio_b64 = base64.b64encode(audio_raw).decode('ascii')", " except Exception:", " audio_b64 = ''", " elif ptype == 'video_url' and not video_b64:", @@ -5414,14 +5448,7 @@ def _build_custom_multimodal_target(custom: dict) -> str: ' headers = {"Content-Type": "application/json"}', auth_lines, "", - " body_str = {}".format(repr(request_template)), - " body_str = body_str.replace('{prompt}', json.dumps(prompt)[1:-1])", - " body_str = body_str.replace('{image_b64}', image_b64)", - " body_str = body_str.replace('{audio_b64}', audio_b64)", - " body_str = body_str.replace('{video_b64}', video_b64)", - " body = json.loads(body_str)", - "", - " _content = json.dumps(body).encode()", + _multimodal_body_block(request_format, request_template, audio_content_type), _sigv4_sign_block(auth_type, url, region, service), " async with httpx.AsyncClient(timeout=120.0) as client:", ' resp = await client.post("{}", content=_content, headers=headers)'.format(url), @@ -5667,6 +5694,9 @@ def generate_multimodal_attack(params: dict) -> dict: # For auth_type="aws_sigv4" (e.g. Amazon SageMaker /invocations). "region": params.get("custom_region") or "us-east-1", "service": params.get("custom_service") or "sagemaker", + # "audio_bytes" posts raw audio to ASR/audio endpoints (e.g. Whisper). + "request_format": params.get("custom_request_format") or "json", + "audio_content_type": params.get("custom_audio_content_type") or "audio/wav", } if not goal: diff --git a/capabilities/ai-red-teaming/tools/attacks.py b/capabilities/ai-red-teaming/tools/attacks.py index 0eff907..56f5188 100644 --- a/capabilities/ai-red-teaming/tools/attacks.py +++ b/capabilities/ai-red-teaming/tools/attacks.py @@ -495,6 +495,15 @@ def generate_multimodal_attack( custom_service: t.Annotated[ str, "AWS service for an aws_sigv4 HTTP target (default 'sagemaker')." ] = "", + custom_request_format: t.Annotated[ + str, + "Request body encoding for custom_url: 'json' (default, renders the request " + "template) or 'audio_bytes' (posts the raw audio file with custom_audio_content_type " + "— for SageMaker ASR/audio endpoints like Whisper that take an audio body, not JSON).", + ] = "", + custom_audio_content_type: t.Annotated[ + str, "Content-Type for a custom_request_format='audio_bytes' target (default 'audio/wav')." + ] = "", custom_voice: t.Annotated[str, "Voice id for a Nova Sonic streaming target (default matthew)."] = "", custom_system_prompt: t.Annotated[str, "System prompt for a streaming S2S target."] = "", custom_model_id: t.Annotated[ @@ -594,6 +603,10 @@ def generate_multimodal_attack( params["custom_region"] = custom_region if custom_service: params["custom_service"] = custom_service + if custom_request_format: + params["custom_request_format"] = custom_request_format + if custom_audio_content_type: + params["custom_audio_content_type"] = custom_audio_content_type if custom_voice: params["custom_voice"] = custom_voice if custom_system_prompt: