REFACTOR: Promote single-class module constants to ClassVar (backend + registry + setup)

pyrit/backend/middleware/auth.py

@@ -18,7 +18,7 @@
 import logging
 import os
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, ClassVar
 
 import httpx
 import jwt
@@ -30,13 +30,6 @@
 
 logger = logging.getLogger(__name__)
 
-# Paths that bypass authentication
-_PUBLIC_PATHS = {
-    "/api/health",
-    "/api/auth/config",
-    "/api/media",
-}
-
 
 @dataclass
 class AuthenticatedUser:
@@ -51,6 +44,13 @@ class AuthenticatedUser:
 class EntraAuthMiddleware(BaseHTTPMiddleware):
     """Validate Entra ID JWTs on every request (except public paths)."""
 
+    # Paths that bypass authentication
+    _PUBLIC_PATHS: ClassVar[set[str]] = {
+        "/api/health",
+        "/api/auth/config",
+        "/api/media",
+    }
+
     def __init__(self, app: ASGIApp) -> None:
         """Initialize the middleware with Entra ID configuration from environment variables."""
         super().__init__(app)
@@ -88,7 +88,7 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -
         """
         # Skip auth for public paths and static files
         path = request.url.path
-        if not self._enabled or path in _PUBLIC_PATHS or not path.startswith("/api"):
+        if not self._enabled or path in self._PUBLIC_PATHS or not path.startswith("/api"):
             return await call_next(request)
 
         result = await self._authenticate_request_async(request)

pyrit/backend/middleware/security_headers.py

@@ -15,6 +15,7 @@
 """
 
 import logging
+from typing import ClassVar
 
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
 from starlette.requests import Request
@@ -23,30 +24,30 @@
 
 logger = logging.getLogger(__name__)
 
-# Swagger / ReDoc paths — these load scripts and styles from CDN,
-# so CSP is skipped in dev mode (production disables these routes entirely).
-_DOCS_PATHS = {"/docs", "/redoc", "/openapi.json"}
-
-# CSP for API responses — as strict as possible.
-_API_CSP = "default-src 'none'; frame-ancestors 'none'"
-
-# CSP for frontend SPA — allows self-hosted scripts and Fluent UI / Griffel
-# runtime style injection ('unsafe-inline' for style-src only).
-_FRONTEND_CSP = (
-    "default-src 'self'; "
-    "script-src 'self'; "
-    "style-src 'self' 'unsafe-inline'; "
-    "img-src 'self' data: https://*.blob.core.windows.net; "
-    "media-src 'self' https://*.blob.core.windows.net; "
-    "font-src 'self' data:; "
-    "connect-src 'self' https://login.microsoftonline.com; "
-    "frame-ancestors 'none'"
-)
-
 
 class SecurityHeadersMiddleware(BaseHTTPMiddleware):
     """Inject security response headers on every request."""
 
+    # Swagger / ReDoc paths — these load scripts and styles from CDN,
+    # so CSP is skipped in dev mode (production disables these routes entirely).
+    _DOCS_PATHS: ClassVar[set[str]] = {"/docs", "/redoc", "/openapi.json"}
+
+    # CSP for API responses — as strict as possible.
+    _API_CSP: ClassVar[str] = "default-src 'none'; frame-ancestors 'none'"
+
+    # CSP for frontend SPA — allows self-hosted scripts and Fluent UI / Griffel
+    # runtime style injection ('unsafe-inline' for style-src only).
+    _FRONTEND_CSP: ClassVar[str] = (
+        "default-src 'self'; "
+        "script-src 'self'; "
+        "style-src 'self' 'unsafe-inline'; "
+        "img-src 'self' data: https://*.blob.core.windows.net; "
+        "media-src 'self' https://*.blob.core.windows.net; "
+        "font-src 'self' data:; "
+        "connect-src 'self' https://login.microsoftonline.com; "
+        "frame-ancestors 'none'"
+    )
+
     def __init__(self, app: ASGIApp, dev_mode: bool = False) -> None:
         """
         Initialize the middleware.
@@ -84,11 +85,11 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -
 
         # --- Path-dependent headers ---
         if path.startswith("/api"):
-            response.headers["Content-Security-Policy"] = _API_CSP
+            response.headers["Content-Security-Policy"] = self._API_CSP
             response.headers["Cache-Control"] = "no-store"
-        elif self._dev_mode and path in _DOCS_PATHS:
+        elif self._dev_mode and path in self._DOCS_PATHS:
             pass  # No CSP — Swagger/ReDoc load from CDN
         else:
-            response.headers["Content-Security-Policy"] = _FRONTEND_CSP
+            response.headers["Content-Security-Policy"] = self._FRONTEND_CSP
 
         return response

pyrit/backend/services/converter_service.py

@@ -19,7 +19,7 @@
 import uuid
 from functools import lru_cache
 from pathlib import Path
-from typing import Any, Literal, Union, get_args, get_origin
+from typing import Any, ClassVar, Literal, Union, get_args, get_origin
 from urllib.parse import parse_qs, urlparse
 
 from pyrit import prompt_converter
@@ -42,13 +42,6 @@
 from pyrit.prompt_target import PromptTarget
 from pyrit.registry.object_registries import ConverterRegistry
 
-_DATA_TYPE_EXTENSION: dict[str, str] = {
-    "image_path": ".png",
-    "audio_path": ".wav",
-    "video_path": ".mp4",
-    "binary_path": ".bin",
-}
-
 
 def _build_converter_class_registry() -> dict[str, type]:
     """
@@ -221,6 +214,13 @@ class ConverterService:
     API metadata is derived from the converter objects.
     """
 
+    _DATA_TYPE_EXTENSION: ClassVar[dict[str, str]] = {
+        "image_path": ".png",
+        "audio_path": ".wav",
+        "video_path": ".mp4",
+        "binary_path": ".bin",
+    }
+
     def __init__(self) -> None:
         """Initialize the converter service."""
         self._registry = ConverterRegistry.get_registry_singleton()
@@ -375,7 +375,7 @@ async def preview_conversion_async(self, *, request: ConverterPreviewRequest) ->
             elif original_value.startswith("data:"):
                 _, _, value = original_value.partition(",")
 
-                ext = _DATA_TYPE_EXTENSION.get(str(data_type), ".bin")
+                ext = self._DATA_TYPE_EXTENSION.get(str(data_type), ".bin")
 
                 serializer = data_serializer_factory(
                     category="prompt-memory-entries",
@@ -389,7 +389,7 @@ async def preview_conversion_async(self, *, request: ConverterPreviewRequest) ->
                 pass
             else:
                 # Treat as raw base64
-                ext = _DATA_TYPE_EXTENSION.get(str(data_type), ".bin")
+                ext = self._DATA_TYPE_EXTENSION.get(str(data_type), ".bin")
 
                 serializer = data_serializer_factory(
                     category="prompt-memory-entries",

pyrit/backend/services/target_service.py

@@ -15,7 +15,7 @@
 import logging
 import os
 from functools import lru_cache
-from typing import Any
+from typing import Any, ClassVar
 from urllib.parse import urlparse
 
 from pyrit import prompt_target
@@ -35,9 +35,6 @@
 
 logger = logging.getLogger(__name__)
 
-# Scope for Azure Machine Learning managed online endpoints.
-_AZURE_ML_SCOPE = "https://ml.azure.com/.default"
-
 # Recognised Azure OpenAI / AI Foundry hostname suffixes. Used for strict
 # endpoint validation when Entra ID auth is requested, so a bearer token is
 # only ever issued for a known Microsoft-operated endpoint.
@@ -139,6 +136,9 @@ class TargetService:
     API metadata is derived from the target objects' identifiers.
     """
 
+    # Scope for Azure Machine Learning managed online endpoints.
+    _AZURE_ML_SCOPE: ClassVar[str] = "https://ml.azure.com/.default"
+
     def __init__(self) -> None:
         """Initialize the target service."""
         self._registry = TargetRegistry.get_registry_singleton()
@@ -406,7 +406,7 @@ def _apply_entra_auth(*, target_class: type, target_type: str, params: dict[str,
                     "Entra ID authentication for AzureMLChatTarget requires an AML endpoint "
                     f"(*.inference.ml.azure.com). Got: {endpoint}"
                 )
-            new_params["api_key"] = get_azure_async_token_provider(_AZURE_ML_SCOPE)
+            new_params["api_key"] = get_azure_async_token_provider(TargetService._AZURE_ML_SCOPE)
             return new_params
 
         raise ValueError(

pyrit/registry/tag_query.py

@@ -29,7 +29,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Protocol, TypeVar, runtime_checkable
+from typing import TYPE_CHECKING, ClassVar, Protocol, TypeVar, runtime_checkable
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -46,9 +46,6 @@ def tags(self) -> list[str]:  # noqa: D102
 
 _T = TypeVar("_T", bound=Taggable)
 
-_VALID_OPS = frozenset({"", "and", "or"})
-_OP_FUNC: dict[str, Callable[..., bool]] = {"and": all, "or": any}
-
 
 @dataclass(frozen=True)
 class TagQuery:
@@ -68,6 +65,9 @@ class TagQuery:
         exclude_tags: Tags that must **not** be present.
     """
 
+    _VALID_OPS: ClassVar[frozenset[str]] = frozenset({"", "and", "or"})
+    _OP_FUNC: ClassVar[dict[str, Callable[..., bool]]] = {"and": all, "or": any}
+
     include_all: frozenset[str] = frozenset()
     include_any: frozenset[str] = frozenset()
     exclude_tags: frozenset[str] = frozenset()
@@ -88,8 +88,8 @@ def __post_init__(self) -> None:
             if not isinstance(val, frozenset):
                 object.__setattr__(self, attr, frozenset(val))
 
-        if self._op not in _VALID_OPS:
-            raise ValueError(f"Invalid TagQuery op {self._op!r}; must be one of {sorted(_VALID_OPS)}")
+        if self._op not in self._VALID_OPS:
+            raise ValueError(f"Invalid TagQuery op {self._op!r}; must be one of {sorted(self._VALID_OPS)}")
         if self._op in ("and", "or") and len(self._children) < 2:
             raise ValueError(f"'{self._op}' TagQuery must have at least 2 children")
         if self._op == "" and self._children:
@@ -166,7 +166,7 @@ def matches(self, tags: set[str] | frozenset[str]) -> bool:
             Whether the tag set matches.
         """
         if self._op:
-            return _OP_FUNC[self._op](c.matches(tags) for c in self._children)
+            return self._OP_FUNC[self._op](c.matches(tags) for c in self._children)
         return self._matches_leaf(tags)
 
     def _matches_leaf(self, tags: set[str] | frozenset[str]) -> bool:

pyrit/setup/configuration_loader.py

@@ -11,7 +11,7 @@
 import pathlib
 from collections.abc import Sequence
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from pyrit.common.path import DEFAULT_CONFIG_PATH
 from pyrit.common.yaml_loadable import YamlLoadable
@@ -32,13 +32,6 @@
 YamlPrimitive = str | int | float | bool | None
 YamlValue = YamlPrimitive | list["YamlValue"] | dict[str, "YamlValue"]
 
-# Mapping from snake_case config values to internal constants
-_MEMORY_DB_TYPE_MAP: dict[str, str] = {
-    "in_memory": IN_MEMORY,
-    "sqlite": SQLITE,
-    "azure_sql": AZURE_SQL,
-}
-
 
 @dataclass
 class InitializerConfig:
@@ -136,6 +129,13 @@ class ConfigurationLoader(YamlLoadable):
         operation: my_operation
     """
 
+    # Mapping from snake_case config values to internal constants
+    _MEMORY_DB_TYPE_MAP: ClassVar[dict[str, str]] = {
+        "in_memory": IN_MEMORY,
+        "sqlite": SQLITE,
+        "azure_sql": AZURE_SQL,
+    }
+
     memory_db_type: str = "sqlite"
     initializers: list[str | dict[str, Any]] = field(default_factory=list)
     initialization_scripts: list[str] | None = None
@@ -171,12 +171,12 @@ def _normalize_memory_db_type(self) -> None:
         normalized = self.memory_db_type.lower().replace("-", "_")
 
         # Also handle PascalCase inputs (e.g., "InMemory" -> "in_memory")
-        if normalized not in _MEMORY_DB_TYPE_MAP:
+        if normalized not in self._MEMORY_DB_TYPE_MAP:
             # Try converting from PascalCase
             normalized = class_name_to_snake_case(self.memory_db_type)
 
-        if normalized not in _MEMORY_DB_TYPE_MAP:
-            valid_types = list(_MEMORY_DB_TYPE_MAP.keys())
+        if normalized not in self._MEMORY_DB_TYPE_MAP:
+            valid_types = list(self._MEMORY_DB_TYPE_MAP.keys())
             raise ValueError(
                 f"Invalid memory_db_type '{self.memory_db_type}'. Must be one of: {', '.join(valid_types)}"
             )
@@ -552,7 +552,7 @@ async def initialize_pyrit_async(self) -> None:
         resolved_env_files = self.resolve_env_files()
 
         # Map snake_case memory_db_type to internal constant
-        internal_memory_db_type = _MEMORY_DB_TYPE_MAP[self.memory_db_type]
+        internal_memory_db_type = self._MEMORY_DB_TYPE_MAP[self.memory_db_type]
 
         await initialize_pyrit_async(
             memory_db_type=internal_memory_db_type,