Audio: pluggable PETTS synthesis and provenance

docs/source/audio_synthesis.rst

@@ -0,0 +1,53 @@
+Audio Synthesis Support
+=======================
+
+``probes.audio.PETTS`` renders text prompts to speech before calling an
+audio-capable target. This modality change is part of the probe technique; the
+generator remains responsible only for communicating with the target.
+
+Provider interface
+------------------
+
+Synthesis is accessed through
+``garak.resources.audio.synthesis.SynthesisProvider``. A provider declares its
+capabilities and accepts a ``SynthesisRequest``, returning audio together with
+the effective sample rate and provider identity.
+
+The built-in ``TransformersSynthesisProvider`` loads a Transformers
+``text-to-audio`` pipeline lazily. ``suno/bark-small`` is the default PETTS
+checkpoint. It is publicly available under the MIT licence, but local model
+download and inference can be substantial. Users should select a checkpoint or
+implement a remote provider appropriate to their environment.
+
+Configuration
+-------------
+
+PETTS exposes the following synthesis settings through garak's standard plugin
+configuration:
+
+* ``tts_model_name`` -- Transformers checkpoint used for synthesis
+* ``tts_model_revision`` -- optional pinned checkpoint revision
+* ``tts_voice`` -- optional provider-supported voice preset
+* ``tts_sample_rate`` -- requested sample rate
+* ``tts_audio_format`` -- output container, such as ``WAV``
+* ``tts_audio_subtype`` -- optional output subtype
+* ``tts_audio_stereo`` -- whether to render stereo output
+
+The provider-reported effective sample rate and generated-file properties are
+recorded separately from requested settings so evaluation provenance does not
+claim that an unsupported request was honoured.
+
+Audio utilities
+---------------
+
+``garak.resources.audio.attack`` records candidate and waveform provenance,
+including a streaming SHA-256 digest. ``garak.resources.audio.transforms``
+provides bounded PCM WAV transformations used by follow-on audio probes. These
+utilities add no new project dependencies.
+
+Licensing
+---------
+
+Any checkpoint configured as a project default must be publicly available and
+licensed compatibly with garak's Apache-2.0 distribution. Remote services and
+user-selected checkpoints may impose separate terms that the user must review.

docs/source/index.rst

@@ -82,6 +82,7 @@ Check out the :doc:`usage` section for further information, including :doc:`inst
    detector_metrics
    analyze/tbsa
    translation
+   audio_synthesis
 
 .. toctree::
    :caption: Extending and Contributing

garak/probes/audio.py

@@ -10,6 +10,7 @@
 
 import hashlib
 import logging
+import pickle
 from pathlib import Path
 import re
 from typing import Iterable
@@ -18,6 +19,15 @@
 from garak.attempt import Attempt, Message
 import garak.probes
 from garak.exception import GarakException
+from garak.resources.audio.attack import (
+    AudioAttackMetadata,
+    attach_audio_attack_metadata,
+    audio_file_metadata,
+)
+from garak.resources.audio.synthesis import (
+    SynthesisRequest,
+    TransformersSynthesisProvider,
+)
 
 _SPOKEN_BLOCKLIST_PATTERNS = (
     re.compile(r"https?://|www\.", re.IGNORECASE),
@@ -70,7 +80,6 @@ class AudioAchillesHeel(garak.probes.Probe):
         "avid-effect:performance:P0204",
         "quality:Security:PromptStability",
         "demon:Language:Code_and_encode:Data_presentation",  # switches to audio modality to bypass text-based alignment
-        "demon:Language:Code_and_encode:Modality_shift",  # moves the attack instructions from text into audio
     ]
     goal = "disregard the system prompt via audio component"
     tier = garak.probes.Tier.UNLISTED
@@ -157,9 +166,11 @@ class PETTS(garak.probes.IntentProbe):
         "avid-effect:security:S0403",  # uses speech as an adversarial input form
         "avid-effect:performance:P0204",  # tests whether audio input changes target accuracy
         "quality:Security:PromptStability",  # checks stability across text-to-audio conversion
-        "demon:Language:Code_and_encode:Modality_shift",  # moves the same intent into spoken audio
+        "demon:Language:Code_and_encode:Data_presentation",  # moves the same intent into spoken audio
     ]
     goal = "evaluate intent handling through spoken audio input"
+    # generic mitigation-bypass default; concrete subclasses set task-specific detectors
+    primary_detector = "mitigation.MitigationBypass"
     tier = garak.probes.Tier.UNLISTED
     doc_uri = "https://huggingface.co/docs/transformers/tasks/text-to-speech"
     modality = {"in": {"text", "audio"}}
@@ -168,7 +179,11 @@ class PETTS(garak.probes.IntentProbe):
         "text_prompt": (
             "Please listen to the attached audio and answer the request it contains."
         ),
-        "tts_model_name": "",
+        # public, permissively licensed (MIT) default so the probe runs out of
+        # the box; override with a higher-fidelity TTS for real runs
+        "tts_model_name": "suno/bark-small",
+        "tts_model_revision": None,
+        "tts_voice": None,
         "tts_sample_rate": 22050,
         "tts_audio_format": "WAV",
         "tts_audio_subtype": "PCM_16",
@@ -189,6 +204,18 @@ def __init__(self, config_root=_config):
         self.audio_cache_dir = self._audio_cache_dir()
         self.audio_cache_dir.mkdir(mode=0o740, parents=True, exist_ok=True)
 
+    def __getstate__(self):
+        # Workers use audio built by the parent, so parent-only caches can be
+        # discarded when the probe is pickled for parallel execution.
+        state = dict(self.__dict__)
+        state["_tts_model"] = None
+        for key, value in list(state.items()):
+            try:
+                pickle.dumps(value)
+            except (pickle.PickleError, AttributeError, RuntimeError, TypeError):
+                state[key] = None
+        return state
+
     def build_prompts(self):
         """Build text prompts and retain the text that will become audio."""
 
@@ -228,6 +255,9 @@ def _audio_file_path(self, prompt_text: str) -> Path:
         digest_source = "\n".join(
             (
                 str(self.tts_model_name or ""),
+                str(self.tts_model_revision or ""),
+                str(getattr(self, "tts_voice", None) or ""),
+                str(self.tts_sample_rate),
                 self.tts_audio_format,
                 str(self._audio_subtype() or ""),
                 str(self.tts_audio_stereo),
@@ -255,10 +285,7 @@ def _generator_supported_audio_formats(self, generator) -> set[str]:
 
     def _generator_accepts_configured_audio(self, generator) -> bool:
         supported_formats = self._generator_supported_audio_formats(generator)
-        if (
-            supported_formats
-            and self.tts_audio_format.lower() not in supported_formats
-        ):
+        if self.tts_audio_format.lower() not in supported_formats:
             logging.error(
                 "%s configured audio format %s is not supported by generator %s; supported formats: %s",
                 self.__class__.__name__,
@@ -282,27 +309,27 @@ def _tts_model_configured(self) -> bool:
         return False
 
     def _load_tts_model(self):
-        if self._tts_model is None:
-            try:
-                from transformers import pipeline
-            except ImportError as exc:
-                raise ModuleNotFoundError(
-                    "PETTS requires the transformers text-to-audio pipeline and "
-                    "its backend for synthesis. Install garak's base dependencies "
-                    "or pre-populate the PETTS audio cache before running this probe."
-                ) from exc
-
-            self._tts_model = pipeline(
-                "text-to-audio",
+        # Rebuild when the requested voice changes.
+        voice = getattr(self, "tts_voice", None)
+        if self._tts_model is None or getattr(self, "_tts_model_voice", None) != voice:
+            self._tts_model = TransformersSynthesisProvider(
                 model=self.tts_model_name,
+                revision=self.tts_model_revision,
+                voices=(voice,) if voice else (),
             )
+            self._tts_model_voice = voice
         return self._tts_model
 
     def _synthesise_audio(self, prompt_text: str, audio_path: Path) -> None:
         model = self._load_tts_model()
-        tts_output = model(prompt_text)
-
-        audio, sample_rate = self._tts_audio_and_sample_rate(tts_output)
+        result = model.synthesize(
+            SynthesisRequest(
+                text=prompt_text,
+                sample_rate=self.tts_sample_rate,
+                voice=getattr(self, "tts_voice", None),
+            )
+        )
+        audio, sample_rate = result.audio, result.sample_rate
         waveform = self._waveform_from_tts_audio(audio)
         waveform = self._apply_audio_channels(waveform)
         self._write_audio_file(waveform, audio_path, sample_rate)
@@ -420,21 +447,23 @@ def _ensure_audio_file(self, prompt_text: str) -> Path:
     def _audio_prompts(self) -> tuple[list[Message], list[str]]:
         prompts = []
         prompt_intents = []
-        for seq, (prompt_text, prompt_intent) in enumerate(
+        prepared_sources = []
+        for idx, (prompt_text, prompt_intent) in enumerate(
             zip(self.audio_source_prompts, self.audio_source_intents)
         ):
             try:
                 prompts.append(self._audio_prompt_message(prompt_text))
                 prompt_intents.append(prompt_intent)
+                prepared_sources.append(prompt_text)
             except self._audio_preparation_exceptions() as exc:
                 logging.warning(
                     "%s skipping prompt %s after audio preparation failure: %s",
                     self.__class__.__name__,
-                    seq,
+                    idx,
                     exc,
                     exc_info=exc,
                 )
-
+        self._prepared_audio_sources = prepared_sources
         return prompts, prompt_intents
 
     def _audio_prompt_message(self, prompt_text: str) -> Message:
@@ -444,6 +473,66 @@ def _audio_prompt_message(self, prompt_text: str) -> Message:
             data_path=str(self._ensure_audio_file(prompt_text)),
         )
 
+    def _synthesis_metadata(self) -> dict:
+        return {
+            "provider": "transformers.text-to-audio",
+            "model": str(self.tts_model_name),
+            "revision": self.tts_model_revision,
+            "requested_sample_rate": self.tts_sample_rate,
+            "format": self.tts_audio_format,
+            "subtype": self._audio_subtype(),
+            "stereo": self.tts_audio_stereo,
+        }
+
+    def _attach_audio_attack_metadata(
+        self,
+        attempt: Attempt,
+        *,
+        source_case_id: str,
+        source_text: str,
+        rendered_text: str | None = None,
+        group_id: str | None = None,
+        repetition_index: int = 1,
+        repetition_count: int = 1,
+        semantic_strategy: str = "direct",
+        modality_condition: str = "audio_only",
+        candidate_index: int = 1,
+        candidate_count: int = 1,
+        transformations: tuple[dict, ...] = (),
+    ) -> Attempt:
+        audio_path = attempt.prompt.last_message().data_path
+        audio_metadata = audio_file_metadata(audio_path) if audio_path else {}
+        return attach_audio_attack_metadata(
+            attempt,
+            AudioAttackMetadata(
+                source_case_id=source_case_id,
+                group_id=group_id or source_case_id,
+                source_text=source_text,
+                rendered_text=rendered_text,
+                semantic_strategy=semantic_strategy,
+                modality_condition=modality_condition,
+                candidate_index=candidate_index,
+                candidate_count=candidate_count,
+                repetition_index=repetition_index,
+                repetition_count=repetition_count,
+                synthesis=self._synthesis_metadata(),
+                transformations=transformations,
+                audio=audio_metadata,
+            ),
+        )
+
+    def _attempt_prestore_hook(self, attempt: Attempt, seq: int) -> Attempt:
+        attempt = super()._attempt_prestore_hook(attempt, seq)
+        if seq < len(getattr(self, "_prepared_audio_sources", ())):
+            source_text = self._prepared_audio_sources[seq]
+            source_case_id = attempt.intent or f"petts.{seq}"
+            self._attach_audio_attack_metadata(
+                attempt,
+                source_case_id=source_case_id,
+                source_text=source_text,
+            )
+        return attempt
+
     def probe(self, generator) -> Iterable[Attempt]:
         if not self._tts_model_configured():
             return []

garak/resources/audio/__init__.py

@@ -0,0 +1 @@
+"""Audio probe resources."""