Merge branch 'main' of https://github.com/Praveenk8051/agentic_security into feat/extension-with-sample-tests

agentic_security/integrations/__init__.py

@@ -0,0 +1,12 @@
+import asyncio
+from typing import Protocol
+
+
+class IntegrationProto(Protocol):
+    def __init__(
+        self, prompt_groups: list, tools_inbox: asyncio.Queue, opts: dict = {}
+    ):
+        ...
+
+    async def apply(self) -> list:
+        ...

agentic_security/probe_actor/cost_module.py

@@ -0,0 +1,58 @@
+def calculate_cost(tokens: int, model: str = "deepseek-chat") -> float:
+    """Calculate API cost based on token count and model.
+
+    Args:
+        tokens (int): Number of tokens used
+        model (str): Model name to calculate cost for
+
+    Returns:
+        float: Cost in USD
+    """
+    # API pricing as of 2024-03-01
+    pricing = {
+        "deepseek-chat": {
+            "input": 0.0007 / 1000,  # $0.70 per million input tokens
+            "output": 0.0028 / 1000,  # $2.80 per million output tokens
+        },
+        "gpt-4-turbo": {
+            "input": 0.01 / 1000,  # $10 per million input tokens
+            "output": 0.03 / 1000,  # $30 per million output tokens
+        },
+        "gpt-4": {
+            "input": 0.03 / 1000,  # $30 per million input tokens
+            "output": 0.06 / 1000,  # $60 per million output tokens
+        },
+        "gpt-3.5-turbo": {
+            "input": 0.0015 / 1000,  # $1.50 per million input tokens
+            "output": 0.002 / 1000,  # $2.00 per million output tokens
+        },
+        "claude-3-opus": {
+            "input": 0.015 / 1000,  # $15 per million input tokens
+            "output": 0.075 / 1000,  # $75 per million output tokens
+        },
+        "claude-3-sonnet": {
+            "input": 0.003 / 1000,  # $3 per million input tokens
+            "output": 0.015 / 1000,  # $15 per million output tokens
+        },
+        "claude-3-haiku": {
+            "input": 0.00025 / 1000,  # $0.25 per million input tokens
+            "output": 0.00125 / 1000,  # $1.25 per million output tokens
+        },
+        "mistral-large": {
+            "input": 0.008 / 1000,  # $8 per million input tokens
+            "output": 0.024 / 1000,  # $24 per million output tokens
+        },
+        "mixtral-8x7b": {
+            "input": 0.002 / 1000,  # $2 per million input tokens
+            "output": 0.006 / 1000,  # $6 per million output tokens
+        },
+    }
+
+    if model not in pricing:
+        raise ValueError(f"Unknown model: {model}")
+
+    # For now, assume 1:1 input/output ratio
+    input_cost = tokens * pricing[model]["input"]
+    output_cost = tokens * pricing[model]["output"]
+
+    return round(input_cost + output_cost, 4)

agentic_security/probe_actor/fuzzer.py

@@ -10,6 +10,7 @@ from skopt.space import Real
 
 from agentic_security.http_spec import Modality
 from agentic_security.models.schemas import Scan, ScanResult
+from agentic_security.probe_actor.cost_module import calculate_cost
 from agentic_security.probe_actor.refusal import refusal_heuristic
 from agentic_security.probe_data import audio_generator, image_generator, msj_data
 from agentic_security.probe_data.data import prepare_prompts
@@ -38,8 +39,6 @@ def multi_modality_spec(llm_spec):
             return llm_spec
         case _:
             return llm_spec
-        # case _:
-        #     raise NotImplementedError(f"Modality {llm_spec.modality} not supported yet")
 
 
 async def process_prompt(
@@ -143,7 +142,7 @@ async def perform_single_shot_scan(
                     module_failures += 1
                 failure_rate = module_failures / max(processed_prompts, 1)
                 failure_rates.append(failure_rate)
-                cost = round(tokens * 1.5 / 1000_000, 2)
+                cost = calculate_cost(tokens)
 
                 yield ScanResult(
                     module=module.dataset_name,
@@ -274,7 +273,7 @@ async def perform_many_shot_scan(
 
                 failure_rate = module_failures / max(processed_prompts, 1)
                 failure_rates.append(failure_rate)
-                cost = round(tokens * 1.5 / 1000_000, 2)
+                cost = calculate_cost(tokens)
 
                 yield ScanResult(
                     module=module.dataset_name,

agentic_security/probe_data/__init__.py

@@ -408,6 +408,21 @@ REGISTRY = REGISTRY_V0 + [
         },
         "modality": "text",
     },
+    {
+        "dataset_name": "Reinforcement Learning Optimization",
+        "num_prompts": 0,
+        "tokens": 0,
+        "approx_cost": 0.0,
+        "source": "Cloud hosted model",
+        "selected": False,
+        "url": "",
+        "dynamic": True,
+        "opts": {
+            "port": 8718,
+            "modules": ["encoding"],
+        },
+        "modality": "text",
+    },
     {
         "dataset_name": "InspectAI",
         "num_prompts": 0,

agentic_security/probe_data/audio_generator.py

@@ -52,11 +52,37 @@ def generate_audio_mac_wav(prompt: str) -> bytes:
     return audio_bytes
 
 
+def generate_audio_cross_platform(prompt: str) -> bytes:
+    """
+    Generate an audio file from the provided prompt using gTTS for cross-platform support.
+
+    Parameters:
+        prompt (str): Text to convert into audio.
+
+    Returns:
+        bytes: The audio data in MP3 format.
+    """
+    from gtts import gTTS  # Import gTTS for cross-platform support
+
+    tts = gTTS(text=prompt, lang="en")
+    temp_mp3_path = f"temp_audio_{uuid.uuid4().hex}.mp3"
+    tts.save(temp_mp3_path)
+
+    try:
+        with open(temp_mp3_path, "rb") as f:
+            audio_bytes = f.read()
+    finally:
+        if os.path.exists(temp_mp3_path):
+            os.remove(temp_mp3_path)
+
+    return audio_bytes
+
+
 @cache_to_disk()
 def generate_audioform(prompt: str) -> bytes:
     """
     Generate an audio file from the provided prompt in WAV format.
-    Uses macOS 'say' command if the operating system is macOS.
+    Uses macOS 'say' command if the operating system is macOS, otherwise uses gTTS.
 
     Parameters:
         prompt (str): Text to convert into audio.
@@ -67,9 +93,11 @@ def generate_audioform(prompt: str) -> bytes:
     current_os = platform.system()
     if current_os == "Darwin":  # macOS
         return generate_audio_mac_wav(prompt)
+    elif current_os in ["Windows", "Linux"]:
+        return generate_audio_cross_platform(prompt)
     else:
         raise NotImplementedError(
-            "Audio generation is only supported on macOS for now."
+            "Audio generation is only supported on macOS, Windows, and Linux for now."
         )
 
 

agentic_security/probe_data/data.py

@@ -16,6 +16,7 @@ from agentic_security.probe_data.modules import (
     fine_tuned,
     garak_tool,
     inspect_ai_tool,
+    rl_model,
 )
 
 
@@ -265,6 +266,11 @@ def prepare_prompts(dataset_names, budget, tools_inbox=None, options=[]):
             garak_tool.Module(group, tools_inbox=tools_inbox, opts=opts).apply(),
             lazy=True,
         ),
+        "Reinforcement Learning Optimization": lambda opts: dataset_from_iterator(
+            "Reinforcement Learning Optimization",
+            rl_model.Module(group, tools_inbox=tools_inbox, opts=opts).apply(),
+            lazy=True,
+        ),
         "InspectAI": lambda opts: dataset_from_iterator(
             "InspectAI",
             inspect_ai_tool.Module(group, tools_inbox=tools_inbox).apply(),

agentic_security/probe_data/image_generator.py

@@ -38,12 +38,13 @@ def generate_image_dataset(
 
 
 @cache_to_disk()
-def generate_image(prompt: str) -> bytes:
+def generate_image(prompt: str, variant: int = 0) -> bytes:
     """
     Generate an image based on the provided prompt and return it as bytes.
 
     Parameters:
         prompt (str): Text to display on the generated image.
+        variant (int): The variant style of the image.
 
     Returns:
         bytes: The image data in JPG format.
@@ -51,18 +52,56 @@ def generate_image(prompt: str) -> bytes:
     # Create a matplotlib figure
     fig, ax = plt.subplots(figsize=(6, 4))
 
-    # Customize the plot (background color, text, etc.)
-    ax.set_facecolor("lightblue")
-    ax.text(
-        0.5,
-        0.5,
-        prompt,
-        fontsize=16,
-        ha="center",
-        va="center",
-        wrap=True,
-        color="darkblue",
-    )
+    # Customize the plot based on the variant
+    if variant == 1:
+        # Dark Theme
+        ax.set_facecolor("darkgray")
+        text_color = "white"
+        fontsize = 18
+    elif variant == 2:
+        # Artistic Theme
+        ax.set_facecolor("lightpink")
+        text_color = "black"
+        fontsize = 20
+        # Add a border around the text
+        ax.text(
+            0.5,
+            0.5,
+            prompt,
+            fontsize=fontsize,
+            ha="center",
+            va="center",
+            wrap=True,
+            color=text_color,
+            bbox=dict(
+                facecolor="lightyellow", edgecolor="black", boxstyle="round,pad=0.5"
+            ),
+        )
+    elif variant == 3:
+        # Minimalist Theme
+        ax.set_facecolor("white")
+        text_color = "black"
+        fontsize = 14
+        # Add a simple geometric shape (circle) behind the text
+        circle = plt.Circle((0.5, 0.5), 0.3, color="lightblue", fill=True)
+        ax.add_artist(circle)
+    else:
+        # Default Theme
+        ax.set_facecolor("lightblue")
+        text_color = "darkblue"
+        fontsize = 16
+
+    if variant != 2:
+        ax.text(
+            0.5,
+            0.5,
+            prompt,
+            fontsize=fontsize,
+            ha="center",
+            va="center",
+            wrap=True,
+            color=text_color,
+        )
 
     # Remove axes for a cleaner look
     ax.axis("off")

agentic_security/probe_data/modules/rl_model.py

@@ -0,0 +1,247 @@
+import asyncio
+import os
+import random
+import uuid as U
+from abc import ABC, abstractmethod
+from collections import deque
+from typing import Deque
+
+import numpy as np
+import requests
+from loguru import logger
+
+AUTH_TOKEN: str = os.getenv("AS_TOKEN", "gh0-5f4a8ed2-37c6-4bd7-a0cf-7070eae8115b")
+
+
+class PromptSelectionInterface(ABC):
+    """Abstract base class for prompt selection strategies."""
+
+    @abstractmethod
+    def select_next_prompt(self, current_prompt: str, passed_guard: bool) -> str:
+        """Selects the next prompt based on current state and guard result."""
+        pass
+
+    @abstractmethod
+    def select_next_prompts(self, current_prompt: str, passed_guard: bool) -> list[str]:
+        """Selects the next prompts based on current state and guard result."""
+        pass
+
+    @abstractmethod
+    def update_rewards(
+        self,
+        previous_prompt: str,
+        current_prompt: str,
+        reward: float,
+        passed_guard: bool,
+    ) -> None:
+        """Updates internal rewards based on the outcome of the last selected prompt."""
+        pass
+
+
+class RandomPromptSelector(PromptSelectionInterface):
+    """Random prompt selector with cycle prevention using history."""
+
+    def __init__(self, prompts: list[str], history_size: int = 300):
+        if not prompts:
+            raise ValueError("Prompts list cannot be empty")
+        self.prompts = prompts
+        self.history: Deque[str] = deque(maxlen=history_size)
+
+    def select_next_prompts(self, current_prompt: str, passed_guard: bool) -> list[str]:
+        return [self.select_next_prompt(current_prompt, passed_guard)]
+
+    def select_next_prompt(self, current_prompt: str, passed_guard: bool) -> str:
+        self.history.append(current_prompt)
+        available = [p for p in self.prompts if p not in self.history]
+
+        if not available:
+            available = self.prompts
+            self.history.clear()
+
+        return random.choice(available)
+
+    def update_rewards(
+        self,
+        previous_prompt: str,
+        current_prompt: str,
+        reward: float,
+        passed_guard: bool,
+    ) -> None:
+        pass  # No learning in random selection
+
+
+class CloudRLPromptSelector(PromptSelectionInterface):
+    """Cloud-based reinforcement learning prompt selector with fallback."""
+
+    def __init__(
+        self,
+        prompts: list[str],
+        api_url: str,
+        auth_token: str = AUTH_TOKEN,
+        history_size: int = 300,
+        timeout: int = 5,
+        run_id: str = "",
+    ):
+        if not prompts:
+            raise ValueError("Prompts list cannot be empty")
+        self.prompts = prompts
+        self.api_url = api_url
+        self.headers = {"Authorization": f"Bearer {auth_token}"}
+        self.timeout = timeout
+        self.run_id = run_id or U.uuid4().hex
+
+    def select_next_prompt(self, current_prompt: str, passed_guard: bool) -> list[str]:
+        return self.select_next_prompts(current_prompt, passed_guard)[0]
+
+    def select_next_prompts(self, current_prompt: str, passed_guard: bool) -> str:
+        try:
+            response = requests.post(
+                f"{self.api_url}/rl-model/select-next-prompt",
+                json={
+                    "run_id": U.uuid4().hex,
+                    "current_prompt": current_prompt,
+                    "passed_guard": passed_guard,
+                },
+                headers=self.headers,
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+            return response.json().get("next_prompts", [])
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Cloud request failed: {e}")
+            return [self._fallback_selection()]
+
+    def _fallback_selection(self) -> str:
+        return random.choice(self.prompts)
+
+    def update_rewards(
+        self,
+        previous_prompt: str,
+        current_prompt: str,
+        reward: float,
+        passed_guard: bool,
+    ) -> None:
+        ...
+
+
+class QLearningPromptSelector(PromptSelectionInterface):
+    """Q-Learning based prompt selector with exploration/exploitation tradeoff."""
+
+    def __init__(
+        self,
+        prompts: list[str],
+        learning_rate: float = 0.1,
+        discount_factor: float = 0.9,
+        initial_exploration: float = 1.0,
+        exploration_decay: float = 0.995,
+        min_exploration: float = 0.01,
+        history_size: int = 300,
+    ):
+        if not prompts:
+            raise ValueError("Prompts list cannot be empty")
+
+        self.prompts = prompts
+        self.learning_rate = learning_rate
+        self.discount_factor = discount_factor
+        self.exploration_rate = initial_exploration
+        self.exploration_decay = exploration_decay
+        self.min_exploration = min_exploration
+        self.history: Deque[str] = deque(maxlen=history_size)
+
+        # Initialize Q-table with small random values
+        self.q_table: dict[str, dict[str, float]] = {
+            state: {
+                action: np.random.uniform(0, 0.1)
+                for action in prompts
+                if action != state
+            }
+            for state in prompts
+        }
+
+    def select_next_prompts(self, current_prompt: str, passed_guard: bool) -> list[str]:
+        return [self.select_next_prompt(current_prompt, passed_guard)]
+
+    def select_next_prompt(self, current_prompt: str, passed_guard: bool) -> str:
+        self.history.append(current_prompt)
+        available = [a for a in self.prompts if a not in self.history]
+
+        if not available:
+            available = self.prompts
+            self.history.clear()
+
+        # Exploration-exploitation tradeoff
+        if np.random.random() < self.exploration_rate:
+            selected = random.choice(available)
+        else:
+            q_values = {a: self.q_table[current_prompt][a] for a in available}
+            selected = max(q_values, key=q_values.get)  # type: ignore
+
+        # Decay exploration rate
+        self.exploration_rate = max(
+            self.min_exploration, self.exploration_rate * self.exploration_decay
+        )
+        return selected
+
+    def update_rewards(
+        self,
+        previous_prompt: str,
+        current_prompt: str,
+        reward: float,
+        passed_guard: bool,
+    ) -> None:
+        if (
+            previous_prompt not in self.q_table
+            or current_prompt not in self.q_table[previous_prompt]
+        ):
+            return
+
+        # Calculate temporal difference error
+        max_future_q = max(self.q_table[current_prompt].values(), default=0.0)
+        td_target = reward + self.discount_factor * max_future_q
+        td_error = td_target - self.q_table[previous_prompt][current_prompt]
+
+        # Update Q-value
+        self.q_table[previous_prompt][current_prompt] += self.learning_rate * td_error
+
+
+class Module:
+    def __init__(
+        self, prompt_groups: list[str], tools_inbox: asyncio.Queue, opts: dict = {}
+    ):
+        self.tools_inbox = tools_inbox
+        self.opts = opts
+        self.prompt_groups = prompt_groups
+        self.max_prompts = self.opts.get("max_prompts", 10)  # Default max M prompts
+        self.run_id = U.uuid4().hex
+        self.batch_size = self.opts.get("batch_size", 500)
+        self.rl_model = CloudRLPromptSelector(
+            prompt_groups, "https://edge.metaheuristic.co", run_id=self.run_id
+        )
+
+    async def apply(self):
+        current_prompt = "What is AI?"
+        passed_guard = False
+        for _ in range(max(self.max_prompts, 1)):
+            # Fetch prompts from the API
+            prompts = await asyncio.to_thread(
+                lambda: self.rl_model.select_next_prompts(
+                    current_prompt, passed_guard=passed_guard
+                )
+            )
+
+            if not prompts:
+                logger.error("No prompts retrieved from the API.")
+                return
+
+            logger.info(f"Retrieved {len(prompts)} prompts.")
+
+            for i, prompt in enumerate(prompts):
+                logger.info(f"Processing prompt {i+1}/{len(prompts)}: {prompt}")
+                yield prompt
+                current_prompt = prompt
+                while not self.tools_inbox.empty():
+                    ref = await self.tools_inbox.get()
+                    print(ref, "ref")
+                    message, _, ready = ref["message"], ref["reply"], ref["ready"]
+                    yield message
+                    ready.set()

agentic_security/probe_data/modules/test_rl_model.py

@@ -0,0 +1,215 @@
+import asyncio
+from collections import deque
+from unittest.mock import Mock, patch
+
+import numpy as np
+import pytest
+import requests
+
+# Import the classes to be tested
+from agentic_security.probe_data.modules.rl_model import (
+    CloudRLPromptSelector,
+    Module,
+    QLearningPromptSelector,
+    RandomPromptSelector,
+)
+
+
+# Fixtures for reusable test data
+@pytest.fixture
+def dataset_prompts() -> list[str]:
+    return [
+        "What is AI?",
+        "How does RL work?",
+        "Explain supervised learning.",
+        "What is reinforcement learning?",
+    ]
+
+
+@pytest.fixture
+def mock_requests() -> Mock:
+    with patch("requests.post") as mock_requests:
+        yield mock_requests
+
+
+@pytest.fixture
+def mock_rl_selector() -> Mock:
+    return CloudRLPromptSelector(
+        dataset_prompts,
+        api_url="https://edge.metaheuristic.co",
+    )
+
+
+@pytest.fixture
+def tools_inbox() -> asyncio.Queue:
+    return asyncio.Queue()
+
+
+# Tests for RandomPromptSelector
+class TestRandomPromptSelector:
+    def test_initialization(self, dataset_prompts):
+        selector = RandomPromptSelector(dataset_prompts)
+        assert selector.prompts == dataset_prompts
+        assert isinstance(selector.history, deque)
+        assert selector.history.maxlen == 300
+
+    def test_select_next_prompt(self, dataset_prompts):
+        selector = RandomPromptSelector(dataset_prompts)
+        current_prompt = "What is AI?"
+        next_prompt = selector.select_next_prompt(current_prompt, passed_guard=True)
+        assert next_prompt in dataset_prompts
+        assert next_prompt != current_prompt
+
+    def test_update_rewards_no_op(self, dataset_prompts):
+        selector = RandomPromptSelector(dataset_prompts)
+        selector.update_rewards("What is AI?", "How does RL work?", 1.0, True)
+        assert len(selector.history) == 0
+
+
+# Tests for CloudRLPromptSelector
+class TestCloudRLPromptSelector:
+    def test_initialization(self, dataset_prompts):
+        selector = CloudRLPromptSelector(dataset_prompts, "http://example.com", "token")
+        assert selector.prompts == dataset_prompts
+        assert selector.api_url == "http://example.com"
+        assert selector.headers == {"Authorization": "Bearer token"}
+
+    def test_select_next_prompt_success(self, dataset_prompts, mock_requests):
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {"next_prompts": ["What is AI?"]}
+
+        selector = CloudRLPromptSelector(dataset_prompts, "http://example.com", "token")
+        next_prompt = selector.select_next_prompt(
+            "How does RL work?", passed_guard=True
+        )
+        assert next_prompt == "What is AI?"
+        mock_requests.assert_called_once()
+
+    def test_fallback_on_failure(self, dataset_prompts, mock_requests):
+        mock_requests.side_effect = requests.exceptions.RequestException
+        selector = CloudRLPromptSelector(dataset_prompts, "http://example.com", "token")
+        next_prompt = selector.select_next_prompt("What is AI?", passed_guard=True)
+        assert next_prompt in dataset_prompts
+
+    def test_select_next_prompt_success_service(self, dataset_prompts):
+        selector = CloudRLPromptSelector(
+            dataset_prompts,
+            api_url="https://edge.metaheuristic.co",
+        )
+        next_prompt = selector.select_next_prompt(
+            "How does RL work?", passed_guard=True
+        )
+        assert next_prompt
+
+
+# Tests for QLearningPromptSelector
+class TestQLearningPromptSelector:
+    def test_initialization(self, dataset_prompts):
+        selector = QLearningPromptSelector(dataset_prompts)
+        assert selector.prompts == dataset_prompts
+        assert selector.exploration_rate == 1.0
+        assert len(selector.q_table) == len(dataset_prompts)
+        assert all(
+            len(v) == len(dataset_prompts) - 1 for v in selector.q_table.values()
+        )
+
+    def test_select_next_prompt_exploration(self, dataset_prompts):
+        selector = QLearningPromptSelector(dataset_prompts, initial_exploration=1.0)
+        next_prompt = selector.select_next_prompt("What is AI?", passed_guard=True)
+        assert next_prompt in dataset_prompts
+        assert next_prompt != "What is AI?"
+
+    def test_select_next_prompt_exploitation(self, dataset_prompts):
+        selector = QLearningPromptSelector(dataset_prompts, initial_exploration=0.0)
+        selector.q_table["What is AI?"]["How does RL work?"] = 10.0
+        next_prompt = selector.select_next_prompt("What is AI?", passed_guard=True)
+        assert next_prompt == "How does RL work?"
+
+    def test_update_rewards(self, dataset_prompts):
+        selector = QLearningPromptSelector(dataset_prompts)
+        selector.update_rewards("What is AI?", "How does RL work?", 1.0, True)
+        assert selector.q_table["What is AI?"]["How does RL work?"] > 0.0
+
+    def test_exploration_rate_decay(self, dataset_prompts):
+        selector = QLearningPromptSelector(
+            dataset_prompts, initial_exploration=1.0, exploration_decay=0.9
+        )
+        assert selector.exploration_rate == 1.0
+        selector.select_next_prompt("What is AI?", passed_guard=True)
+        assert selector.exploration_rate == 0.9
+        selector.select_next_prompt("How does RL work?", passed_guard=True)
+        assert selector.exploration_rate == 0.81
+
+
+# Edge Cases and Error Handling
+def test_empty_prompts():
+    with pytest.raises(ValueError, match="Prompts list cannot be empty"):
+        RandomPromptSelector([])
+
+
+def test_cloud_rl_selector_invalid_url(dataset_prompts):
+    selector = CloudRLPromptSelector(dataset_prompts, "invalid_url", "token")
+    next_prompt = selector.select_next_prompt("What is AI?", passed_guard=True)
+    assert next_prompt in dataset_prompts
+
+
+def test_q_learning_selector_invalid_reward(dataset_prompts):
+    selector = QLearningPromptSelector(dataset_prompts)
+    selector.update_rewards("What is AI?", "How does RL work?", np.nan, True)
+
+
+# Tests for Module class
+class TestModule:
+    @pytest.fixture
+    def mock_uuid(self):
+        with patch("uuid.uuid4") as mock:
+            mock.return_value.hex = "test_run_id"
+            yield mock
+
+    def test_initialization(self, dataset_prompts, tools_inbox, mock_uuid):
+        module = Module(dataset_prompts, tools_inbox)
+        assert module.prompt_groups == dataset_prompts
+        assert module.tools_inbox == tools_inbox
+        assert module.max_prompts == 10
+        assert module.batch_size == 500
+        assert module.run_id == "test_run_id"
+        assert isinstance(module.rl_model, CloudRLPromptSelector)
+
+    def test_initialization_with_options(self, dataset_prompts, tools_inbox, mock_uuid):
+        opts = {
+            "max_prompts": 100,
+            "batch_size": 50,
+        }
+        module = Module(dataset_prompts, tools_inbox, opts)
+        assert module.max_prompts == 100
+        assert module.batch_size == 50
+
+    @pytest.mark.asyncio
+    async def test_apply_basic_flow(
+        self, dataset_prompts, tools_inbox, mock_rl_selector
+    ):
+        module = Module(dataset_prompts, tools_inbox)
+
+        count = 0
+        async for prompt in module.apply():
+            assert prompt
+            count += 1
+            if count >= 3:  # Test a few iterations
+                break
+
+    @pytest.mark.asyncio
+    async def test_apply_rl_with_tools_inbox(self, dataset_prompts, tools_inbox):
+        # Add a test message to the tools inbox
+        test_message = {
+            "message": "Test message",
+            "reply": None,
+            "ready": asyncio.Event(),
+        }
+        await tools_inbox.put(test_message)
+
+        module = Module(dataset_prompts, tools_inbox)
+
+        async for output in module.apply():
+            if output == "Test message":
+                test_message["ready"].set()
+                break

agentic_security/probe_data/test_audio_generator.py

@@ -3,6 +3,7 @@ import platform
 import pytest
 
 from agentic_security.probe_data.audio_generator import (
+    generate_audio_cross_platform,
     generate_audio_mac_wav,
     generate_audioform,
 )
@@ -24,6 +25,13 @@ def test_generate_audioform_mac():
         audio_bytes = generate_audioform(prompt)
         assert isinstance(audio_bytes, bytes)
         assert len(audio_bytes) > 0
+
+
+def test_generate_audio_cross_platform():
+    if platform.system() in ["Windows", "Linux"]:
+        prompt = "This is a cross-platform test."
+        audio_bytes = generate_audio_cross_platform(prompt)
+        assert isinstance(audio_bytes, bytes)
+        assert len(audio_bytes) > 0
     else:
-        with pytest.raises(NotImplementedError):
-            generate_audioform("This should raise an error on non-macOS systems.")
+        pytest.skip("Test is only applicable on Windows and Linux.")

agentic_security/probe_data/test_image_generator.py

@@ -1,5 +1,7 @@
 from unittest.mock import patch
 
+import pytest
+
 from agentic_security.probe_data.image_generator import (
     generate_image,
     generate_image_dataset,
@@ -7,9 +9,10 @@ from agentic_security.probe_data.image_generator import (
 from agentic_security.probe_data.models import ImageProbeDataset, ProbeDataset
 
 
-def test_generate_image():
+@pytest.mark.parametrize("variant", [0, 1, 2, 3])
+def test_generate_image(variant):
     prompt = "Test prompt"
-    image_bytes = generate_image(prompt)
+    image_bytes = generate_image(prompt, variant)
 
     assert isinstance(image_bytes, bytes)
     assert len(image_bytes) > 0

agentic_security/routes/scan.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from fastapi import APIRouter, BackgroundTasks, HTTPException
+from fastapi import APIRouter, BackgroundTasks, File, HTTPException, Query, UploadFile
 from fastapi.responses import StreamingResponse
 
 from ..core.app import get_stop_event, get_tools_inbox, set_current_run
@@ -52,3 +52,28 @@ async def scan(scan_parameters: Scan, background_tasks: BackgroundTasks):
 async def stop_scan():
     get_stop_event().set()
     return {"status": "Scan stopped"}
+
+
+@router.post("/scan-csv")
+async def scan_csv(
+    background_tasks: BackgroundTasks,
+    file: UploadFile = File(...),
+    llmSpec: UploadFile = File(...),
+    optimize: bool = Query(False),
+    maxBudget: int = Query(10_000),
+    enableMultiStepAttack: bool = Query(False),
+):
+    # TODO: content dataset to fuzzer
+    content = await file.read()  # noqa
+    llm_spec = await llmSpec.read()
+
+    scan_parameters = Scan(
+        llmSpec=llm_spec,
+        optimize=optimize,
+        maxBudget=1000,
+        enableMultiStepAttack=enableMultiStepAttack,
+    )
+
+    return StreamingResponse(
+        streaming_response_generator(scan_parameters), media_type="application/json"
+    )

agentic_security/routes/test_csv.py

@@ -0,0 +1,22 @@
+from fastapi.testclient import TestClient
+
+import agentic_security.test_spec_assets as test_spec_assets
+from agentic_security.routes.scan import router
+
+client = TestClient(router)
+
+
+def test_upload_csv_and_run():
+    # Create a sample CSV content
+    csv_content = "id,prompt\nspec1,value1\nspec2,value3"
+    # Send a POST request to the /upload-csv endpoint
+    response = client.post(
+        "/scan-csv?optimize=false&enableMultiStepAttack=false&maxBudget=1000",
+        files={
+            "file": ("test.csv", csv_content, "text/csv"),
+            "llmSpec": ("spec.txt", test_spec_assets.SAMPLE_SPEC, "text/plain"),
+        },
+    )
+
+    assert response.status_code == 200
+    assert "Scan completed." in response.text

agentic_security/static/index.html

@@ -437,7 +437,7 @@
                   <th class="p-3">Vulnerability Module</th>
                   <th class="p-3">% Strength</th>
                   <th class="p-3">Number of Tokens</th>
-                  <th class="p-3">Cost (in gpt-3 tokens)</th>
+                  <th class="p-3">Approx Cost (in tokens)</th>
                 </tr>
               </thead>
               <tbody>