[BREAKING] FIX: Require administrator access to create targets that run model repository code

doc/gui/0_gui.md

@@ -16,8 +16,9 @@ pyrit_backend
 
 Then open `http://localhost:8000` in your browser.
 
-Authentication-disabled local servers deny administrator operations by default. To enable configuration and initializer
-administration for a trusted local development server, set `PYRIT_ALLOW_UNAUTHENTICATED_ADMIN=true`. Never use this
+Authentication-disabled local servers deny administrator operations by default. To enable administrator operations
+(configuration, initializer administration, and creating targets that run model repository code) for a trusted local
+development server, set `PYRIT_ALLOW_UNAUTHENTICATED_ADMIN=true`. Never use this
 setting on a network-accessible deployment.
 
 ### Docker

pyrit/backend/middleware/auth.py

@@ -58,16 +58,27 @@ def authorization_environment(request: Request) -> Mapping[str, str]:
     return cast("Mapping[str, str]", environment) if isinstance(environment, dict) else os.environ
 
 
-def require_admin(request: Request) -> None:
-    """Require an administrator when authentication is enabled."""
+def has_admin_access(request: Request) -> bool:
+    """
+    Return whether the caller is an administrator.
+
+    Without authentication, only a server that sets ``PYRIT_ALLOW_UNAUTHENTICATED_ADMIN=true``
+    grants administrator access.
+
+    Returns:
+        bool: Whether the caller has administrator access.
+    """
     user = getattr(request.state, "user", None)
     if user is None:
-        allow_unauthenticated = (
+        return (
             authorization_environment(request).get("PYRIT_ALLOW_UNAUTHENTICATED_ADMIN", "").strip().casefold() == "true"
         )
-        if allow_unauthenticated:
-            return
-    if not isinstance(user, AuthenticatedUser) or not user.is_admin:
+    return isinstance(user, AuthenticatedUser) and user.is_admin
+
+
+def require_admin(request: Request) -> None:
+    """Require an administrator when authentication is enabled."""
+    if not has_admin_access(request):
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail="Administrator access is required",

pyrit/backend/routes/targets.py

@@ -8,8 +8,9 @@
 Target types are set at app startup via initializers - you cannot add new types at runtime.
 """
 
-from fastapi import APIRouter, HTTPException, Query, status
+from fastapi import APIRouter, HTTPException, Query, Request, status
 
+from pyrit.backend.middleware.auth import has_admin_access
 from pyrit.backend.models.common import CursorStr, IdentifierStr, ProblemDetail
 from pyrit.backend.models.targets import (
     CreateTargetRequest,
@@ -72,10 +73,15 @@ async def list_target_types() -> TargetTypeResponse:  # pyrit-async-suffix-exemp
             "model": ProblemDetail,
             "description": "Invalid target type or parameters",
         },
+        403: {
+            "model": ProblemDetail,
+            "description": "Administrator access is required to run model repository code",
+        },
     },
 )
 async def create_target(
-    request: CreateTargetRequest,
+    body: CreateTargetRequest,
+    request: Request,
 ) -> TargetInstance:  # pyrit-async-suffix-exempt
     """
     Create a new target instance.
@@ -91,7 +97,12 @@ async def create_target(
     service = get_target_service()
 
     try:
-        return await service.create_target_async(request=request)
+        return await service.create_target_async(request=body, is_admin=has_admin_access(request))
+    except PermissionError as e:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=str(e),
+        ) from e
     except ValueError as e:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,

pyrit/backend/services/target_service.py

@@ -220,7 +220,7 @@ async def list_target_types_async(self) -> TargetTypeResponse:
         ]
         return TargetTypeResponse(items=items)
 
-    async def create_target_async(self, *, request: CreateTargetRequest) -> TargetInstance:
+    async def create_target_async(self, *, request: CreateTargetRequest, is_admin: bool = False) -> TargetInstance:
         """
         Create a new target instance from API request.
 
@@ -232,12 +232,14 @@ async def create_target_async(self, *, request: CreateTargetRequest) -> TargetIn
         supports it and omits the api_key plus any registry-flagged
         identity-conflicting parameters so the target validates its own
         endpoint and authenticates itself. GitHub Copilot requires an explicit
-        or environment token rather than SDK login discovery in CoPyRIT.
-        The response is built before the
-        target is registered, so a failed request leaves no registered target.
+        or environment token rather than SDK login discovery in CoPyRIT. A target
+        whose construction runs model repository code is built only for an
+        administrator. The response is built before the target is registered, so
+        a failed request leaves no registered target.
 
         Args:
             request: The create target request with type, params, and auth_mode.
+            is_admin: Whether the caller may create a target that runs model repository code.
 
         Returns:
             TargetInstance with the new target's details.
@@ -247,6 +249,7 @@ async def create_target_async(self, *, request: CreateTargetRequest) -> TargetIn
                 requested but unsupported by the target type. Construction errors
                 (unknown params, incompatible inner targets, unrecognized identity
                 endpoints) are raised by the registry / target classes.
+            PermissionError: If a non-administrator enables a target's model code execution.
         """
         if request.type not in self._registry:
             raise ValueError(
@@ -283,6 +286,8 @@ async def create_target_async(self, *, request: CreateTargetRequest) -> TargetIn
                     "GITHUB_TOKEN for this CoPyRIT instance through its environment or loaded .env/.env.local files."
                 )
 
+        await self._enforce_code_execution_permission_async(target_type=request.type, params=params, is_admin=is_admin)
+
         # LEGACY COMPATIBILITY: The current configuration UI omits the name.
         # Remove this generated fallback after that UI sends an explicit name.
         target_registry_name = request.name or f"compat_{uuid.uuid4().hex}"
@@ -292,6 +297,50 @@ async def create_target_async(self, *, request: CreateTargetRequest) -> TargetIn
         self._registry.instances.register(target_obj, name=target_registry_name)
         return target
 
+    async def _enforce_code_execution_permission_async(
+        self, *, target_type: str, params: dict[str, Any], is_admin: bool
+    ) -> None:
+        """
+        Require administrator access when a target's model-code-execution parameter resolves true.
+
+        A target names the boolean parameter that runs model repository code through
+        ``code_execution_parameter``. The supplied value is coerced through the registry (so the
+        textual boolean forms the catalog accepts are honored) and written back, so the permission
+        check and the constructor agree and a value the registry cannot interpret as a boolean is
+        rejected before construction rather than reaching the loader as a truthy non-boolean.
+
+        Args:
+            target_type (str): Registered target class name.
+            params (dict[str, Any]): Constructor parameters, updated in place with the coerced value.
+            is_admin (bool): Whether the caller may enable model code execution.
+
+        Raises:
+            ValueError: If the parameter value cannot be coerced to a boolean, or the target names a
+                model-code-execution parameter that the registry does not expose (so the check would
+                otherwise be skipped silently).
+            PermissionError: If the parameter resolves true and the caller is not an administrator.
+        """
+        parameter_name = self._registry.get_class(target_type).code_execution_parameter
+        if parameter_name is None:
+            return
+        metadata = await asyncio.to_thread(self._registry.get_registered_class_metadata, target_type)
+        parameter = next((p for p in (metadata.parameters if metadata else ()) if p.name == parameter_name), None)
+        if parameter is None:
+            raise ValueError(
+                f"Target type '{target_type}' names '{parameter_name}' as its model-code-execution parameter, "
+                "but the registry does not expose it, so the administrator requirement cannot be enforced."
+            )
+        if parameter_name in params:
+            params[parameter_name] = parameter.coerce_value(params[parameter_name])
+            enabled = bool(params[parameter_name])
+        else:
+            enabled = bool(parameter.default)
+        if enabled and not is_admin:
+            raise PermissionError(
+                f"Creating '{target_type}' with '{parameter_name}' enabled runs model repository code and "
+                "requires administrator access."
+            )
+
 
 @lru_cache(maxsize=1)
 def get_target_service() -> TargetService:

pyrit/prompt_target/common/prompt_target.py

@@ -84,6 +84,12 @@ class PromptTarget(Identifiable):
     # Azure Blob Storage, Prompt Shield) override this to add ``"identity"``.
     supported_auth_modes: ClassVar[tuple[AuthMode, ...]] = ("api_key",)
 
+    # Targets whose construction can run model repository code name the boolean constructor
+    # parameter that enables it (e.g. ``"trust_remote_code"``). The create-target API builds
+    # them only for an administrator when that parameter resolves true; direct Python use is
+    # unaffected.
+    code_execution_parameter: ClassVar[str | None] = None
+
     def __init_subclass__(cls, **kwargs: object) -> None:
         """
         Validate that subclasses follow the keyword-only ``__init__`` contract.

pyrit/prompt_target/hugging_face/hugging_face_chat_target.py

@@ -6,7 +6,7 @@
 import logging
 import warnings
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, ClassVar, cast
 
 from pyrit.common import default_values
 from pyrit.exceptions import EmptyResponseException, pyrit_target_retry
@@ -36,6 +36,8 @@ class HuggingFaceChatTarget(PromptTarget):
         )
     )
 
+    code_execution_parameter: ClassVar[str | None] = "trust_remote_code"
+
     # Class-level cache for model and tokenizer
     _cached_model: Any = None
     _cached_tokenizer: Any = None

tests/unit/backend/test_api_routes.py

@@ -1260,6 +1260,45 @@ def test_create_target_internal_error(self, client: TestClient) -> None:
 
             assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
 
+    def test_create_target_forbids_code_execution_for_non_admin(self, client: TestClient) -> None:
+        """A PermissionError from the service maps to 403, not 400 or 500."""
+        with patch("pyrit.backend.routes.targets.get_target_service") as mock_get_service:
+            mock_service = MagicMock()
+            mock_service.create_target_async = AsyncMock(
+                side_effect=PermissionError("Creating 'HuggingFaceChatTarget' ... requires administrator access.")
+            )
+            mock_get_service.return_value = mock_service
+
+            response = client.post(
+                "/api/targets",
+                json={"type": "HuggingFaceChatTarget", "params": {"trust_remote_code": True}},
+            )
+
+            assert response.status_code == status.HTTP_403_FORBIDDEN
+            assert "administrator access" in response.json()["detail"]
+
+    def test_create_target_passes_admin_flag_from_request(self, client: TestClient) -> None:
+        """The route resolves administrator access and forwards it to the service."""
+        with (
+            patch("pyrit.backend.routes.targets.get_target_service") as mock_get_service,
+            patch("pyrit.backend.routes.targets.has_admin_access", return_value=True) as mock_has_admin,
+        ):
+            mock_service = MagicMock()
+            mock_service.create_target_async = AsyncMock(
+                return_value=TargetInstance(
+                    target_registry_name="target-1",
+                    identifier=TargetIdentifier(class_name="TextTarget"),
+                    capabilities=TargetCapabilities(),
+                )
+            )
+            mock_get_service.return_value = mock_service
+
+            response = client.post("/api/targets", json={"type": "TextTarget", "params": {}})
+
+            assert response.status_code == status.HTTP_201_CREATED
+            mock_has_admin.assert_called_once()
+            assert mock_service.create_target_async.call_args.kwargs["is_admin"] is True
+
     def test_get_target_success(self, client: TestClient) -> None:
         """Test getting a target by ID."""
         with patch("pyrit.backend.routes.targets.get_target_service") as mock_get_service: