[FEAT]: Add Base64Converter and Rot13Converter payload text transforms

Author: sy0u1ti

rampart/converters/__init__.py

@@ -8,6 +8,8 @@
 
 from __future__ import annotations
 
+from rampart.converters.base64 import Base64Converter
 from rampart.converters.docx import DocxConverter
+from rampart.converters.rot13 import Rot13Converter
 
-__all__ = ["DocxConverter"]
+__all__ = ["Base64Converter", "DocxConverter", "Rot13Converter"]

rampart/converters/base64.py

@@ -0,0 +1,89 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""Base64Converter — encode text payloads into Base64 format.
+
+Adapts PyRIT's ``Base64Converter`` to RAMPART's ``PayloadConverter``
+protocol. Converts a text ``Payload`` into a Base64-encoded text ``Payload``.
+
+PyRIT types do not leak into the public interface — callers interact
+only with RAMPART's ``Payload`` and ``PayloadConverter`` protocol.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from rampart.core.types import Payload, PayloadFormat
+
+if TYPE_CHECKING:
+    from pyrit.prompt_converter.base64_converter import (
+        Base64Converter as _PyritBase64Converter,
+    )
+
+
+class Base64Converter:
+    """Encode text payloads into Base64.
+
+    Thin wrapper around PyRIT's ``Base64Converter``. Accepts a
+    RAMPART ``Payload`` (text format) and returns a new ``Payload``
+    with ``format=TEXT``, encoded ``content``, and ``artifact=None``.
+
+    PyRIT's import chain is heavy, so initialization is deferred until
+    the first ``convert_async`` call.
+    """
+
+    def __init__(self) -> None:
+        """Initialize with deferred PyRIT converter."""
+        self._pyrit_converter: _PyritBase64Converter | None = None
+
+    def _get_converter(self) -> _PyritBase64Converter:
+        """Lazily import and instantiate the PyRIT converter.
+
+        Returns:
+            _PyritBase64Converter: The PyRIT Base64Converter instance, either
+                cached or newly created on first call.
+        """
+        if self._pyrit_converter is None:
+            from pyrit.prompt_converter.base64_converter import (  # ruff: ignore[import-outside-top-level]
+                Base64Converter as _Converter,
+            )
+
+            self._pyrit_converter = _Converter()
+        return self._pyrit_converter
+
+    async def convert_async(self, *, payload: Payload) -> Payload:
+        """Convert a text payload into a Base64-encoded text payload.
+
+        Delegates encoding to PyRIT's ``Base64Converter``.
+        Preserves ``payload.id`` for traceability and carries forward metadata.
+
+        Args:
+            payload (Payload): A text-format payload to convert.
+
+        Returns:
+            Payload: A new payload with ``format=TEXT`` and encoded content.
+
+        Raises:
+            ValueError: If the payload format is not a text format.
+        """
+        if not payload.format.is_text:
+            msg = (
+                f"Base64Converter requires a text payload, got {payload.format.value}."
+            )
+            raise ValueError(msg)
+
+        result = await self._get_converter().convert_async(
+            prompt=payload.content,
+            input_type="text",
+        )
+
+        metadata = {**payload.metadata, "converter": "Base64Converter"}
+
+        return Payload(
+            content=result.output_text,
+            id=payload.id,
+            format=PayloadFormat.TEXT,
+            artifact=None,
+            metadata=metadata,
+        )

rampart/converters/rot13.py

@@ -0,0 +1,87 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""Rot13Converter — obfuscate text payloads with ROT13 cipher.
+
+Adapts PyRIT's ``ROT13Converter`` to RAMPART's ``PayloadConverter``
+protocol. Converts a text ``Payload`` into a ROT13-obfuscated text ``Payload``.
+
+PyRIT types do not leak into the public interface — callers interact
+only with RAMPART's ``Payload`` and ``PayloadConverter`` protocol.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from rampart.core.types import Payload, PayloadFormat
+
+if TYPE_CHECKING:
+    from pyrit.prompt_converter.rot13_converter import (
+        ROT13Converter as _PyritROT13Converter,
+    )
+
+
+class Rot13Converter:
+    """Obfuscate text payloads using the ROT13 cipher.
+
+    Thin wrapper around PyRIT's ``ROT13Converter``. Accepts a
+    RAMPART ``Payload`` (text format) and returns a new ``Payload``
+    with ``format=TEXT``, ROT13-encoded ``content``, and ``artifact=None``.
+
+    PyRIT's import chain is heavy, so initialization is deferred until
+    the first ``convert_async`` call.
+    """
+
+    def __init__(self) -> None:
+        """Initialize with deferred PyRIT converter."""
+        self._pyrit_converter: _PyritROT13Converter | None = None
+
+    def _get_converter(self) -> _PyritROT13Converter:
+        """Lazily import and instantiate the PyRIT converter.
+
+        Returns:
+            _PyritROT13Converter: The PyRIT ROT13Converter instance, either
+                cached or newly created on first call.
+        """
+        if self._pyrit_converter is None:
+            from pyrit.prompt_converter.rot13_converter import (  # ruff: ignore[import-outside-top-level]
+                ROT13Converter as _Converter,
+            )
+
+            self._pyrit_converter = _Converter()
+        return self._pyrit_converter
+
+    async def convert_async(self, *, payload: Payload) -> Payload:
+        """Convert a text payload into a ROT13-obfuscated text payload.
+
+        Delegates transformation to PyRIT's ``ROT13Converter``.
+        Preserves ``payload.id`` for traceability and carries forward metadata.
+
+        Args:
+            payload (Payload): A text-format payload to convert.
+
+        Returns:
+            Payload: A new payload with ``format=TEXT`` and ROT13-transformed content.
+
+        Raises:
+            ValueError: If the payload format is not a text format.
+        """
+        if not payload.format.is_text:
+            msg = f"Rot13Converter requires a text payload, got {payload.format.value}."
+            raise ValueError(msg)
+
+        result = await self._get_converter().convert_async(
+            prompt=payload.content,
+            input_type="text",
+        )
+
+        metadata = {**payload.metadata, "converter": "Rot13Converter"}
+
+        return Payload(
+            content=result.output_text,
+            id=payload.id,
+            format=PayloadFormat.TEXT,
+            artifact=None,
+            metadata=metadata,
+        )

tests/unit/converters/test_base64.py

@@ -0,0 +1,119 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""Tests for Base64Converter."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock, MagicMock, patch
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+import pytest
+
+from rampart.converters.base64 import Base64Converter
+from rampart.core.types import Payload, PayloadFormat
+
+_PATCH_TARGET = "pyrit.prompt_converter.base64_converter.Base64Converter"
+
+
+def _text_payload(content: str = "test content", payload_id: str = "p-1") -> Payload:
+    return Payload(content=content, id=payload_id)
+
+
+class TestBase64ConverterInit:
+    """Construction defers PyRIT import until first use."""
+
+    def test_no_pyrit_import_at_construction(self) -> None:
+        with patch(_PATCH_TARGET) as mock_cls:
+            Base64Converter()
+            mock_cls.assert_not_called()
+
+    async def test_creates_pyrit_converter_on_first_use_async(self) -> None:
+        mock_result = MagicMock(output_text="dGVzdA==", output_type="text")
+
+        with patch(_PATCH_TARGET) as mock_cls:
+            mock_cls.return_value.convert_async = AsyncMock(return_value=mock_result)
+            converter = Base64Converter()
+            await converter.convert_async(payload=_text_payload())
+            mock_cls.assert_called_once()
+
+
+class TestBase64ConverterConversion:
+    """Conversion delegates to PyRIT Base64Converter and maps result."""
+
+    async def test_produces_text_payload_async(self) -> None:
+        mock_result = MagicMock(output_text="dGVzdA==", output_type="text")
+
+        with patch(_PATCH_TARGET) as mock_cls:
+            mock_cls.return_value.convert_async = AsyncMock(return_value=mock_result)
+            converter = Base64Converter()
+            result = await converter.convert_async(payload=_text_payload())
+
+        assert result.format is PayloadFormat.TEXT
+        assert result.artifact is None
+        assert result.content == "dGVzdA=="
+
+    async def test_preserves_id_async(self) -> None:
+        mock_result = MagicMock(output_text="dGVzdA==", output_type="text")
+
+        with patch(_PATCH_TARGET) as mock_cls:
+            mock_cls.return_value.convert_async = AsyncMock(return_value=mock_result)
+            converter = Base64Converter()
+            result = await converter.convert_async(
+                payload=_text_payload(payload_id="keep-me")
+            )
+
+        assert result.id == "keep-me"
+
+    async def test_metadata_includes_converter_name_async(self) -> None:
+        mock_result = MagicMock(output_text="dGVzdA==", output_type="text")
+
+        with patch(_PATCH_TARGET) as mock_cls:
+            mock_cls.return_value.convert_async = AsyncMock(return_value=mock_result)
+            converter = Base64Converter()
+            result = await converter.convert_async(payload=_text_payload())
+
+        assert result.metadata["converter"] == "Base64Converter"
+
+    async def test_source_metadata_carried_forward_async(self) -> None:
+        mock_result = MagicMock(output_text="dGVzdA==", output_type="text")
+
+        with patch(_PATCH_TARGET) as mock_cls:
+            mock_cls.return_value.convert_async = AsyncMock(return_value=mock_result)
+            converter = Base64Converter()
+            source = Payload(content="x", id="m-1", metadata={"origin": "adversarial"})
+            result = await converter.convert_async(payload=source)
+
+        assert result.metadata["origin"] == "adversarial"
+        assert result.metadata["converter"] == "Base64Converter"
+
+
+class TestBase64ConverterValidation:
+    """Input validation."""
+
+    async def test_rejects_binary_payload_async(self, tmp_path: Path) -> None:
+        artifact = tmp_path / "existing.docx"
+        artifact.write_bytes(b"PK")
+
+        binary_payload = Payload(
+            content="already docx",
+            format=PayloadFormat.DOCX,
+            artifact=artifact,
+        )
+
+        converter = Base64Converter()
+        with pytest.raises(ValueError, match="text payload"):
+            await converter.convert_async(payload=binary_payload)
+
+
+class TestBase64ConverterProtocol:
+    """Verify the converter satisfies PayloadConverter protocol."""
+
+    def test_satisfies_protocol(self) -> None:
+        from rampart.core.converter import PayloadConverter
+
+        converter = Base64Converter()
+        assert isinstance(converter, PayloadConverter)