MAINT: enable strict mypy checking and fix violations

pyproject.toml

@@ -167,12 +167,15 @@ asyncio_mode = "auto"
 [tool.mypy]
 plugins = []
 ignore_missing_imports = true
-strict = false
+strict = true
 follow_imports = "silent"
-strict_optional = false
 disable_error_code = ["empty-body"]
 exclude = ["doc/code/", "pyrit/auxiliary_attacks/"]
 
+[[tool.mypy.overrides]]
+module = "pyrit.prompt_target.hugging_face.*"
+disallow_untyped_calls = false
+
 [tool.uv]
 constraint-dependencies = [
     "aiohttp>=3.13.4",

pyrit/analytics/result_analysis.py

@@ -62,9 +62,8 @@ def analyze_results(attack_results: list[AttackResult]) -> dict[str, AttackStats
             raise TypeError(f"Expected AttackResult, got {type(attack).__name__}: {attack!r}")
 
         outcome = attack.outcome
-        attack_type = (
-            attack.get_attack_strategy_identifier().class_name if attack.get_attack_strategy_identifier() else "unknown"
-        )
+        _strategy_id = attack.get_attack_strategy_identifier()
+        attack_type = _strategy_id.class_name if _strategy_id is not None else "unknown"
 
         if outcome == AttackOutcome.SUCCESS:
             overall_counts["successes"] += 1

pyrit/auth/azure_auth.py

@@ -296,7 +296,7 @@ def get_access_token_from_interactive_login(scope: str) -> str:
     """
     try:
         token_provider = get_bearer_token_provider(InteractiveBrowserCredential(), scope)
-        return token_provider()
+        return str(token_provider())
     except Exception as e:
         logger.error(f"Failed to obtain token for '{scope}': {e}")
         raise

pyrit/auth/copilot_authenticator.py

@@ -353,6 +353,9 @@ async def _run_playwright_browser_automation(self) -> Optional[str]:
 
         Returns:
             Optional[str]: The bearer token if successfully retrieved, None otherwise.
+
+        Raises:
+            ValueError: If the username is not set.
         """
         from playwright.async_api import async_playwright
 
@@ -415,11 +418,15 @@ async def response_handler(response: Any) -> None:
 
                 logger.info("Waiting for email input...")
                 await page.wait_for_selector("#i0116", timeout=self._elements_timeout)
+                if self._username is None:
+                    raise ValueError("Username is not set")
                 await page.fill("#i0116", self._username)
                 await page.click("#idSIButton9")
 
                 logger.info("Waiting for password input...")
                 await page.wait_for_selector("#i0118", timeout=self._elements_timeout)
+                if self._password is None:
+                    raise ValueError("Password is not set")
                 await page.fill("#i0118", self._password)
                 await page.click("#idSIButton9")
 
@@ -450,7 +457,7 @@ async def response_handler(response: Any) -> None:
                 else:
                     logger.error(f"Failed to retrieve bearer token within {self._token_capture_timeout} seconds.")
 
-                return bearer_token  # type: ignore[no-any-return]
+                return bearer_token
             except Exception as e:
                 logger.error("Failed to retrieve access token using Playwright.")
 

pyrit/backend/middleware/auth.py

@@ -61,6 +61,7 @@ def __init__(self, app: ASGIApp) -> None:
         self._allowed_group_ids: set[str] = {g.strip() for g in groups_raw.split(",") if g.strip()}
         self._enabled = bool(self._tenant_id and self._client_id)
 
+        self._jwks_client: PyJWKClient | None
         if self._enabled:
             jwks_url = f"https://login.microsoftonline.com/{self._tenant_id}/discovery/v2.0/keys"
             self._jwks_client = PyJWKClient(jwks_url, cache_keys=True)
@@ -251,6 +252,8 @@ def _validate_token(self, token: str) -> tuple[Optional[AuthenticatedUser], dict
             Tuple of (AuthenticatedUser, claims) if valid, (None, {}) if validation fails.
         """
         try:
+            if self._jwks_client is None:
+                raise RuntimeError("JWKS client not initialized")
             signing_key = self._jwks_client.get_signing_key_from_jwt(token)
             claims = jwt.decode(
                 token,

pyrit/backend/routes/media.py

@@ -123,6 +123,8 @@ async def serve_media_async(
     """
     try:
         memory = CentralMemory.get_memory_instance()
+        if not memory.results_path:
+            raise HTTPException(status_code=500, detail="Memory results_path is not configured.")
         allowed_root = os.path.realpath(memory.results_path)
     except Exception as exc:
         raise HTTPException(status_code=500, detail="Memory not initialized; cannot determine results path.") from exc

pyrit/backend/routes/version.py

@@ -67,7 +67,7 @@ async def get_version_async(request: Request) -> VersionResponse:
         memory = CentralMemory.get_memory_instance()
         db_type = type(memory).__name__
         db_name = None
-        if memory.engine.url.database:
+        if memory.engine is not None and memory.engine.url.database:
             db_name = memory.engine.url.database.split("?")[0]
         database_info = f"{db_type} ({db_name})" if db_name else f"{db_type} (None)"
     except Exception as e:

pyrit/cli/_banner.py

@@ -568,11 +568,11 @@ def _render_line_with_segments(
     result: list[str] = []
     current_role: Optional[ColorRole] = None
     for pos, ch in enumerate(line):
-        role = char_roles[pos]
-        if role != current_role:
-            color = _get_color(role, theme) if role else reset
+        char_role = char_roles[pos]
+        if char_role != current_role:
+            color = _get_color(char_role, theme) if char_role else reset
             result.append(color)
-            current_role = role
+            current_role = char_role
         result.append(ch)
     result.append(reset)
     return "".join(result)

pyrit/cli/_cli_args.py

@@ -482,29 +482,29 @@ def _parse_shell_arguments(*, parts: list[str], arg_specs: list[_ArgSpec]) -> di
     i = 0
     while i < len(parts):
         token = parts[i]
-        spec = flag_to_spec.get(token)
+        matched_spec: _ArgSpec | None = flag_to_spec.get(token)
 
-        if spec is None:
+        if matched_spec is None:
             valid = sorted(flag_to_spec.keys())
             raise ValueError(f"Unknown argument: {token}. Valid arguments: {', '.join(valid)}")
 
         i += 1
 
-        if spec.multi_value:
+        if matched_spec.multi_value:
             values: list[Any] = []
             # Collect values until the next flag (whether valid or invalid)
             while i < len(parts) and not (parts[i].startswith("--") or parts[i] in flag_to_spec):
-                item = spec.parser(parts[i]) if spec.parser else parts[i]
+                item = matched_spec.parser(parts[i]) if matched_spec.parser else parts[i]
                 values.append(item)
                 i += 1
             if len(values) == 0:
-                raise ValueError(f"{spec.flags[0]} requires at least one value")
-            result[spec.result_key] = values
+                raise ValueError(f"{matched_spec.flags[0]} requires at least one value")
+            result[matched_spec.result_key] = values
         else:
             if i >= len(parts):
-                raise ValueError(f"{spec.flags[0]} requires a value")
+                raise ValueError(f"{matched_spec.flags[0]} requires a value")
             raw = parts[i]
-            result[spec.result_key] = spec.parser(raw) if spec.parser else raw
+            result[matched_spec.result_key] = matched_spec.parser(raw) if matched_spec.parser else raw
             i += 1
 
     return result

pyrit/cli/frontend_core.py

@@ -56,7 +56,7 @@ class termcolor:  # type: ignore[no-redef]  # noqa: N801
         """Dummy termcolor fallback for colored printing if termcolor is not installed."""
 
         @staticmethod
-        def cprint(text: str, color: str = None, attrs: list = None) -> None:  # type: ignore[type-arg]
+        def cprint(text: str, color: str | None = None, attrs: list[Any] | None = None) -> None:
             """Print text without color."""
             print(text)
 
@@ -249,12 +249,14 @@ def scenario_registry(self) -> ScenarioRegistry:
 
         Raises:
             RuntimeError: If initialize_async() has not been called.
+            ValueError: If the scenario registry is not initialized.
         """
         if not self._initialized:
             raise RuntimeError(
                 "FrontendCore not initialized. Call 'await context.initialize_async()' before accessing registries."
             )
-        assert self._scenario_registry is not None
+        if self._scenario_registry is None:
+            raise ValueError("self._scenario_registry is not initialized")
         return self._scenario_registry
 
     @property
@@ -264,12 +266,14 @@ def initializer_registry(self) -> InitializerRegistry:
 
         Raises:
             RuntimeError: If initialize_async() has not been called.
+            ValueError: If the initializer registry is not initialized.
         """
         if not self._initialized:
             raise RuntimeError(
                 "FrontendCore not initialized. Call 'await context.initialize_async()' before accessing registries."
             )
-        assert self._initializer_registry is not None
+        if self._initializer_registry is None:
+            raise ValueError("self._initializer_registry is not initialized")
         return self._initializer_registry
 
 

pyrit/cli/pyrit_shell.py

@@ -19,6 +19,8 @@
 from typing import TYPE_CHECKING, Any, Optional
 
 if TYPE_CHECKING:
+    import types
+
     from pyrit.cli import frontend_core
     from pyrit.models.scenario_result import ScenarioResult
 
@@ -119,16 +121,17 @@ def __init__(
                 new_item="PyRITShell(database=..., log_level=..., ...)",
                 removed_in="0.14.0",
             )
-            self._deprecated_context = context
+            self._deprecated_context: frontend_core.FrontendCore | None = context
         else:
             self._deprecated_context = None
 
         # Track scenario execution history: list of (command_string, ScenarioResult) tuples
         self._scenario_history: list[tuple[str, ScenarioResult]] = []
 
         # Set by the background thread after importing frontend_core.
-        self.context: Optional[frontend_core.FrontendCore] = None
-        self.default_log_level: Optional[int] = None
+        self._fc: types.ModuleType | None = None
+        self.context: frontend_core.FrontendCore | None = None
+        self.default_log_level: int | None = None
 
         # Initialize PyRIT in background thread for faster startup.
         self._init_thread = threading.Thread(target=self._background_init, daemon=True)
@@ -159,12 +162,19 @@ def _raise_init_error(self) -> None:
             raise self._init_error
 
     def _ensure_initialized(self) -> None:
-        """Wait for initialization to complete if not already done."""
+        """
+        Wait for initialization to complete if not already done.
+
+        Raises:
+            RuntimeError: If frontend core initialization failed or is not complete.
+        """
         if not self._init_complete.is_set():
             print("Waiting for PyRIT initialization to complete...")
             sys.stdout.flush()
             self._init_complete.wait()
         self._raise_init_error()
+        if self._fc is None or self.context is None:
+            raise RuntimeError("Frontend core not initialized")
 
     def cmdloop(self, intro: Optional[str] = None) -> None:
         """Override cmdloop to play animated banner before starting the REPL."""
@@ -188,22 +198,36 @@ def cmdloop(self, intro: Optional[str] = None) -> None:
         super().cmdloop(intro=self.intro)
 
     def do_list_scenarios(self, arg: str) -> None:
-        """List all available scenarios."""
+        """
+        List all available scenarios.
+
+        Raises:
+            RuntimeError: If initialization has not completed.
+        """
         if arg.strip():
             print(f"Error: list-scenarios does not accept arguments, got: {arg.strip()}")
             return
         self._ensure_initialized()
+        if self._fc is None or self.context is None:
+            raise RuntimeError("Frontend core not initialized")
         try:
             asyncio.run(self._fc.print_scenarios_list_async(context=self.context))
         except Exception as e:
             print(f"Error listing scenarios: {e}")
 
     def do_list_initializers(self, arg: str) -> None:
-        """List all available initializers."""
+        """
+        List all available initializers.
+
+        Raises:
+            RuntimeError: If initialization has not completed.
+        """
         if arg.strip():
             print(f"Error: list-initializers does not accept arguments, got: {arg.strip()}")
             return
         self._ensure_initialized()
+        if self._fc is None or self.context is None:
+            raise RuntimeError("Frontend core not initialized")
         try:
             asyncio.run(self._fc.print_initializers_list_async(context=self.context))
         except Exception as e:
@@ -225,8 +249,13 @@ def do_list_targets(self, arg: str) -> None:
         Examples:
             list-targets --initializers target
             list-targets --initializers target:tags=default,scorer
+
+        Raises:
+            RuntimeError: If initialization has not completed.
         """
         self._ensure_initialized()
+        if self._fc is None or self.context is None:
+            raise RuntimeError("Frontend core not initialized")
         try:
             list_targets_context = self.context
             if arg.strip():
@@ -289,8 +318,13 @@ def do_run(self, line: str) -> None:
             --target is required for every run.
             Initializers can be specified per-run or configured in .pyrit_conf.
             Database and env-files are configured via the config file.
+
+        Raises:
+            RuntimeError: If initialization has not completed.
         """
         self._ensure_initialized()
+        if self._fc is None or self.context is None:
+            raise RuntimeError("Frontend core not initialized")
 
         if not line.strip():
             print("Error: Specify a scenario name")

pyrit/common/data_url_converter.py

@@ -23,7 +23,7 @@ async def convert_local_image_to_data_url(image_path: str) -> str:
         str: A string containing the MIME type and the base64-encoded data of the image, formatted as a data URL.
     """
     ext = DataTypeSerializer.get_extension(image_path)
-    if ext.lower() not in AZURE_OPENAI_GPT4O_SUPPORTED_IMAGE_FORMATS:
+    if not ext or ext.lower() not in AZURE_OPENAI_GPT4O_SUPPORTED_IMAGE_FORMATS:
         raise ValueError(
             f"Unsupported image format: {ext}. Supported formats are: {AZURE_OPENAI_GPT4O_SUPPORTED_IMAGE_FORMATS}"
         )