fix(ai-red-teaming): authenticate agentic attacks + lighten ATLAS default budget

capabilities/ai-red-teaming/capability.yaml

@@ -1,6 +1,6 @@
 schema: 1
 name: ai-red-teaming
-version: "1.18.4"
+version: "1.18.5"
 description: >
   Probe the security and safety of AI applications, agents, and foundation models.
   Orchestrates adversarial attack workflows to discover vulnerabilities in LLMs,

capabilities/ai-red-teaming/scripts/attack_runner.py

@@ -133,6 +133,37 @@ def _auto_execute_workflow(filename: str, timeout: int = 3600) -> str:
             e.msg, e.lineno
         )
 
+    # Inject the provisioned environment's bearer/api-key token so the target is called
+    # AUTHENTICATED. The generated agent target reads it from an env var (AGENT_API_KEY,
+    # TARGET_API_KEY, or a scenario name like FINOPS_EXECUTE_TOKEN). The one-shot token is
+    # recorded in the env registry by provision_environment; without this the target gets
+    # an empty bearer, returns no response, and every trial errors. Match the registry
+    # base_url against the URL baked into the script, then set each token-like env var the
+    # script reads (excluding DREADNODE_* platform creds).
+    try:
+        import re as _re
+
+        reg_path = Path(os.environ.get("AIRT_ENV_REGISTRY_PATH", "/tmp/airt_environments.json"))
+        entries = json.loads(reg_path.read_text()) if reg_path.exists() else []
+        token, best_len = "", -1
+        for e in entries if isinstance(entries, list) else []:
+            base = (e.get("base_url") or "").rstrip("/")
+            tok = e.get("execute_token") or ""
+            if base and tok and base in source and len(base) > best_len:
+                token, best_len = tok, len(base)
+        if token:
+            token_vars = {
+                v
+                for v in _re.findall(
+                    r'os\.environ\.get\(\s*["\']([A-Z0-9_]+(?:TOKEN|API_KEY))["\']', source
+                )
+                if not v.startswith("DREADNODE_")
+            }
+            for var in token_vars:
+                os.environ[var] = token
+    except Exception:  # noqa: BLE001 - best-effort auth wiring; script still runs
+        pass
+
     try:
         python_executable = resolve_python_executable()
         print(

capabilities/ai-red-teaming/tools/attacks.py

@@ -617,7 +617,11 @@ def generate_atlas_attack(
         "policy: finops, devsecops, healthcare, soc, devops (RCE), support (exfil). "
         "The environment-derived name also works (e.g. 'devops-rce', 'support-exfil').",
     ] = "",
-    total_budget: t.Annotated[int, "Total attack episodes across the campaign"] = 64,
+    total_budget: t.Annotated[
+        int,
+        "Total attack episodes across the campaign. Defaults to a light, fast "
+        "interactive budget (matches the cookbook demo); raise for deeper campaigns.",
+    ] = 12,
     evaluator_model: t.Annotated[str, "Judge LLM (defaults to attacker_model)"] = "",
     agent_auth_type: t.Annotated[
         str, "Auth scheme: 'none', 'bearer', or 'api_key'"
@@ -637,6 +641,10 @@ def generate_atlas_attack(
     execution* (not just verbal compliance). Point it at a deployed multi-agent
     environment that exposes POST /attack {prompt, surface, injection} ->
     {content, tool_calls, ...}.
+
+    ``total_budget`` defaults to a light interactive budget (12 episodes, the
+    cookbook demo config) so a single call returns in a couple of minutes rather
+    than running the full 64-episode campaign. Raise it for a deeper sweep.
     """
     params: dict[str, t.Any] = {
         "agent_url": agent_url,

capabilities/ai-red-teaming/tools/environments.py

@@ -73,11 +73,23 @@ def _registry_remove(env_id: str, registry_path: _Path = REGISTRY_PATH) -> None:
 
 
 def _register_provisioned(
-    env: t.Any, task_ref: str, org: str, workspace: str, registry_path: _Path = REGISTRY_PATH
+    env: t.Any,
+    task_ref: str,
+    org: str,
+    workspace: str,
+    registry_path: _Path = REGISTRY_PATH,
+    *,
+    base_url: str = "",
+    execute_token: str = "",
 ) -> str:
     """Record a just-provisioned environment in the session registry. Returns its id
     (empty string if the environment exposed none, e.g. a provision that never became
-    ready)."""
+    ready).
+
+    ``base_url`` + ``execute_token`` are persisted so the attack runner can inject the
+    one-shot bearer token into the attack subprocess env (the generated attack script
+    reads it from an env var). Without this the target is called unauthenticated and
+    returns no response, so every trial errors."""
     env_id = getattr(env, "id", None) or ""
     if not env_id:
         return ""
@@ -87,6 +99,8 @@ def _register_provisioned(
             "task_ref": task_ref,
             "org": org,
             "workspace": workspace,
+            "base_url": base_url,
+            "execute_token": execute_token,
             "provisioned_at": _iso_now(),
             "provisioned_at_ts": _time.time(),
         },
@@ -321,7 +335,9 @@ def _mk(ref: str) -> t.Any:
     token = env._execute_token or ""  # noqa: SLF001 - one-shot provision token
     # Record the sandbox so it is torn down at assessment completion even if the
     # attack path forgets - a hosted sandbox bills for its whole lifetime.
-    env_id = _register_provisioned(env, task_ref, org, workspace)
+    env_id = _register_provisioned(
+        env, task_ref, org, workspace, base_url=url.rstrip("/"), execute_token=token
+    )
     if not url:
         return f"Environment '{task_ref}' provisioned but exposed no 'challenge' URL: {ctx.get('service_urls')}"