Skip to content
Open
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
3 changes: 2 additions & 1 deletion pyrit/backend/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
import pyrit
from pyrit.backend.middleware import RequestIdMiddleware, SecurityHeadersMiddleware, register_error_handlers
from pyrit.backend.middleware.auth import EntraAuthMiddleware
from pyrit.backend.routes import attacks, auth, converters, health, labels, media, targets, version
from pyrit.backend.routes import attacks, auth, converters, health, labels, media, scenarios, targets, version
from pyrit.memory import CentralMemory

# Check for development mode from environment variable
Expand Down Expand Up @@ -85,6 +85,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
app.include_router(attacks.router, prefix="/api", tags=["attacks"])
app.include_router(targets.router, prefix="/api", tags=["targets"])
app.include_router(converters.router, prefix="/api", tags=["converters"])
app.include_router(scenarios.router, prefix="/api", tags=["scenarios"])
app.include_router(labels.router, prefix="/api", tags=["labels"])
app.include_router(health.router, prefix="/api", tags=["health"])
app.include_router(auth.router, prefix="/api", tags=["auth"])
Expand Down
7 changes: 7 additions & 0 deletions pyrit/backend/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,10 @@
CreateConverterResponse,
PreviewStep,
)
from pyrit.backend.models.scenarios import (
ScenarioListResponse,
ScenarioSummary,
)
from pyrit.backend.models.targets import (
CreateTargetRequest,
TargetInstance,
Expand Down Expand Up @@ -91,6 +95,9 @@
"CreateConverterRequest",
"CreateConverterResponse",
"PreviewStep",
# Scenarios
"ScenarioListResponse",
"ScenarioSummary",
# Targets
"CreateTargetRequest",
"TargetInstance",
Expand Down
37 changes: 37 additions & 0 deletions pyrit/backend/models/scenarios.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""
Scenario API response models.

Scenarios are multi-attack security testing campaigns. These models represent
the metadata about available scenarios (listing), not scenario execution results.
"""

from typing import Optional

from pydantic import BaseModel, Field

from pyrit.backend.models.common import PaginationInfo


class ScenarioSummary(BaseModel):
"""Summary of a registered scenario."""

scenario_name: str = Field(..., description="Registry key (e.g., 'foundry.red_team_agent')")
class_name: str = Field(..., description="Python class name (e.g., 'RedTeamAgentScenario')")
description: str = Field(..., description="Human-readable description of the scenario")
default_strategy: str = Field(..., description="Default strategy name used when none specified")
aggregate_strategies: list[str] = Field(
..., description="Aggregate strategies that combine multiple attack approaches"
)
all_strategies: list[str] = Field(..., description="All available concrete strategy names")
default_datasets: list[str] = Field(..., description="Default dataset names used by the scenario")
max_dataset_size: Optional[int] = Field(None, description="Maximum items per dataset (None means unlimited)")


class ScenarioListResponse(BaseModel):
"""Response for listing scenarios."""

items: list[ScenarioSummary] = Field(..., description="List of scenario summaries")
pagination: PaginationInfo = Field(..., description="Pagination metadata")
3 changes: 2 additions & 1 deletion pyrit/backend/routes/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,15 @@
API route handlers.
"""

from pyrit.backend.routes import attacks, converters, health, labels, media, targets, version
from pyrit.backend.routes import attacks, converters, health, labels, media, scenarios, targets, version

__all__ = [
"attacks",
"converters",
"health",
"labels",
"media",
"scenarios",
"targets",
"version",
]
68 changes: 68 additions & 0 deletions pyrit/backend/routes/scenarios.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""
Scenario API routes.

Provides endpoints for listing available scenarios and their metadata.
"""

from typing import Optional

from fastapi import APIRouter, HTTPException, Query, status

from pyrit.backend.models.common import ProblemDetail
from pyrit.backend.models.scenarios import ScenarioListResponse, ScenarioSummary
from pyrit.backend.services.scenario_service import get_scenario_service

router = APIRouter(prefix="/scenarios", tags=["scenarios"])


@router.get(
"",
response_model=ScenarioListResponse,
)
async def list_scenarios(
limit: int = Query(50, ge=1, le=200, description="Maximum items per page"),
cursor: Optional[str] = Query(None, description="Pagination cursor (scenario_name to start after)"),
) -> ScenarioListResponse:
"""
List all available scenarios.

Returns scenario metadata including strategies, datasets, and defaults.
Use GET /api/scenarios/{scenario_name} for full details on a specific scenario.

Returns:
ScenarioListResponse: Paginated list of scenario summaries.
"""
service = get_scenario_service()
return await service.list_scenarios_async(limit=limit, cursor=cursor)


@router.get(
"/{scenario_name:path}",
response_model=ScenarioSummary,
responses={
404: {"model": ProblemDetail, "description": "Scenario not found"},
},
)
async def get_scenario(scenario_name: str) -> ScenarioSummary:
"""
Get details for a specific scenario.

Args:
scenario_name: Registry name of the scenario (e.g., 'foundry.red_team_agent').

Returns:
ScenarioSummary: Full scenario metadata.
"""
service = get_scenario_service()

scenario = await service.get_scenario_async(scenario_name=scenario_name)
if not scenario:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Scenario '{scenario_name}' not found",
)

return scenario
6 changes: 6 additions & 0 deletions pyrit/backend/services/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,10 @@
ConverterService,
get_converter_service,
)
from pyrit.backend.services.scenario_service import (
ScenarioService,
get_scenario_service,
)
from pyrit.backend.services.target_service import (
TargetService,
get_target_service,
Expand All @@ -25,6 +29,8 @@
"get_attack_service",
"ConverterService",
"get_converter_service",
"ScenarioService",
"get_scenario_service",
"TargetService",
"get_target_service",
]
133 changes: 133 additions & 0 deletions pyrit/backend/services/scenario_service.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""
Scenario service for listing available scenarios.

Provides read-only access to the ScenarioRegistry, exposing scenario metadata
through the REST API.
"""

from functools import lru_cache
from typing import Optional

from pyrit.backend.models.common import PaginationInfo
from pyrit.backend.models.scenarios import ScenarioListResponse, ScenarioSummary
from pyrit.registry import ScenarioMetadata, ScenarioRegistry


def _metadata_to_summary(metadata: ScenarioMetadata) -> ScenarioSummary:
"""
Convert a ScenarioMetadata dataclass to a ScenarioSummary Pydantic model.

Args:
metadata: The registry metadata for a scenario.

Returns:
ScenarioSummary Pydantic model.
"""
return ScenarioSummary(
scenario_name=metadata.registry_name,
class_name=metadata.class_name,
description=metadata.class_description,
default_strategy=metadata.default_strategy,
aggregate_strategies=list(metadata.aggregate_strategies),
all_strategies=list(metadata.all_strategies),
default_datasets=list(metadata.default_datasets),
max_dataset_size=metadata.max_dataset_size,
)


class ScenarioService:
"""
Service for listing available scenarios.

Uses ScenarioRegistry as the source of truth for scenario metadata.
"""

def __init__(self) -> None:
"""Initialize the scenario service."""
self._registry = ScenarioRegistry.get_registry_singleton()

async def list_scenarios_async(
self,
*,
limit: int = 50,
cursor: Optional[str] = None,
) -> ScenarioListResponse:
"""
List all available scenarios with pagination.

Args:
limit: Maximum items to return per page.
cursor: Pagination cursor (scenario_name to start after).

Returns:
ScenarioListResponse with paginated scenario summaries.
"""
all_metadata = self._registry.list_metadata()
all_summaries = [_metadata_to_summary(m) for m in all_metadata]

page, has_more = self._paginate(items=all_summaries, cursor=cursor, limit=limit)
next_cursor = page[-1].scenario_name if has_more and page else None

return ScenarioListResponse(
items=page,
pagination=PaginationInfo(limit=limit, has_more=has_more, next_cursor=next_cursor, prev_cursor=cursor),
)

async def get_scenario_async(self, *, scenario_name: str) -> Optional[ScenarioSummary]:
"""
Get a single scenario by registry name.

Args:
scenario_name: The registry key of the scenario (e.g., 'foundry.red_team_agent').

Returns:
ScenarioSummary if found, None otherwise.
"""
all_metadata = self._registry.list_metadata()
for metadata in all_metadata:
if metadata.registry_name == scenario_name:
return _metadata_to_summary(metadata)
return None

@staticmethod
def _paginate(
*,
items: list[ScenarioSummary],
cursor: Optional[str],
limit: int,
) -> tuple[list[ScenarioSummary], bool]:
"""
Apply cursor-based pagination.

Args:
items: Full list of items.
cursor: Scenario name to start after.
limit: Maximum items per page.

Returns:
Tuple of (paginated items, has_more flag).
"""
start_idx = 0
if cursor:
for i, item in enumerate(items):
if item.scenario_name == cursor:
start_idx = i + 1
break

page = items[start_idx : start_idx + limit]
has_more = len(items) > start_idx + limit
return page, has_more


@lru_cache(maxsize=1)
def get_scenario_service() -> ScenarioService:
"""
Get the global scenario service instance.

Returns:
The singleton ScenarioService instance.
"""
return ScenarioService()
Loading
Loading