MAINT: Frontend core refactor

pyproject.toml

@@ -157,7 +157,7 @@ all = [
 ]
 
 [project.scripts]
-pyrit_backend = "pyrit.cli.pyrit_backend:main"
+pyrit_backend = "pyrit.backend.pyrit_backend:main"
 pyrit_scan = "pyrit.cli.pyrit_scan:main"
 pyrit_shell = "pyrit.cli.pyrit_shell:main"
 

pyrit/backend/main.py

@@ -30,7 +30,6 @@
     targets,
     version,
 )
-from pyrit.memory import CentralMemory
 
 # Check for development mode from environment variable
 DEV_MODE = os.getenv("PYRIT_DEV_MODE", "false").lower() == "true"
@@ -40,17 +39,40 @@
 
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
-    """Manage application startup and shutdown lifecycle."""
-    # Initialization is handled by the pyrit_backend CLI before uvicorn starts.
-    # Running 'uvicorn pyrit.backend.main:app' directly is not supported;
-    # use 'pyrit_backend' instead.
-    try:
-        CentralMemory.get_memory_instance()
-    except ValueError:
-        logger.warning(
-            "CentralMemory is not initialized. "
-            "Start the server via 'pyrit_backend' CLI instead of running uvicorn directly."
-        )
+    """
+    Initialize PyRIT on startup using the config file, then yield.
+
+    Config resolution order:
+    1. ``PYRIT_CONFIG_FILE`` env var (if set)
+    2. ``~/.pyrit/.pyrit_conf`` (if it exists)
+    3. Built-in defaults (SQLite, no initializers)
+    """
+    from pyrit.setup.configuration_loader import ConfigurationLoader
+
+    config_file_env = os.getenv("PYRIT_CONFIG_FILE")
+    config_file = Path(config_file_env) if config_file_env else None
+
+    config = ConfigurationLoader.load_with_overrides(config_file=config_file)
+    await config.initialize_pyrit_async()
+
+    # Expose config values to route handlers via app.state
+    default_labels: dict[str, str] = {}
+    if config.operator:
+        default_labels["operator"] = config.operator
+    if config.operation:
+        default_labels["operation"] = config.operation
+    app.state.default_labels = default_labels
+    app.state.max_concurrent_scenario_runs = config.max_concurrent_scenario_runs
+    app.state.allow_custom_initializers = config.allow_custom_initializers
+
+    if config.allow_custom_initializers:
+        logger.warning("Custom initializer registration is ENABLED (allow_custom_initializers: true).")
+
+    # Mount the bundled frontend (or print a dev/missing-frontend notice).
+    # Done here rather than at module load so test imports of `pyrit.backend.main`
+    # don't emit noise and don't perform filesystem side effects.
+    setup_frontend()
+
     yield
 
 
@@ -124,7 +146,3 @@ def setup_frontend() -> None:
         print("   The frontend must be built and included in the package.")
         print("   Run: python build_scripts/prepare_package.py")
         print("   API endpoints will still work but the UI won't be available.")
-
-
-# Set up frontend at module load time (needed when running via uvicorn)
-setup_frontend()

pyrit/backend/models/scenarios.py

@@ -14,7 +14,6 @@
 
 from pydantic import BaseModel, Field
 
-from pyrit.backend.models.attacks import AttackSummary
 from pyrit.backend.models.common import PaginationInfo
 
 
@@ -124,28 +123,3 @@ class ScenarioRunListResponse(BaseModel):
     """Response for listing scenario runs."""
 
     items: list[ScenarioRunSummary] = Field(..., description="List of scenario runs")
-
-
-# ============================================================================
-# Scenario Results Detail Models
-# ============================================================================
-
-
-class AtomicAttackResults(BaseModel):
-    """Results grouped by atomic attack name."""
-
-    atomic_attack_name: str = Field(..., description="Name of the atomic attack (strategy)")
-    display_group: str | None = Field(None, description="Display group label for UI grouping")
-    results: list[AttackSummary] = Field(..., description="Individual attack results")
-    success_count: int = Field(0, ge=0, description="Number of successful attacks")
-    failure_count: int = Field(0, ge=0, description="Number of failed attacks")
-    total_count: int = Field(0, ge=0, description="Total number of attack results")
-    total_retries: int = Field(0, ge=0, description="Sum of retries across all attacks in this group")
-    error_count: int = Field(0, ge=0, description="Number of attacks with errors")
-
-
-class ScenarioRunDetail(BaseModel):
-    """Full detailed results of a scenario run."""
-
-    run: ScenarioRunSummary = Field(..., description="The scenario run summary")
-    attacks: list[AtomicAttackResults] = Field(..., description="Results grouped by atomic attack")

pyrit/backend/pyrit_backend.py

@@ -0,0 +1,132 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+PyRIT Backend CLI - Thin wrapper around uvicorn for the PyRIT backend server.
+
+All initialization (config loading, memory setup, initializer execution) is
+handled by the FastAPI lifespan in ``pyrit.backend.main``.  This CLI simply
+parses host/port/config-file/log-level/reload and starts uvicorn.
+
+The config file path is forwarded to the app via the ``PYRIT_CONFIG_FILE``
+environment variable.
+"""
+
+import logging
+import os
+import sys
+from argparse import ArgumentParser, Namespace, RawDescriptionHelpFormatter
+from pathlib import Path
+from typing import Optional
+
+from pyrit.common.cli_helpers import CONFIG_FILE_HELP, validate_log_level_argparse
+
+
+def parse_args(*, args: Optional[list[str]] = None) -> Namespace:
+    """
+    Parse command-line arguments for the PyRIT backend server.
+
+    Returns:
+        Namespace: Parsed command-line arguments.
+    """
+    parser = ArgumentParser(
+        prog="pyrit_backend",
+        description="""PyRIT Backend - Run the PyRIT backend API server
+
+All configuration (database, initializers, env-files, etc.) is read from
+the config file (~/.pyrit/.pyrit_conf by default, or --config-file).
+
+Examples:
+  # Start backend with default settings
+  pyrit_backend
+
+  # Start with a custom config file
+  pyrit_backend --config-file ./my_config.yaml
+
+  # Start with custom port and host
+  pyrit_backend --host 0.0.0.0 --port 8080
+
+  # Start with auto-reload for development
+  pyrit_backend --reload
+""",
+        formatter_class=RawDescriptionHelpFormatter,
+    )
+
+    parser.add_argument(
+        "--host",
+        type=str,
+        default="localhost",
+        help="Host to bind the server to (default: localhost)",
+    )
+
+    parser.add_argument(
+        "--port",
+        type=int,
+        default=8000,
+        help="Port to bind the server to (default: 8000)",
+    )
+
+    parser.add_argument(
+        "--config-file",
+        type=Path,
+        help=CONFIG_FILE_HELP,
+    )
+
+    parser.add_argument(
+        "--log-level",
+        type=validate_log_level_argparse,
+        default="WARNING",
+        help="Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) (default: WARNING)",
+    )
+
+    parser.add_argument(
+        "--reload",
+        action="store_true",
+        help="Enable auto-reload for development (watches for file changes)",
+    )
+
+    return parser.parse_args(args)
+
+
+def main(*, args: Optional[list[str]] = None) -> int:
+    """
+    Start the PyRIT backend server.
+
+    Returns:
+        int: Exit code (0 for success, 1 for error).
+    """
+    sys.stdout.reconfigure(errors="replace")  # type: ignore[ty:unresolved-attribute]
+    sys.stderr.reconfigure(errors="replace")  # type: ignore[ty:unresolved-attribute]
+
+    try:
+        parsed_args = parse_args(args=args)
+    except SystemExit as e:
+        return e.code if isinstance(e.code, int) else 1
+
+    # Forward config file to the FastAPI lifespan via env var
+    if parsed_args.config_file is not None:
+        os.environ["PYRIT_CONFIG_FILE"] = str(parsed_args.config_file)
+
+    try:
+        import uvicorn
+
+        uvicorn.run(
+            "pyrit.backend.main:app",
+            host=parsed_args.host,
+            port=parsed_args.port,
+            log_level=logging.getLevelName(parsed_args.log_level).lower()
+            if isinstance(parsed_args.log_level, int)
+            else parsed_args.log_level.lower(),
+            reload=parsed_args.reload,
+        )
+        return 0
+    except KeyboardInterrupt:
+        print("\n🛑 Backend stopped")
+        return 0
+    except Exception as e:
+        print(f"\nError: {e}")
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

pyrit/backend/routes/scenarios.py

@@ -21,7 +21,6 @@
     ListRegisteredScenariosResponse,
     RegisteredScenario,
     RunScenarioRequest,
-    ScenarioRunDetail,
     ScenarioRunListResponse,
     ScenarioRunSummary,
 )
@@ -197,24 +196,22 @@ async def cancel_scenario_run(scenario_result_id: str) -> ScenarioRunSummary:
 
 @router.get(
     "/runs/{scenario_result_id}/results",
-    response_model=ScenarioRunDetail,
     responses={
         404: {"model": ProblemDetail, "description": "Run not found"},
         409: {"model": ProblemDetail, "description": "Run not yet completed"},
     },
 )
-async def get_scenario_run_results(scenario_result_id: str) -> ScenarioRunDetail:
+async def get_scenario_run_results(scenario_result_id: str) -> dict:
     """
     Get detailed results for a completed scenario run.
 
-    Returns per-attack outcomes including objectives, responses, scores,
-    and success/failure counts.
+    Returns the full ScenarioResult serialization.
 
     Args:
         scenario_result_id: The scenario_result_id.
 
     Returns:
-        ScenarioRunDetail: Full attack-level results.
+        dict: ScenarioResult.to_dict() payload.
     """
     service = get_scenario_run_service()
     try:
@@ -227,4 +224,4 @@ async def get_scenario_run_results(scenario_result_id: str) -> ScenarioRunDetail
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"Scenario run '{scenario_result_id}' not found",
         )
-    return result
+    return result.to_dict()