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
2 changes: 1 addition & 1 deletion 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.16.2"
version: "1.16.3"
description: >
Probe the security and safety of AI applications, agents, and foundation models.
Orchestrates adversarial attack workflows to discover vulnerabilities in LLMs,
Expand Down
89 changes: 65 additions & 24 deletions capabilities/ai-red-teaming/hooks/block_cli_provisioning.py
Original file line number Diff line number Diff line change
@@ -1,36 +1,69 @@
"""Structural guardrail: deny environment provisioning through the shell.

The `provisioning-and-lifecycle` skill instructs the agent to provision only via
the `provision_environment` tool and to NEVER run `dreadnode env ...` in a shell.
Skill prose is advisory - a weaker driver model can ignore it and fall back to
the CLI (ENG-8477). This hook enforces the rule structurally: if the agent tries
to provision an environment through a shell tool, the call is denied before it
runs and the agent is told to use `provision_environment` (or stop and inform
the user).
"""Structural guardrails: deny shell fallbacks that bypass first-class tools.

The AIRT skills mandate tool-only paths for a few operations - most importantly,
`provisioning-and-lifecycle` and `error-troubleshooting` both say to NEVER run
`dreadnode env ...` in a shell and to use the `provision_environment` /
`teardown_environment` tools instead. Skill prose is advisory; a weaker driver
model can ignore it and fall back to the CLI (ENG-8477).

This module enforces those rules structurally with a small policy table. A
`ToolStart` hook inspects every shell tool call and, when its command matches a
rule, denies the call before it runs and redirects the agent to the correct
tool (via `RetryWithFeedback`, which the runtime turns into a `[POLICY DENIED]`
tool result). Rules are scoped narrowly so legitimate shell and CLI usage
(for example `dn airt run`, which the skills permit) is never affected.

To add a guardrail, append a `_Rule` - one line per policy.
"""

from __future__ import annotations

import json
import re
import typing as t
from dataclasses import dataclass

from dreadnode.agents.events import ToolStart
from dreadnode.agents.reactions import RetryWithFeedback
from dreadnode.core.hook import hook

# Shell-style tools that can execute a CLI command.
_SHELL_TOOLS = {"bash", "shell", "sh", "python", "python3", "run_command", "execute"}
_SHELL_TOOLS = frozenset({"bash", "shell", "sh", "python", "python3", "run_command", "execute"})


@dataclass(frozen=True)
class _Rule:
"""A single deny-and-redirect policy for shell tool calls."""

# `dreadnode env ...` / `dn env ...` / `dreadnode environment ...` in a shell.
# Matches the exact pattern the skill forbids without touching other CLI usage.
_CLI_PROVISION = re.compile(r"\b(?:dreadnode|dn)\s+env(?:ironment)?\b", re.IGNORECASE)
tools: frozenset[str]
pattern: re.Pattern[str]
feedback: str

_FEEDBACK = (
"Provisioning an environment through the shell/CLI is not permitted. "
"Use the `provision_environment` tool instead - it resolves the catalog and "
"credentials for you. If `provision_environment` is unavailable, STOP and "
"inform the user; do not fall back to the shell."

# NOTE: order matters - the first matching rule wins, so list specific rules
# (provision / teardown) before the generic `dreadnode env` catch-all.
_RULES: tuple[_Rule, ...] = (
_Rule(
_SHELL_TOOLS,
re.compile(r"\b(?:dreadnode|dn)\s+env(?:ironment)?\s+(?:provision|create|up|start|new)\b", re.IGNORECASE),
"Provisioning an environment through the shell/CLI is not permitted. Use the "
"`provision_environment` tool instead - it resolves the catalog and credentials for you. "
"If it is unavailable, STOP and inform the user; do not fall back to the shell.",
),
_Rule(
_SHELL_TOOLS,
re.compile(r"\b(?:dreadnode|dn)\s+env(?:ironment)?\s+(?:teardown|delete|destroy|down|stop|rm|remove)\b", re.IGNORECASE),
"Tearing down an environment through the shell/CLI is not permitted. Use the "
"`teardown_environment` tool instead. If it is unavailable, STOP and inform the user; "
"do not fall back to the shell.",
),
_Rule(
_SHELL_TOOLS,
re.compile(r"\b(?:dreadnode|dn)\s+env(?:ironment)?\b", re.IGNORECASE),
"Managing environments through the shell/CLI is not permitted. Use the environment tools "
"(`provision_environment`, `list_environments`, `teardown_environment`) instead. If they "
"are unavailable, STOP and inform the user; do not fall back to the shell.",
),
)


Expand All @@ -46,19 +79,27 @@ def _iter_strings(value: t.Any) -> t.Iterator[str]:
yield from _iter_strings(item)


def _match(tool_name: str, args: t.Any) -> _Rule | None:
"""Return the first rule that denies this tool call, if any."""
name = tool_name.lower()
texts = list(_iter_strings(args))
for rule in _RULES:
if name in rule.tools and any(rule.pattern.search(text) for text in texts):
return rule
return None


@hook(ToolStart)
async def block_cli_provisioning(event: ToolStart) -> RetryWithFeedback | None:
"""Deny shell-based environment provisioning and redirect to the tool."""
if event.tool_call.name.lower() not in _SHELL_TOOLS:
return None

"""Deny shell calls that bypass first-class tools and redirect the agent."""
raw = event.tool_call.function.arguments
try:
args: t.Any = json.loads(raw)
except (json.JSONDecodeError, TypeError):
args = raw

if any(_CLI_PROVISION.search(text) for text in _iter_strings(args)):
return RetryWithFeedback(feedback=_FEEDBACK, tool_call_id=event.tool_call.id)
rule = _match(event.tool_call.name, args)
if rule is not None:
return RetryWithFeedback(feedback=rule.feedback, tool_call_id=event.tool_call.id)

return None
13 changes: 13 additions & 0 deletions capabilities/ai-red-teaming/tests/test_block_cli_provisioning.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,16 @@
import asyncio
import importlib.util
import json
import sys
from pathlib import Path

_HOOK_PATH = Path(__file__).resolve().parents[1] / "hooks" / "block_cli_provisioning.py"
_spec = importlib.util.spec_from_file_location("airt_block_cli_provisioning", _HOOK_PATH)
assert _spec and _spec.loader
_mod = importlib.util.module_from_spec(_spec)
# Register before exec so dataclasses can resolve the module namespace under
# `from __future__ import annotations` (dataclasses looks the module up in sys.modules).
sys.modules[_spec.name] = _mod
_spec.loader.exec_module(_mod)


Expand Down Expand Up @@ -51,8 +55,17 @@ def test_blocks_dn_env_and_environment_variants() -> None:
assert _run("python", json.dumps({"code": "os.system('dreadnode env provision m')"})) is not None


def test_blocks_teardown_via_cli_and_redirects_to_teardown_tool() -> None:
r = _run("bash", json.dumps({"command": "dreadnode env teardown finops-mesh"}))
assert r is not None
assert "teardown_environment" in r.feedback
# destroy/delete variants also redirect to the teardown tool
assert "teardown_environment" in _run("bash", json.dumps({"command": "dn env delete soc-mesh"})).feedback


def test_allows_non_provisioning_shell_commands() -> None:
assert _run("bash", json.dumps({"command": "ls -la /home/user"})) is None
# dn airt run is a permitted CLI (no skill forbids it) - must not be blocked
assert _run("bash", json.dumps({"command": "dn airt run --goal x --attack tap"})) is None
assert _run("bash", json.dumps({"command": "python attack.py"})) is None

Expand Down
Loading