Skip to content
41 changes: 41 additions & 0 deletions pyrit/output/_image_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""
Internal image utilities for the output module.

Used by pretty and markdown conversation printers to apply a Gaussian blur
to images before they are displayed to a reviewer (the ``blur_images`` flag).
"""

import io
import logging

logger = logging.getLogger(__name__)


def blur_image_bytes(*, image_bytes: bytes, radius: int = 20) -> bytes:
"""
Apply a Gaussian blur to the given image bytes and return blurred PNG bytes.

Args:
image_bytes (bytes): The original encoded image bytes.
radius (int): The Gaussian blur radius. Larger values blur more.
Defaults to 20.

Returns:
bytes: The blurred image encoded as PNG. If blurring fails for any reason,
returns the original ``image_bytes`` unchanged and logs a warning.
"""
try:
from PIL import Image, ImageFilter

with Image.open(io.BytesIO(image_bytes)) as image:
image.load()
blurred = image.filter(ImageFilter.GaussianBlur(radius=radius))
buffer = io.BytesIO()
blurred.save(buffer, format="PNG")
return buffer.getvalue()
except Exception as exc:
logger.warning(f"Failed to blur image (radius={radius}); returning original bytes. Error: {exc}")
return image_bytes
38 changes: 36 additions & 2 deletions pyrit/output/attack_result/markdown.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

import os
from datetime import datetime, timezone

from pyrit.common.deprecation import print_deprecation_message
Expand All @@ -26,6 +27,9 @@ def __init__(
display_inline: bool = True,
conversation_printer: MarkdownConversationPrinter | None = None,
score_printer: MarkdownScorePrinter | None = None,
blur_images: bool = False,
blur_radius: int = 20,
blurred_dir: str | os.PathLike[str] | None = None,
) -> None:
"""
Initialize the markdown printer.
Expand All @@ -38,13 +42,23 @@ def __init__(
Defaults to a new MarkdownConversationPrinter with matching sink.
score_printer (MarkdownScorePrinter | None): Score printer.
Defaults to a new MarkdownScorePrinter with matching sink.
blur_images (bool): If True, write a blurred copy of each referenced
image and link to it instead of the original. Forwarded to the default
conversation printer when one is not supplied. Defaults to False.
blur_radius (int): Gaussian blur radius applied when ``blur_images`` is True.
Defaults to 20.
blurred_dir (str | PathLike | None): Directory to write blurred copies into
Comment thread
rlundeen2 marked this conversation as resolved.
when ``blur_images`` is True. Defaults to None (sibling of the original).
"""
super().__init__(sink=sink)
self._display_inline = display_inline
self._score_printer = score_printer or MarkdownScorePrinter(sink=sink)
self._conversation_printer = conversation_printer or MarkdownConversationPrinter(
sink=sink,
score_printer=self._score_printer,
blur_images=blur_images,
blur_radius=blur_radius,
blurred_dir=blurred_dir,
)

async def render_async(
Expand Down Expand Up @@ -322,20 +336,40 @@ class MarkdownAttackResultMemoryPrinter(MarkdownAttackResultPrinter):
All formatting logic lives in MarkdownAttackResultPrinter.
"""

def __init__(self, *, sink: Sink | None = None, display_inline: bool = True) -> None:
def __init__(
self,
*,
sink: Sink | None = None,
display_inline: bool = True,
blur_images: bool = False,
blur_radius: int = 20,
blurred_dir: str | os.PathLike[str] | None = None,
) -> None:
"""
Initialize the markdown printer with CentralMemory data source.

Args:
sink (Sink | None): Output sink. Defaults to StdoutSink().
display_inline (bool): Kept for backward compatibility but unused.
All output is routed through the sink. Defaults to True.
blur_images (bool): If True, write a blurred copy of each referenced
image and link to it instead of the original. Defaults to False.
blur_radius (int): Gaussian blur radius applied when ``blur_images`` is True.
Defaults to 20.
blurred_dir (str | PathLike | None): Directory to write blurred copies into.
Defaults to None (sibling of the original).
"""
from pyrit.memory import CentralMemory
from pyrit.output.conversation.markdown import MarkdownConversationMemoryPrinter

score_printer = MarkdownScorePrinter(sink=sink)
conversation_printer = MarkdownConversationMemoryPrinter(sink=sink, score_printer=score_printer)
conversation_printer = MarkdownConversationMemoryPrinter(
sink=sink,
score_printer=score_printer,
blur_images=blur_images,
blur_radius=blur_radius,
blurred_dir=blurred_dir,
)
super().__init__(
sink=sink,
display_inline=display_inline,
Expand Down
24 changes: 23 additions & 1 deletion pyrit/output/attack_result/pretty.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@ def __init__(
enable_colors: bool = True,
conversation_printer: PrettyConversationPrinter | None = None,
score_printer: PrettyScorePrinter | None = None,
blur_images: bool = False,
blur_radius: int = 20,
) -> None:
"""
Initialize the pretty printer.
Expand All @@ -44,6 +46,11 @@ def __init__(
Defaults to a new PrettyConversationPrinter with matching settings.
score_printer (PrettyScorePrinter | None): Score printer.
Defaults to a new PrettyScorePrinter with matching settings.
blur_images (bool): If True, apply a Gaussian blur to image outputs before
displaying them. Forwarded to the default conversation printer when one
is not supplied. Defaults to False.
blur_radius (int): Gaussian blur radius applied when ``blur_images`` is True.
Defaults to 20.
"""
super().__init__(sink=sink)
self._width = width
Expand All @@ -58,6 +65,8 @@ def __init__(
indent_size=indent_size,
enable_colors=enable_colors,
score_printer=self._score_printer,
blur_images=blur_images,
blur_radius=blur_radius,
)

def _format_colored(self, text: str, *colors: str) -> str:
Expand Down Expand Up @@ -439,7 +448,14 @@ class PrettyAttackResultMemoryPrinter(PrettyAttackResultPrinter):
"""

def __init__(
self, *, sink: Sink | None = None, width: int = 100, indent_size: int = 2, enable_colors: bool = True
self,
*,
sink: Sink | None = None,
width: int = 100,
indent_size: int = 2,
enable_colors: bool = True,
blur_images: bool = False,
blur_radius: int = 20,
) -> None:
"""
Initialize the pretty printer with CentralMemory data source.
Expand All @@ -449,6 +465,10 @@ def __init__(
width (int): Maximum width for text wrapping. Defaults to 100.
indent_size (int): Number of spaces for indentation. Defaults to 2.
enable_colors (bool): Whether to enable ANSI color output. Defaults to True.
blur_images (bool): If True, apply a Gaussian blur to image outputs before
displaying them. Defaults to False.
blur_radius (int): Gaussian blur radius applied when ``blur_images`` is True.
Defaults to 20.
"""
from pyrit.memory import CentralMemory
from pyrit.output.conversation.pretty import PrettyConversationMemoryPrinter
Expand All @@ -460,6 +480,8 @@ def __init__(
indent_size=indent_size,
enable_colors=enable_colors,
score_printer=score_printer,
blur_images=blur_images,
blur_radius=blur_radius,
)
super().__init__(
sink=sink,
Expand Down
106 changes: 104 additions & 2 deletions pyrit/output/conversation/markdown.py
Comment thread
rlundeen2 marked this conversation as resolved.
Original file line number Diff line number Diff line change
@@ -1,13 +1,17 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

import contextlib
import logging
import os

from pyrit.models import Message, MessagePiece, Score
from pyrit.output.conversation.base import ConversationPrinterBase
from pyrit.output.score.markdown import MarkdownScorePrinter
from pyrit.output.sink import Sink

logger = logging.getLogger(__name__)


class MarkdownConversationPrinter(ConversationPrinterBase):
"""
Expand All @@ -22,6 +26,9 @@ def __init__(
*,
sink: Sink | None = None,
score_printer: MarkdownScorePrinter | None = None,
blur_images: bool = False,
blur_radius: int = 20,
blurred_dir: str | os.PathLike[str] | None = None,
) -> None:
"""
Initialize the markdown conversation printer.
Expand All @@ -30,9 +37,24 @@ def __init__(
sink (Sink | None): Output sink. Defaults to StdoutSink().
score_printer (MarkdownScorePrinter | None): Score printer for inline score rendering.
Defaults to a new MarkdownScorePrinter with matching sink.
blur_images (bool): If True, write a blurred copy of each referenced image
and emit the markdown link pointing at the blurred copy. Defaults to False.

Note: blurred files are cached by path. If the original image content
changes but the blurred file already exists, the stale blurred copy
is reused. Callers are responsible for cleaning up blurred artifacts.
blur_radius (int): Gaussian blur radius applied when ``blur_images`` is True.
Defaults to 20.
blurred_dir (str | PathLike | None): Directory to write blurred copies into.
When None (default), blurred files are written as ``<stem>_blurred.png``
next to the original. When set, blurred files are written under this
directory using the original basename plus ``_blurred.png``.
"""
super().__init__(sink=sink)
self._score_printer = score_printer or MarkdownScorePrinter(sink=sink)
self._blur_images = blur_images
self._blur_radius = blur_radius
self._blurred_dir = os.fspath(blurred_dir) if blurred_dir is not None else None

async def render_async(
self,
Expand Down Expand Up @@ -181,10 +203,75 @@ def _format_image_content(self, *, image_path: str) -> list[str]:
Returns:
list[str]: Markdown lines for the image.
"""
relative_path = os.path.relpath(image_path)
display_path = self._maybe_blur_image_on_disk(image_path=image_path) if self._blur_images else image_path
try:
relative_path = os.path.relpath(display_path)
except ValueError:
# Different mount/drive than cwd (Windows). Fall back to the absolute path.
relative_path = os.path.abspath(display_path)
posix_path = relative_path.replace("\\", "/")
return [f"![Image]({posix_path})\n"]

def _maybe_blur_image_on_disk(self, *, image_path: str) -> str:
"""
Produce a blurred copy of ``image_path`` and return its path.

By default the blurred file is written as ``<stem>_blurred.png`` next to the
original. When ``blurred_dir`` was supplied to the constructor, the blurred
file is written under that directory using the original basename plus
``_blurred.png``. Existing blurred files are reused (cached by path). The
write is atomic — bytes are written to a temp sibling then ``os.replace``\\d
into place — so concurrent renders cannot observe a partial file. On any
failure the original ``image_path`` is returned and a warning is logged.

Args:
image_path (str): The path to the source image file.

Returns:
str: The path to the blurred image, or the original path on failure.
"""
try:
blurred_path = self._blurred_destination(image_path=image_path)
if os.path.exists(blurred_path):
return blurred_path
Comment thread
rlundeen2 marked this conversation as resolved.

os.makedirs(os.path.dirname(blurred_path) or ".", exist_ok=True)

from pyrit.output._image_utils import blur_image_bytes

with open(image_path, "rb") as f:
original_bytes = f.read()
blurred_bytes = blur_image_bytes(image_bytes=original_bytes, radius=self._blur_radius)

temp_path = f"{blurred_path}.tmp.{os.getpid()}"
try:
with open(temp_path, "wb") as f:
f.write(blurred_bytes)
os.replace(temp_path, blurred_path)
except Exception:
if os.path.exists(temp_path):
with contextlib.suppress(OSError):
os.remove(temp_path)
raise
return blurred_path
except Exception as exc:
logger.warning(f"Failed to write blurred image for {image_path}; using original. Error: {exc}")
return image_path

def _blurred_destination(self, *, image_path: str) -> str:
"""
Compute the destination path for a blurred copy of ``image_path``.

Args:
image_path (str): The path to the source image file.

Returns:
str: Path to the blurred file (sibling by default, or under ``blurred_dir``).
"""
directory = self._blurred_dir if self._blurred_dir is not None else os.path.dirname(image_path)
stem = os.path.splitext(os.path.basename(image_path))[0]
return os.path.join(directory, f"{stem}_blurred.png")

def _format_audio_content(self, *, audio_path: str) -> list[str]:
"""
Format audio content as HTML5 audio player.
Expand Down Expand Up @@ -273,15 +360,30 @@ def __init__(
*,
sink: Sink | None = None,
score_printer: MarkdownScorePrinter | None = None,
blur_images: bool = False,
blur_radius: int = 20,
blurred_dir: str | os.PathLike[str] | None = None,
) -> None:
"""
Initialize the markdown conversation printer with CentralMemory data source.

Args:
sink (Sink | None): Output sink. Defaults to StdoutSink().
score_printer (MarkdownScorePrinter | None): Score printer for inline score rendering.
blur_images (bool): If True, write a blurred copy next to each image and
link to it instead of the original. Defaults to False.
blur_radius (int): Gaussian blur radius applied when ``blur_images`` is True.
Defaults to 20.
blurred_dir (str | PathLike | None): Directory to write blurred copies into.
Defaults to None (sibling of the original).
"""
super().__init__(sink=sink, score_printer=score_printer)
super().__init__(
sink=sink,
score_printer=score_printer,
blur_images=blur_images,
blur_radius=blur_radius,
blurred_dir=blurred_dir,
)
from pyrit.memory import CentralMemory

self._memory = CentralMemory.get_memory_instance()
Expand Down
Loading
Loading