test(integration): cover guardrails_policies endpoint contracts (#44722)

* test(integration): cover guardrails_policies endpoint contracts

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): let the admin submissions list carry other rows

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): close guardrails policies review gaps

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): cover guardrail auth and policy attachments

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): tidy guardrail policy test helpers

* test(integration): keep plain non-admin submission refusal live

* test: expand policy attachment integration coverage

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): accept the intended 403 for plain submission keys

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): pin multi-tag header and spend row tags for policy tag scope

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): drop the duplicate pydantic import left by the main merge

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: mateo <mateo@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Devin <noreply@devin.ai>

tests/integration/authorization/test_guardrail_submissions.py

@@ -0,0 +1,521 @@
+import json
+import uuid
+from contextlib import ExitStack
+from datetime import datetime
+from hashlib import sha256
+from typing import Final
+
+import httpx
+import pytest
+from integration._support.client import Gateway, object_value
+from integration._support.database import read_rows
+from integration._support.wire import Reply, Request, wire_server
+from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
+
+_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
+_PROMPT_PENDING: Final = "pending submission must not run"
+_PROMPT_APPROVED: Final = "approved submission runs"
+_PROMPT_REJECTED: Final = "rejected submission must not run"
+_APPLY_TEXT: Final = "apply guardrail after approval"
+_PROVIDER_RESPONSE: Final = {
+    "id": "chatcmpl-guardrail-submission",
+    "object": "chat.completion",
+    "created": 1,
+    "model": "gpt-4o-mini",
+    "choices": [
+        {"index": 0, "message": {"role": "assistant", "content": "provider response"}, "finish_reason": "stop"}
+    ],
+    "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+}
+
+
+class _RegisterResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    guardrail_id: str
+    guardrail_name: str
+    status: str
+    submitted_at: datetime
+
+
+class _GuardrailRow(BaseModel):
+    guardrail_id: str
+    status: str
+    team_id: str
+
+
+class _ProxyError(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    message: str
+    type: str
+    param: str | None
+    code: str
+
+
+class _ErrorResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    error: _ProxyError
+
+
+class _SubmissionForbiddenResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    detail: str
+
+
+class _AdminActionResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    guardrail_id: str
+    status: str
+    message: str
+
+
+class _ApplyResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    response_text: str
+
+
+class _ChatMessage(BaseModel):
+    role: str
+    content: str | None
+
+
+class _ChatChoice(BaseModel):
+    index: int
+    message: _ChatMessage
+    finish_reason: str | None
+
+
+class _ChatResponse(BaseModel):
+    choices: list[_ChatChoice]
+
+
+class _SubmissionItem(BaseModel):
+    guardrail_id: str
+
+
+class _SubmissionsResponse(BaseModel):
+    submissions: list[_SubmissionItem]
+    summary: dict[str, int]
+
+
+def _delete_guardrail(gateway: Gateway, guardrail_id: str) -> None:
+    deleted: Final = gateway.request("DELETE", f"/guardrails/{guardrail_id}")
+    assert deleted.status_code == 200, deleted.text
+    assert read_rows('SELECT guardrail_id FROM "LiteLLM_GuardrailsTable" WHERE guardrail_id=%s', (guardrail_id,)) == []
+
+
+def _assert_submission_status(guardrail_id: str, status: str, team_id: str) -> None:
+    rows: Final = read_rows(
+        'SELECT guardrail_id, status, team_id FROM "LiteLLM_GuardrailsTable" WHERE guardrail_id=%s',
+        (guardrail_id,),
+    )
+    assert len(rows) == 1, rows
+    assert _GuardrailRow.model_validate(rows[0]) == _GuardrailRow(
+        guardrail_id=guardrail_id, status=status, team_id=team_id
+    ), rows
+
+
+def _assert_submission_actions_are_forbidden(responses: tuple[httpx.Response, ...]) -> None:
+    assert tuple(response.status_code for response in responses) == (403, 403, 403, 403), tuple(
+        response.text for response in responses
+    )
+    expected_error: Final = {"detail": "Admin access required"}
+    assert tuple(
+        _SubmissionForbiddenResponse.model_validate_json(response.content).model_dump() for response in responses
+    ) == (expected_error,) * 4, tuple(response.text for response in responses)
+
+
+def _mask_user_id(user_id: str) -> str:
+    return f"{user_id[:6]}{'*' * (len(user_id) - 8)}{user_id[-2:]}"
+
+
+def _register(
+    gateway: Gateway,
+    scenario_cleanups: ExitStack,
+    name: str,
+    team_id: str | None,
+    key: str,
+    api_base: str,
+) -> str:
+    params: Final[dict[str, JsonValue]] = {
+        "guardrail": "generic_guardrail_api",
+        "mode": "pre_call",
+        "default_on": False,
+        "api_base": api_base,
+        "api_key": "synthetic-submission-guardrail-key",
+    }
+    request_body: Final[dict[str, JsonValue]] = {"guardrail_name": name, "litellm_params": params}
+    complete_body: Final[dict[str, JsonValue]] = {
+        **request_body,
+        **({"team_id": team_id} if team_id is not None else {}),
+    }
+    response: Final = gateway.request("POST", "/guardrails/register", complete_body, key=key)
+    assert response.status_code == 200, response.text
+    created: Final = _RegisterResponse.model_validate_json(response.content)
+    assert (created.guardrail_name, created.status) == (name, "pending_review"), response.text
+    scenario_cleanups.callback(_delete_guardrail, gateway, created.guardrail_id)
+    return created.guardrail_id
+
+
+def _chat(gateway: Gateway, model: str, key: str, text: str, guardrail: str) -> _ChatResponse:
+    response: Final = gateway.request(
+        "POST",
+        "/v1/chat/completions",
+        {"model": model, "messages": [{"role": "user", "content": text}], "guardrails": [guardrail]},
+        key=key,
+    )
+    assert response.status_code == 200, response.text
+    parsed: Final = _ChatResponse.model_validate_json(response.content)
+    assert parsed.choices[0].message.content == "provider response", response.text
+    return parsed
+
+
+def _assert_provider_request(request: Request, text: str) -> None:
+    assert (request.method, request.target) == ("POST", "/v1/chat/completions")
+    payload: Final = _JSON_OBJECT.validate_json(request.body)
+    assert payload == {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": text}]}, payload
+
+
+def _assert_guardrail_request(
+    request: Request,
+    text: str,
+    model: str | None,
+    key: str,
+    *,
+    key_alias: str | None = None,
+) -> None:
+    assert (request.method, request.target) == ("POST", "/beta/litellm_basic_guardrail_api")
+    payload: Final = _JSON_OBJECT.validate_json(request.body)
+    assert set(payload) == {
+        "input_type",
+        "litellm_call_id",
+        "litellm_trace_id",
+        "structured_messages",
+        "images",
+        "tools",
+        "texts",
+        "request_data",
+        "request_headers",
+        "litellm_version",
+        "additional_provider_specific_params",
+        "tool_calls",
+        "model",
+    }, payload
+    assert {
+        field: payload[field]
+        for field in (
+            "input_type",
+            "structured_messages",
+            "images",
+            "tools",
+            "texts",
+            "additional_provider_specific_params",
+            "tool_calls",
+            "model",
+        )
+    } == {
+        "input_type": "request",
+        "structured_messages": [{"role": "user", "content": text}] if model is not None else None,
+        "images": None,
+        "tools": None,
+        "texts": [text],
+        "additional_provider_specific_params": {},
+        "tool_calls": None,
+        "model": model,
+    }, payload
+    assert payload["litellm_call_id"] is None or isinstance(payload["litellm_call_id"], str), payload
+    assert payload["litellm_trace_id"] is None or isinstance(payload["litellm_trace_id"], str), payload
+    assert payload["litellm_version"] is None or isinstance(payload["litellm_version"], str), payload
+    request_data: Final = object_value(payload["request_data"])
+    if model is None:
+        assert request_data == {}, payload
+    else:
+        assert set(request_data).issubset(
+            {
+                "user_api_key_hash",
+                "user_api_key_alias",
+                "user_api_key_user_id",
+                "user_api_key_user_email",
+                "user_api_key_team_id",
+                "user_api_key_team_alias",
+                "user_api_key_end_user_id",
+                "user_api_key_org_id",
+            }
+        ), request_data
+        assert request_data["user_api_key_hash"] == sha256(key.encode()).hexdigest(), request_data
+        assert request_data.get("user_api_key_alias") == key_alias, request_data
+        assert all(value is None or isinstance(value, str) for value in request_data.values()), request_data
+    request_headers: Final = payload["request_headers"]
+    assert request_headers is None or isinstance(request_headers, dict), payload
+    if isinstance(request_headers, dict):
+        assert "authorization" not in request_headers
+    assert request.headers["x-api-key"] == "synthetic-submission-guardrail-key"
+
+
+def test_non_admin_keys_scoped_to_the_submission_routes_get_403_from_the_admin_check(gateway: Gateway) -> None:
+    with (
+        wire_server(lambda _request: Reply(body=json.dumps(_PROVIDER_RESPONSE).encode())) as provider,
+        wire_server(lambda _request: Reply(body=b'{"action":"NONE"}')) as guardrail,
+        gateway.scenario() as scenario,
+    ):
+        model: Final = scenario.model(api_base=f"{provider.url}/v1", api_key="synthetic-provider-key")
+        team_id: Final = scenario.team(models=[model])
+        member: Final = scenario.member(team_id)
+        team_admin: Final = scenario.member(team_id, role="admin")
+        registration_key: Final = scenario.key(user_id=member, team_id=team_id, models=[model])
+        plain_member_key: Final = scenario.key(user_id=member, team_id=team_id, models=[model])
+        plain_team_admin_key: Final = scenario.key(user_id=team_admin, team_id=team_id, models=[model])
+        member_key: Final = scenario.key(
+            user_id=member,
+            team_id=team_id,
+            models=[model],
+            allowed_routes=["/guardrails/submissions"],
+        )
+        team_admin_key: Final = scenario.key(
+            user_id=team_admin,
+            team_id=team_id,
+            models=[model],
+            allowed_routes=["/guardrails/submissions"],
+        )
+        name: Final = f"submitted-auth-{uuid.uuid4().hex}"
+        guardrail_id: Final = _register(
+            gateway,
+            scenario.cleanups,
+            name,
+            team_id,
+            registration_key,
+            f"{guardrail.url}/beta/litellm_basic_guardrail_api",
+        )
+        plain_responses: Final = (
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/approve", key=plain_team_admin_key),
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/reject", key=plain_team_admin_key),
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/approve", key=plain_member_key),
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/reject", key=plain_member_key),
+        )
+        expected_routes: Final = tuple(
+            f"/guardrails/submissions/{guardrail_id}/{action}" for action in ("approve", "reject", "approve", "reject")
+        )
+        expected_user_ids: Final = (team_admin, team_admin, member, member)
+        expected_plain_errors: Final = tuple(
+            _ErrorResponse(
+                error=_ProxyError(
+                    message=(
+                        "Authentication Error, Only proxy admin can be used to generate, delete, update info for "
+                        f"new keys/users/teams. Route={route}. Your role=internal_user. "
+                        f"Your user_id={_mask_user_id(user_id)}"
+                    ),
+                    type="auth_error",
+                    param="None",
+                    code="401",
+                )
+            )
+            for route, user_id in zip(expected_routes, expected_user_ids, strict=True)
+        )
+        route_check_refusals: Final = tuple((401, error.model_dump()) for error in expected_plain_errors)
+        handler_forbidden: Final = _SubmissionForbiddenResponse(detail="Admin access required").model_dump()
+        handler_refusals: Final = ((403, handler_forbidden),) * 4
+        response_models: Final = {401: _ErrorResponse, 403: _SubmissionForbiddenResponse}
+        observed: Final = tuple(
+            (
+                response.status_code,
+                (
+                    response_models[response.status_code].model_validate_json(response.content).model_dump()
+                    if response.status_code in response_models
+                    else response.text
+                ),
+            )
+            for response in plain_responses
+        )
+        assert observed in (route_check_refusals, handler_refusals), tuple(
+            response.text for response in plain_responses
+        )
+        _assert_submission_status(guardrail_id, "pending_review", team_id)
+        responses: Final = (
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/approve", key=team_admin_key),
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/reject", key=team_admin_key),
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/approve", key=member_key),
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/reject", key=member_key),
+        )
+        _assert_submission_actions_are_forbidden(responses)
+        assert guardrail.drain() == ()
+        response: Final = _chat(gateway, model, registration_key, _PROMPT_PENDING, name)
+        assert response.choices[0].message.content == "provider response"
+        assert guardrail.drain() == ()
+
+
+def test_team_admins_and_internal_users_get_403_on_guardrail_submission_actions(gateway: Gateway) -> None:
+    pytest.skip(
+        "BUG: team admins and internal users get 401 from the route check instead of the "
+        "documented 403 on guardrail submission approve and reject"
+    )
+    with (
+        wire_server(lambda _request: Reply(body=json.dumps(_PROVIDER_RESPONSE).encode())) as provider,
+        wire_server(lambda _request: Reply(body=b'{"action":"NONE"}')) as guardrail,
+        gateway.scenario() as scenario,
+    ):
+        model: Final = scenario.model(api_base=f"{provider.url}/v1", api_key="synthetic-provider-key")
+        team_id: Final = scenario.team(models=[model])
+        member: Final = scenario.member(team_id)
+        team_admin: Final = scenario.member(team_id, role="admin")
+        member_key: Final = scenario.key(user_id=member, team_id=team_id, models=[model])
+        team_admin_key: Final = scenario.key(user_id=team_admin, team_id=team_id, models=[model])
+        name: Final = f"submitted-auth-{uuid.uuid4().hex}"
+        guardrail_id: Final = _register(
+            gateway,
+            scenario.cleanups,
+            name,
+            team_id,
+            member_key,
+            f"{guardrail.url}/beta/litellm_basic_guardrail_api",
+        )
+        responses: Final = (
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/approve", key=team_admin_key),
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/reject", key=team_admin_key),
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/approve", key=member_key),
+            gateway.request("POST", f"/guardrails/submissions/{guardrail_id}/reject", key=member_key),
+        )
+        _assert_submission_actions_are_forbidden(responses)
+        _assert_submission_status(guardrail_id, "pending_review", team_id)
+        assert guardrail.drain() == ()
+
+
+def test_team_guardrail_submissions_require_admin_approval(
+    gateway: Gateway,
+) -> None:
+    with (
+        wire_server(lambda _request: Reply(body=json.dumps(_PROVIDER_RESPONSE).encode())) as provider,
+        wire_server(lambda _request: Reply(body=b'{"action":"NONE"}')) as guardrail,
+        gateway.scenario() as scenario,
+    ):
+        model: Final = scenario.model(api_base=f"{provider.url}/v1", api_key="synthetic-provider-key")
+        team_id: Final = scenario.team(models=[model])
+        member: Final = scenario.member(team_id)
+        team_admin: Final = scenario.member(team_id, role="admin")
+        member_key_alias: Final = f"submission-member-{uuid.uuid4().hex}"
+        member_key: Final = scenario.key(
+            user_id=member,
+            team_id=team_id,
+            models=[model],
+            key_alias=member_key_alias,
+        )
+        team_admin_key: Final = scenario.key(
+            user_id=team_admin,
+            team_id=team_id,
+            models=[model],
+            key_alias=f"submission-team-admin-{uuid.uuid4().hex}",
+        )
+        other_team_id: Final = scenario.team(models=[model])
+        api_base: Final = f"{guardrail.url}/beta/litellm_basic_guardrail_api"
+        first_name: Final = f"submitted-member-{uuid.uuid4().hex}"
+        second_name: Final = f"submitted-admin-{uuid.uuid4().hex}"
+        other_name: Final = f"submitted-other-team-{uuid.uuid4().hex}"
+        first_id: Final = _register(gateway, scenario.cleanups, first_name, team_id, member_key, api_base)
+        second_id: Final = _register(gateway, scenario.cleanups, second_name, None, team_admin_key, api_base)
+        other_id: Final = _register(gateway, scenario.cleanups, other_name, other_team_id, gateway.key, api_base)
+
+        _assert_submission_status(first_id, "pending_review", team_id)
+        _assert_submission_status(second_id, "pending_review", team_id)
+        _assert_submission_status(other_id, "pending_review", other_team_id)
+
+        pending_apply: Final = gateway.request(
+            "POST", "/apply_guardrail", {"guardrail_name": first_name, "text": _APPLY_TEXT}
+        )
+        assert pending_apply.status_code == 404, pending_apply.text
+        assert _ErrorResponse.model_validate_json(pending_apply.content).model_dump() == {
+            "error": {
+                "message": (
+                    f"Guardrail '{first_name}' not found. "
+                    "Please ensure the guardrail is configured in your LiteLLM proxy."
+                ),
+                "type": "internal_server_error",
+                "param": None,
+                "code": "404",
+            }
+        }, pending_apply.text
+
+        pending_chat: Final = _chat(gateway, model, member_key, _PROMPT_PENDING, first_name)
+        assert len(pending_chat.choices) == 1
+        assert guardrail.drain() == ()
+
+        _assert_submission_status(first_id, "pending_review", team_id)
+
+        approved: Final = gateway.request("POST", f"/guardrails/submissions/{first_id}/approve")
+        assert approved.status_code == 200, approved.text
+        assert _AdminActionResponse.model_validate_json(approved.content).model_dump() == {
+            "guardrail_id": first_id,
+            "status": "active",
+            "message": "Guardrail approved",
+        }, approved.text
+        _assert_submission_status(first_id, "active", team_id)
+
+        approved_chat: Final = _chat(gateway, model, member_key, _PROMPT_APPROVED, first_name)
+        assert len(approved_chat.choices) == 1
+        active_apply: Final = gateway.request(
+            "POST", "/apply_guardrail", {"guardrail_name": first_name, "text": _APPLY_TEXT}
+        )
+        assert active_apply.status_code == 200, active_apply.text
+        assert _ApplyResponse.model_validate_json(active_apply.content).response_text == _APPLY_TEXT, active_apply.text
+
+        rejected: Final = gateway.request("POST", f"/guardrails/submissions/{second_id}/reject")
+        assert rejected.status_code == 200, rejected.text
+        assert _AdminActionResponse.model_validate_json(rejected.content).model_dump() == {
+            "guardrail_id": second_id,
+            "status": "rejected",
+            "message": "Guardrail rejected",
+        }, rejected.text
+        _assert_submission_status(second_id, "rejected", team_id)
+        rejected_chat: Final = _chat(gateway, model, member_key, _PROMPT_REJECTED, second_name)
+        assert len(rejected_chat.choices) == 1
+        rejected_apply: Final = gateway.request(
+            "POST", "/apply_guardrail", {"guardrail_name": second_name, "text": _APPLY_TEXT}
+        )
+        assert rejected_apply.status_code == 404, rejected_apply.text
+        assert _ErrorResponse.model_validate_json(rejected_apply.content).model_dump() == {
+            "error": {
+                "message": (
+                    f"Guardrail '{second_name}' not found. "
+                    "Please ensure the guardrail is configured in your LiteLLM proxy."
+                ),
+                "type": "internal_server_error",
+                "param": None,
+                "code": "404",
+            }
+        }, rejected_apply.text
+
+        team_list: Final = gateway.request("GET", "/guardrails/submissions", key=member_key)
+        assert team_list.status_code == 200, team_list.text
+        team_submissions: Final = _SubmissionsResponse.model_validate_json(team_list.content)
+        assert {item.guardrail_id for item in team_submissions.submissions} == {first_id, second_id}, team_list.text
+        admin_list: Final = gateway.request("GET", "/guardrails/submissions")
+        assert admin_list.status_code == 200, admin_list.text
+        admin_submissions: Final = _SubmissionsResponse.model_validate_json(admin_list.content)
+        assert {first_id, second_id, other_id} <= {item.guardrail_id for item in admin_submissions.submissions}, (
+            admin_list.text
+        )
+
+        provider_requests: Final = provider.drain()
+        assert [(request.method, request.target) for request in provider_requests] == [
+            ("POST", "/v1/chat/completions"),
+            ("POST", "/v1/chat/completions"),
+            ("POST", "/v1/chat/completions"),
+        ]
+        for request, text in zip(provider_requests, (_PROMPT_PENDING, _PROMPT_APPROVED, _PROMPT_REJECTED), strict=True):
+            _assert_provider_request(request, text)
+        guardrail_requests: Final = guardrail.drain()
+        assert [(request.method, request.target) for request in guardrail_requests] == [
+            ("POST", "/beta/litellm_basic_guardrail_api"),
+            ("POST", "/beta/litellm_basic_guardrail_api"),
+        ]
+        _assert_guardrail_request(
+            guardrail_requests[0],
+            _PROMPT_APPROVED,
+            model,
+            member_key,
+            key_alias=member_key_alias,
+        )
+        _assert_guardrail_request(guardrail_requests[1], _APPLY_TEXT, None, gateway.key)

tests/integration/observability/test_apply_guardrail_endpoint.py

@@ -0,0 +1,313 @@
+import json
+import uuid
+from contextlib import ExitStack
+from pathlib import Path
+from typing import Final
+
+import httpx
+import yaml
+from integration._support.client import Gateway, string_value
+from integration._support.process import owned_proxy
+from integration._support.wire import Reply, Request, wire_server
+from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
+
+_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
+_BLOCK_REASON: Final = "synthetic blocked by apply endpoint"
+_BLOCKED_TEXT: Final = "forbidden integration prompt"
+_METADATA_TEXT: Final = "metadata forwarding prompt"
+_PROVIDER_RESPONSE: Final = {
+    "id": "chatcmpl-apply-guardrail",
+    "object": "chat.completion",
+    "created": 1,
+    "model": "gpt-4o-mini",
+    "choices": [
+        {"index": 0, "message": {"role": "assistant", "content": "provider response"}, "finish_reason": "stop"}
+    ],
+    "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+}
+
+
+class _CreatedGuardrail(BaseModel):
+    guardrail_id: str
+    guardrail_name: str
+
+
+class _ApplyResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    response_text: str
+
+
+class _ApplyErrorBody(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    message: str
+    type: str
+    param: str | None
+    code: int
+
+
+class _ApplyError(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    error: _ApplyErrorBody
+
+
+def _delete_guardrail(gateway: Gateway, guardrail_id: str) -> None:
+    deleted: Final = gateway.request("DELETE", f"/guardrails/{guardrail_id}")
+    assert deleted.status_code == 200, deleted.text
+
+
+def _create_guardrail(
+    gateway: Gateway,
+    cleanups: ExitStack,
+    name: str,
+    params: dict[str, JsonValue],
+) -> str:
+    response: Final = gateway.request(
+        "POST",
+        "/guardrails",
+        {"guardrail": {"guardrail_name": name, "litellm_params": params}},
+    )
+    assert response.status_code == 200, response.text
+    created: Final = _CreatedGuardrail.model_validate_json(response.content)
+    assert created.guardrail_name == name, response.text
+    cleanups.callback(_delete_guardrail, gateway, created.guardrail_id)
+    return created.guardrail_name
+
+
+def _assert_guardrail_request(request: Request, text: str, response: httpx.Response) -> None:
+    assert (request.method, request.target) == ("POST", "/beta/litellm_basic_guardrail_api")
+    payload: Final = _JSON_OBJECT.validate_json(request.body)
+    assert set(payload) == {
+        "input_type",
+        "litellm_call_id",
+        "litellm_trace_id",
+        "structured_messages",
+        "images",
+        "tools",
+        "texts",
+        "request_data",
+        "request_headers",
+        "litellm_version",
+        "additional_provider_specific_params",
+        "tool_calls",
+        "model",
+    }, response.text
+    assert {
+        field: payload[field]
+        for field in (
+            "input_type",
+            "structured_messages",
+            "images",
+            "tools",
+            "texts",
+            "additional_provider_specific_params",
+            "tool_calls",
+            "model",
+        )
+    } == {
+        "input_type": "request",
+        "structured_messages": None,
+        "images": None,
+        "tools": None,
+        "texts": [text],
+        "additional_provider_specific_params": {},
+        "tool_calls": None,
+        "model": None,
+    }, response.text
+    assert payload["request_data"] == {}, response.text
+    request_headers: Final = payload["request_headers"]
+    assert request_headers is None or isinstance(request_headers, dict), response.text
+    if isinstance(request_headers, dict):
+        assert "authorization" not in request_headers, response.text
+    assert payload["litellm_call_id"] is None or isinstance(payload["litellm_call_id"], str), response.text
+    assert payload["litellm_trace_id"] is None or isinstance(payload["litellm_trace_id"], str), response.text
+    assert payload["litellm_version"] is None or isinstance(payload["litellm_version"], str), response.text
+    if response.headers.get("x-litellm-call-id") is not None:
+        assert payload["litellm_call_id"] == response.headers["x-litellm-call-id"], response.text
+    assert request.headers["x-api-key"] == "synthetic-apply-guardrail-key", response.text
+
+
+def test_apply_guardrail_returns_a_block_error_for_generic_guardrails(gateway: Gateway) -> None:
+    def block(_request: Request) -> Reply:
+        return Reply(body=json.dumps({"action": "BLOCKED", "blocked_reason": _BLOCK_REASON}).encode())
+
+    with wire_server(block) as guardrail, gateway.scenario() as scenario:
+        name: Final = _create_guardrail(
+            gateway,
+            scenario.cleanups,
+            f"blocked-apply-{uuid.uuid4().hex}",
+            {
+                "guardrail": "generic_guardrail_api",
+                "mode": "pre_call",
+                "default_on": False,
+                "api_base": f"{guardrail.url}/beta/litellm_basic_guardrail_api",
+                "api_key": "synthetic-apply-guardrail-key",
+            },
+        )
+        response: Final = gateway.request(
+            "POST",
+            "/guardrails/apply_guardrail",
+            {"guardrail_name": name, "text": _BLOCKED_TEXT},
+        )
+        assert response.status_code == 400, response.text
+        error: Final = _ApplyError.model_validate_json(response.content)
+        assert error.model_dump() == {
+            "error": {
+                "message": _BLOCK_REASON,
+                "type": "internal_server_error",
+                "param": None,
+                "code": 400,
+            }
+        }, response.text
+        requests: Final = guardrail.drain()
+        assert [(request.method, request.target) for request in requests] == [
+            ("POST", "/beta/litellm_basic_guardrail_api")
+        ]
+        request: Final = requests[0]
+        _assert_guardrail_request(request, _BLOCKED_TEXT, response)
+
+
+def test_apply_guardrail_forwards_messages_to_the_guardrail(gateway: Gateway) -> None:
+    def allow(_request: Request) -> Reply:
+        return Reply(body=json.dumps({"data": {"guardrailsResult": {"Allowed": True, "Reason": ""}}}).encode())
+
+    text: Final = f"akto apply text {uuid.uuid4().hex}"
+    messages: Final = [
+        {"role": "system", "content": "akto system prompt"},
+        {"role": "user", "content": "akto user content different from the text field"},
+    ]
+    with wire_server(allow) as akto, gateway.scenario() as scenario:
+        name: Final = _create_guardrail(
+            gateway,
+            scenario.cleanups,
+            f"akto-apply-{uuid.uuid4().hex}",
+            {
+                "guardrail": "akto",
+                "mode": "pre_call",
+                "default_on": False,
+                "akto_base_url": akto.url,
+                "akto_api_key": "synthetic-akto-apply-key",
+            },
+        )
+        with_messages: Final = gateway.request(
+            "POST",
+            "/guardrails/apply_guardrail",
+            {"guardrail_name": name, "text": text, "messages": messages},
+        )
+        assert with_messages.status_code == 200, with_messages.text
+        assert _ApplyResponse.model_validate_json(with_messages.content).model_dump() == {"response_text": text}, (
+            with_messages.text
+        )
+        without_messages: Final = gateway.request(
+            "POST",
+            "/guardrails/apply_guardrail",
+            {"guardrail_name": name, "text": text},
+        )
+        assert without_messages.status_code == 200, without_messages.text
+        assert _ApplyResponse.model_validate_json(without_messages.content).model_dump() == {"response_text": text}, (
+            without_messages.text
+        )
+
+        requests: Final = akto.drain()
+        assert [(request.method, request.target) for request in requests] == [
+            ("POST", "/api/http-proxy?akto_connector=litellm&guardrails=true"),
+            ("POST", "/api/http-proxy?akto_connector=litellm&guardrails=true"),
+        ]
+        assert [request.headers["authorization"] for request in requests] == ["synthetic-akto-apply-key"] * 2
+        bodies: Final = tuple(
+            _JSON_OBJECT.validate_json(
+                string_value(
+                    json.loads(string_value(_JSON_OBJECT.validate_json(request.body)["requestPayload"]))["body"]
+                )
+            )
+            for request in requests
+        )
+        assert bodies[0]["messages"] == messages, bodies[0]
+        assert bodies[1]["messages"] == [{"role": "user", "content": text}], bodies[1]
+
+
+def test_apply_guardrail_forwards_metadata_to_custom_code(
+    gateway: Gateway,
+    tmp_path: Path,
+) -> None:
+    def record(request: Request) -> Reply:
+        assert request.target == "/metadata"
+        return Reply(body=b'{"received":true}')
+
+    with wire_server(record) as recorder:
+        config: Final = _JSON_OBJECT.validate_python(
+            yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+        )
+        settings: Final = _JSON_OBJECT.validate_python(config["litellm_settings"])
+        config_path: Final = tmp_path / "apply-guardrail-owned-proxy.yaml"
+        config_path.write_text(
+            yaml.safe_dump(
+                {
+                    **config,
+                    "litellm_settings": {**settings, "user_url_allowed_hosts": ["127.0.0.1"]},
+                }
+            )
+        )
+        custom_code: Final = f"""
+async def apply_guardrail(inputs, request_data, input_type):
+    await http_post(
+        "{recorder.url}/metadata",
+        body={{
+            "texts": inputs.get("texts", []),
+            "metadata": request_data.get("metadata", {{}}),
+            "input_type": input_type,
+        }},
+    )
+    return allow()
+"""
+        with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate:
+            with candidate.scenario() as scenario:
+                name: Final = _create_guardrail(
+                    candidate,
+                    scenario.cleanups,
+                    f"metadata-apply-{uuid.uuid4().hex}",
+                    {
+                        "guardrail": "custom_code",
+                        "mode": "pre_call",
+                        "default_on": False,
+                        "custom_code": custom_code,
+                    },
+                )
+                with_metadata: Final = candidate.request(
+                    "POST",
+                    "/guardrails/apply_guardrail",
+                    {
+                        "guardrail_name": name,
+                        "text": _METADATA_TEXT,
+                        "metadata": {"forbidden_topics": ["tax"]},
+                    },
+                )
+                assert with_metadata.status_code == 200, with_metadata.text
+                assert _ApplyResponse.model_validate_json(with_metadata.content).model_dump() == {
+                    "response_text": _METADATA_TEXT
+                }, with_metadata.text
+                without_metadata: Final = candidate.request(
+                    "POST",
+                    "/guardrails/apply_guardrail",
+                    {"guardrail_name": name, "text": _METADATA_TEXT},
+                )
+                assert without_metadata.status_code == 200, without_metadata.text
+                assert _ApplyResponse.model_validate_json(without_metadata.content).model_dump() == {
+                    "response_text": _METADATA_TEXT
+                }, without_metadata.text
+
+        requests: Final = recorder.drain()
+        assert [(request.method, request.target) for request in requests] == [
+            ("POST", "/metadata"),
+            ("POST", "/metadata"),
+        ]
+        assert tuple(_JSON_OBJECT.validate_json(request.body) for request in requests) == (
+            {
+                "texts": [_METADATA_TEXT],
+                "metadata": {"forbidden_topics": ["tax"]},
+                "input_type": "request",
+            },
+            {"texts": [_METADATA_TEXT], "metadata": {}, "input_type": "request"},
+        )

tests/integration/observability/test_presidio_entity_masking.py

@@ -11,7 +11,7 @@ import httpx
 import pytest
 from integration._support.client import Gateway, string_value
 from integration._support.wire import Reply, Request, Wire, wire_server
-from pydantic import JsonValue
+from pydantic import BaseModel, ConfigDict, JsonValue
 
 CARD: Final = "4111-1111-1111-1111"
 EMAIL: Final = "jane.doe@example.com"
@@ -24,6 +24,27 @@ RECOGNIZERS: Final = {
 }
 
 
+class _ApplyGuardrailResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    response_text: str
+
+
+class _ApplyGuardrailErrorBody(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    message: str
+    type: str
+    param: str | None
+    code: str
+
+
+class _ApplyGuardrailError(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    error: _ApplyGuardrailErrorBody
+
+
 def _detect(entity: str, text: str) -> tuple[dict[str, JsonValue], ...]:
     return tuple(
         {"entity_type": entity, "start": match.start(), "end": match.end(), "score": 0.95}
@@ -91,6 +112,56 @@ def _requested_entities(analyzer: Wire) -> list[JsonValue]:
     return [json.loads(request.body).get("entities") for request in analyzer.drain()]
 
 
+def _assert_apply_guardrail_successes(
+    presidio: Presidio,
+    text: str,
+    responses: tuple[httpx.Response, ...],
+) -> None:
+    assert tuple(response.status_code for response in responses) == (200,) * len(responses), tuple(
+        response.text for response in responses
+    )
+    expected_response: Final = {"response_text": "alias contract <CREDIT_CARD> and <EMAIL_ADDRESS>"}
+    assert tuple(
+        _ApplyGuardrailResponse.model_validate_json(response.content).model_dump() for response in responses
+    ) == (expected_response,) * len(responses), tuple(response.text for response in responses)
+
+    analyzer_requests: Final = presidio.analyzer.drain()
+    assert len(analyzer_requests) == len(responses)
+    assert tuple((request.method, request.target) for request in analyzer_requests) == (
+        ("POST", "/analyze"),
+    ) * len(responses)
+    expected_analyzer_body: Final = {"text": text, "language": "en"}
+    assert tuple(json.loads(request.body) for request in analyzer_requests) == (
+        expected_analyzer_body,
+    ) * len(responses)
+
+    anonymizer_requests: Final = presidio.anonymizer.drain()
+    assert len(anonymizer_requests) == len(responses)
+    assert tuple((request.method, request.target) for request in anonymizer_requests) == (
+        ("POST", "/anonymize"),
+    ) * len(responses)
+    expected_anonymizer_body: Final = {
+        "text": text,
+        "analyzer_results": [
+            {
+                "entity_type": "CREDIT_CARD",
+                "start": text.index(CARD),
+                "end": text.index(CARD) + len(CARD),
+                "score": 0.95,
+            },
+            {
+                "entity_type": "EMAIL_ADDRESS",
+                "start": text.index(EMAIL),
+                "end": text.index(EMAIL) + len(EMAIL),
+                "score": 0.95,
+            },
+        ],
+    }
+    assert tuple(json.loads(request.body) for request in anonymizer_requests) == (
+        expected_anonymizer_body,
+    ) * len(responses)
+
+
 def test_pre_call_masks_only_the_configured_entities_before_the_provider_sees_the_prompt(gateway: Gateway) -> None:
     with (
         _presidio(gateway, "pre_call", {"CREDIT_CARD": "MASK", "EMAIL_ADDRESS": "MASK"}) as presidio,
@@ -139,3 +210,192 @@ def test_apply_guardrail_with_the_default_config_masks_every_detected_entity(
         assert masked == "card <CREDIT_CARD> and email <EMAIL_ADDRESS>", masked
         assert _requested_entities(presidio.analyzer) == [None]
         assert len(presidio.anonymizer.drain()) == 1
+
+
+def test_apply_guardrail_forwards_documented_language_and_entities_to_presidio(gateway: Gateway) -> None:
+    pytest.skip("BUG: /guardrails/apply_guardrail drops the documented language and entities fields before Presidio")
+    with _presidio(gateway, "pre_call", None) as presidio:
+        text: Final = f"apply language contract {CARD} and {EMAIL}"
+        response: Final = gateway.request(
+            "POST",
+            "/guardrails/apply_guardrail",
+            {
+                "guardrail_name": presidio.name,
+                "text": text,
+                "language": "fr",
+                "entities": ["EMAIL_ADDRESS"],
+            },
+        )
+        assert response.status_code == 200, response.text
+        masked: Final = string_value(response.json()["response_text"])
+        assert masked == f"apply language contract {CARD} and <EMAIL_ADDRESS>", masked
+        analyzer_requests: Final = presidio.analyzer.drain()
+        assert len(analyzer_requests) == 1
+        assert json.loads(analyzer_requests[0].body) == {
+            "text": text,
+            "language": "fr",
+            "entities": ["EMAIL_ADDRESS"],
+        }, analyzer_requests[0].body
+        assert len(presidio.anonymizer.drain()) == 1
+
+
+def test_apply_guardrail_aliases_accept_admin_auth(gateway: Gateway) -> None:
+    with _presidio(gateway, "pre_call", None) as presidio:
+        text: Final = f"alias contract {CARD} and {EMAIL}"
+        body: Final = {"guardrail_name": presidio.name, "text": text}
+        paths: Final = ("/apply_guardrail", "/guardrails/apply_guardrail")
+
+        unauthenticated_responses: Final = tuple(gateway.client.post(path, json=body) for path in paths)
+        assert tuple(response.status_code for response in unauthenticated_responses) == (401, 401), tuple(
+            response.text for response in unauthenticated_responses
+        )
+        assert presidio.analyzer.drain() == ()
+        assert presidio.anonymizer.drain() == ()
+
+        admin_requests: Final = tuple(
+            gateway.client.post(path, json=body, headers=headers)
+            for path, headers in (
+                (paths[0], {"Authorization": f"Bearer {gateway.key}"}),
+                (paths[1], {"Authorization": f"Bearer {gateway.key}"}),
+                (paths[0], {"x-litellm-api-key": gateway.key}),
+                (paths[1], {"x-litellm-api-key": gateway.key}),
+            )
+        )
+        assert tuple(response.status_code for response in admin_requests) == (200, 200, 200, 200), tuple(
+            response.text for response in admin_requests
+        )
+        for response in admin_requests:
+            assert _ApplyGuardrailResponse.model_validate_json(response.content).model_dump() == {
+                "response_text": "alias contract <CREDIT_CARD> and <EMAIL_ADDRESS>"
+            }, response.text
+
+        analyzer_requests: Final = presidio.analyzer.drain()
+        assert len(analyzer_requests) == 4
+        assert all(request.method == "POST" and request.target == "/analyze" for request in analyzer_requests)
+        assert len({request.body for request in analyzer_requests}) == 1
+        assert json.loads(analyzer_requests[0].body) == {"text": text, "language": "en"}, analyzer_requests[0].body
+        anonymizer_requests: Final = presidio.anonymizer.drain()
+        assert len(anonymizer_requests) == 4
+        assert all(request.method == "POST" and request.target == "/anonymize" for request in anonymizer_requests)
+        assert len({request.body for request in anonymizer_requests}) == 1
+        assert json.loads(anonymizer_requests[0].body) == {
+            "text": text,
+            "analyzer_results": [
+                {
+                    "entity_type": "CREDIT_CARD",
+                    "start": text.index(CARD),
+                    "end": text.index(CARD) + len(CARD),
+                    "score": 0.95,
+                },
+                {
+                    "entity_type": "EMAIL_ADDRESS",
+                    "start": text.index(EMAIL),
+                    "end": text.index(EMAIL) + len(EMAIL),
+                    "score": 0.95,
+                },
+            ],
+        }, anonymizer_requests[0].body
+
+
+def test_long_apply_guardrail_alias_accepts_an_llm_api_scoped_key(gateway: Gateway) -> None:
+    with (
+        _presidio(gateway, "pre_call", None) as presidio,
+        gateway.scenario() as scenario,
+    ):
+        proxy_admin: Final = scenario.user(user_role="proxy_admin")
+        key: Final = scenario.key(user_id=proxy_admin, key_type="llm_api")
+        text: Final = f"alias contract {CARD} and {EMAIL}"
+        body: Final = {"guardrail_name": presidio.name, "text": text}
+        path: Final = "/guardrails/apply_guardrail"
+
+        assert presidio.analyzer.drain() == ()
+        assert presidio.anonymizer.drain() == ()
+        responses: Final = tuple(
+            gateway.client.post(path, json=body, headers=headers)
+            for headers in (
+                {"Authorization": f"Bearer {key}"},
+                {"x-litellm-api-key": key},
+            )
+        )
+        _assert_apply_guardrail_successes(presidio, text, responses)
+
+        unauthenticated: Final = gateway.client.post(path, json=body)
+        assert unauthenticated.status_code == 401, unauthenticated.text
+        assert _ApplyGuardrailError.model_validate_json(unauthenticated.content).model_dump() == {
+            "error": {
+                "message": "Authentication Error, No api key passed in.",
+                "type": "auth_error",
+                "param": "None",
+                "code": "401",
+            }
+        }, unauthenticated.text
+        assert presidio.analyzer.drain() == ()
+        assert presidio.anonymizer.drain() == ()
+
+
+def test_short_apply_guardrail_alias_accepts_an_llm_api_scoped_key(gateway: Gateway) -> None:
+    pytest.skip(
+        "BUG: llm_api scoped keys are refused on /apply_guardrail while "
+        "/guardrails/apply_guardrail serves them"
+    )
+    with (
+        _presidio(gateway, "pre_call", None) as presidio,
+        gateway.scenario() as scenario,
+    ):
+        proxy_admin: Final = scenario.user(user_role="proxy_admin")
+        key: Final = scenario.key(user_id=proxy_admin, key_type="llm_api")
+        text: Final = f"alias contract {CARD} and {EMAIL}"
+        body: Final = {"guardrail_name": presidio.name, "text": text}
+        path: Final = "/apply_guardrail"
+
+        assert presidio.analyzer.drain() == ()
+        assert presidio.anonymizer.drain() == ()
+        responses: Final = tuple(
+            gateway.client.post(path, json=body, headers=headers)
+            for headers in (
+                {"Authorization": f"Bearer {key}"},
+                {"x-litellm-api-key": key},
+            )
+        )
+        _assert_apply_guardrail_successes(presidio, text, responses)
+
+
+def test_apply_guardrail_aliases_accept_internal_user_auth(gateway: Gateway) -> None:
+    pytest.skip("BUG: non-admin virtual keys get 401 on /apply_guardrail and /guardrails/apply_guardrail")
+    with (
+        _presidio(gateway, "pre_call", None) as presidio,
+        gateway.scenario() as scenario,
+    ):
+        internal_user: Final = scenario.user(user_role="internal_user")
+        internal_key: Final = scenario.key(
+            user_id=internal_user,
+            key_alias=f"presidio-apply-short-{uuid.uuid4().hex}",
+        )
+        text: Final = f"alias contract {CARD} and {EMAIL}"
+        body: Final = {"guardrail_name": presidio.name, "text": text}
+        paths: Final = ("/apply_guardrail", "/guardrails/apply_guardrail")
+        internal_auth_requests: Final = (
+            (paths[0], {"Authorization": f"Bearer {internal_key}"}),
+            (paths[1], {"Authorization": f"Bearer {internal_key}"}),
+            (paths[0], {"x-litellm-api-key": internal_key}),
+            (paths[1], {"x-litellm-api-key": internal_key}),
+        )
+        responses: Final = tuple(
+            gateway.client.post(path, json=body, headers=headers) for path, headers in internal_auth_requests
+        )
+        assert tuple(response.status_code for response in responses) == (200, 200, 200, 200), tuple(
+            response.text for response in responses
+        )
+        assert (
+            tuple(_ApplyGuardrailResponse.model_validate_json(response.content).model_dump() for response in responses)
+            == ({"response_text": "alias contract <CREDIT_CARD> and <EMAIL_ADDRESS>"},) * 4
+        )
+        analyzer_requests: Final = presidio.analyzer.drain()
+        assert len(analyzer_requests) == 4
+        assert all(request.method == "POST" and request.target == "/analyze" for request in analyzer_requests)
+        assert len({request.body for request in analyzer_requests}) == 1
+        assert json.loads(analyzer_requests[0].body)["text"] == text
+        anonymizer_requests: Final = presidio.anonymizer.drain()
+        assert len(anonymizer_requests) == 4
+        assert all(request.method == "POST" and request.target == "/anonymize" for request in anonymizer_requests)
+        assert len({request.body for request in anonymizer_requests}) == 1

tests/integration/security/test_stored_config_slots.py

@@ -64,6 +64,10 @@ from integration.security._sweeps import (
     sweep_all,
     sweep_sink,
 )
+from litellm.types.guardrails import (
+    GuardrailInfoResponse,
+    ListGuardrailsResponse,
+)
 
 OUTCOMES: Final = ("success", "provider_4xx")
 BEDROCK_MODEL: Final = "bedrock/converse/anthropic.claude-haiku-4-5-20251001-v1:0"
@@ -81,6 +85,11 @@ VERTEX_MODEL_PATH: Final = (
 )
 
 
+class _CreatedGuardrail(BaseModel):
+    guardrail_id: str
+    guardrail_name: str
+
+
 @pytest.fixture(scope="module")
 def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
     """Shared proxy for slots created through the API, with team model_config overrides on."""
@@ -785,7 +794,7 @@ def _guardrail(request: Request) -> Reply:
     return Reply(body=json.dumps({"action": "NONE"}).encode())
 
 
-def _guardrail_params(url: str, secret: Canary) -> dict[str, object]:
+def _guardrail_params(url: str, secret: Canary) -> dict[str, JsonValue]:
     return {
         "guardrail": "generic_guardrail_api",
         "mode": "pre_call",
@@ -849,6 +858,121 @@ def test_config_guardrail_api_key_reaches_only_the_guardrail(
             )
 
 
+@pytest.mark.timeout(240)
+@pytest.mark.parametrize("outcome", OUTCOMES)
+def test_api_created_guardrail_api_key_is_encrypted_at_rest_and_reaches_only_the_guardrail(
+    rig: Rig, outcome: str, request: pytest.FixtureRequest
+) -> None:
+    started: Final = datetime.now(UTC)
+    secret: Final = canary("E1")
+    marker: Final = canary(MARKER)
+    name: Final = f"canary-api-guardrail-{uuid.uuid4().hex}"
+    with wire_server(_guardrail) as wire, rig.proxy.scenario() as scenario:
+        guardrail: Final = Recorder(wire)
+        created_response: Final = rig.proxy.request(
+            "POST",
+            "/guardrails",
+            {"guardrail": {"guardrail_name": name, "litellm_params": _guardrail_params(wire.url, secret)}},
+        )
+        assert created_response.status_code == 200, created_response.text
+        created: Final = _CreatedGuardrail.model_validate_json(created_response.content)
+        assert created.guardrail_name == name, created_response.text
+        guardrail_id: Final = created.guardrail_id
+
+        def delete_guardrail() -> None:
+            deleted: Final = rig.proxy.request("DELETE", f"/guardrails/{guardrail_id}")
+            assert deleted.status_code == 200, deleted.text
+            assert (
+                read_rows(
+                    'SELECT guardrail_id FROM "LiteLLM_GuardrailsTable" WHERE guardrail_id=%s',
+                    (guardrail_id,),
+                )
+                == []
+            )
+
+        scenario.cleanups.callback(delete_guardrail)
+        caller: Final = _caller(scenario, models=[CONFIG_MODEL])
+        response, request_id = _chat(rig.proxy, caller.key, CONFIG_MODEL, "E1", marker, outcome)
+        _guardrail_delivered(guardrail, marker, secret)
+        guardrail_requests: Final = guardrail.requests()
+        assert [(item.method, item.target) for item in guardrail_requests] == [("POST", GUARDRAIL_PATH)]
+
+        _assert_stored_without_canary(
+            "SELECT litellm_params->>'api_key' FROM \"LiteLLM_GuardrailsTable\" WHERE guardrail_id=%s",
+            (guardrail_id,),
+            secret,
+        )
+        params_rows: Final = read_rows(
+            'SELECT litellm_params::text AS params FROM "LiteLLM_GuardrailsTable" WHERE guardrail_id=%s',
+            (guardrail_id,),
+        )
+        assert len(params_rows) == 1, params_rows
+        stored_params: Final = string_value(params_rows[0]["params"])
+        assert find_canary(stored_params, (secret,)) == (), f"Slot {secret.slot} stored in plaintext JSON"
+
+        masked_secret: Final = f"{secret.value[:2]}{'*' * 4}{secret.value[-2:]}"
+        detail_routes: Final = (f"/guardrails/{guardrail_id}", f"/guardrails/{guardrail_id}/info")
+        detail_responses: Final = tuple(rig.proxy.request("GET", path) for path in detail_routes)
+        assert tuple(item.status_code for item in detail_responses) == (200, 200), tuple(
+            item.text for item in detail_responses
+        )
+        details: Final = tuple(GuardrailInfoResponse.model_validate_json(item.content) for item in detail_responses)
+        assert all(item.created_at is not None and item.updated_at is not None for item in details), details
+        assert (
+            tuple(
+                (
+                    item.guardrail_id,
+                    item.guardrail_name,
+                    item.guardrail_definition_location.value,
+                )
+                for item in details
+            )
+            == ((guardrail_id, name, "db"),) * 2
+        ), tuple(item.text for item in detail_responses)
+        assert (
+            tuple((item.created_at, item.updated_at) for item in details)
+            == ((details[0].created_at, details[0].updated_at),) * 2
+        )
+        assert all(item.litellm_params is not None for item in details), tuple(item.text for item in detail_responses)
+        assert (
+            tuple(
+                (
+                    item.litellm_params.default_on,
+                    item.litellm_params.api_base,
+                    item.litellm_params.api_key,
+                )
+                for item in details
+                if item.litellm_params is not None
+            )
+            == ((True, wire.url, masked_secret),) * 2
+        ), tuple(item.text for item in detail_responses)
+
+        list_response: Final = rig.proxy.request("GET", "/v2/guardrails/list")
+        assert list_response.status_code == 200, list_response.text
+        listed_guardrails: Final = ListGuardrailsResponse.model_validate_json(list_response.content).guardrails
+        matching_entries: Final = tuple(item for item in listed_guardrails if item.guardrail_id == guardrail_id)
+        assert len(matching_entries) == 1, list_response.text
+        assert find_canary(list_response.content, (secret,)) == (), list_response.text
+
+        _finish(
+            rig,
+            rig.proxy,
+            request,
+            secrets=(secret,),
+            marker=marker,
+            response=response,
+            request_id=request_id,
+            caller=caller,
+            ids={"model": CONFIG_MODEL, "guardrail_id": guardrail_id},
+            detail_routes=detail_routes,
+            reads=_reads(rig.proxy, {"/guardrails/list": {}, "/v2/guardrails/list": {}}),
+            extra_sinks={GUARDRAIL_SINK: guardrail.requests},
+            own_headers={GUARDRAIL_SINK: ("x-api-key", "E1")},
+            context=f"slot E1 (API-created DB guardrail), {outcome}",
+            since=started,
+        )
+
+
 def _langfuse(request: Request) -> Reply:
     if request.method == "GET" and request.target.startswith("/api/public/projects"):
         return Reply(body=json.dumps({"data": [{"id": "canary-project", "name": "canary"}]}).encode())