fix(guardrails): Alice WonderFence rejects non-string credential overrides

MASK verdicts on legacy functions[].description now write the redacted value
back into request_data["functions"] via the same path-based mechanism used for
inputs["tools"]. Previously the path was discarded and MASK was silently
ignored, leaving the original unredacted description forwarded to the model.

Also update the test that asserted the old "detection only, no mask" behaviour
to reflect the corrected semantics (DETECT still logs without mutating; only
MASK writes back).
This commit is contained in:
lior-k 2026-06-21 21:21:26 +03:00
parent d61eda2d07
commit 8c99df4cfc
No known key found for this signature in database
4 changed files with 74 additions and 35 deletions

View file

@ -180,10 +180,10 @@ class WonderFenceGuardrail(CustomGuardrail):
tool_def_paths, tool_def_segments = tool_definition_segments(inputs)
# Legacy top-level functions[] only exist on the request body; the
# translation layer does not surface them in inputs, so read request_data.
function_def_segments = (
function_def_paths, function_def_segments = (
function_definition_segments(request_data)
if input_type == "request"
else []
else ([], [])
)
if (
not texts
@ -264,7 +264,9 @@ class WonderFenceGuardrail(CustomGuardrail):
tool_def_verdicts=verdicts[
n_text + n_tool : n_text + n_tool + n_tool_def
],
function_def_paths=function_def_paths,
function_def_verdicts=verdicts[n_text + n_tool + n_tool_def :],
function_def_request_data=request_data,
)
except WonderFenceBlockedError as e:

View file

@ -125,23 +125,25 @@ def tool_definition_segments(
return paths, segments
def function_definition_segments(request_data: dict) -> list[str]:
"""Description strings from the deprecated top-level ``functions[]`` request
parameter.
def function_definition_segments(
request_data: dict,
) -> tuple[list[list[Any]], list[str]]:
"""Description paths and texts from the deprecated ``functions[]`` parameter.
Each entry is shaped like a tool's ``function`` object
(``{name, description, parameters}``) and LiteLLM forwards it to providers,
so its descriptions are scanned. Read from ``request_data`` because the chat
translation layer surfaces ``tools`` but not ``functions`` in ``inputs``.
Detection only (BLOCK/DETECT) -- ``functions`` has no inputs write-back path,
so it is not masked.
Each entry is shaped like a tool's ``function`` object so the same
description walker applies. Returns ``(paths, segments)`` so MASK verdicts
can be written back into ``request_data["functions"]`` the same way
``tool_def_paths`` are used for ``inputs["tools"]``.
"""
functions = request_data.get("functions") or []
paths: list[list[Any]] = []
segments: list[str] = []
for fn in functions:
for i, fn in enumerate(functions):
if isinstance(fn, dict):
segments.extend(text for _path, text in _description_strings(fn, []))
return segments
for sub_path, text in _description_strings(fn, []):
paths.append([i, *sub_path])
segments.append(text)
return paths, segments
def _set_by_path(root: Any, path: list[Any], value: Any) -> None:
@ -209,21 +211,23 @@ def apply_verdicts(
tool_verdicts: list[SegmentVerdict] | None = None,
tool_def_paths: list[list[Any]] | None = None,
tool_def_verdicts: list[SegmentVerdict] | None = None,
function_def_paths: list[list[Any]] | None = None,
function_def_verdicts: list[SegmentVerdict] | None = None,
function_def_request_data: dict | None = None,
) -> GenericGuardrailAPIInputs:
"""Apply per-segment verdicts back onto request text, tool-call args, and
tool-definition descriptions.
"""Apply per-segment verdicts back onto request text, tool-call args,
tool-definition descriptions, and legacy function-definition descriptions.
Any BLOCK across any group raises ``WonderFenceBlockedError`` with
detections/correlation ids aggregated across all blocked segments. Otherwise
each MASK verdict rewrites the slot its segment came from and DETECT is
logged. ``function_def_verdicts`` (legacy ``functions[]``) are detection
only: BLOCK raises, anything else is logged, never masked.
logged.
"""
tool_indices = tool_indices or []
tool_verdicts = tool_verdicts or []
tool_def_paths = tool_def_paths or []
tool_def_verdicts = tool_def_verdicts or []
function_def_paths = function_def_paths or []
function_def_verdicts = function_def_verdicts or []
blocked = [
@ -260,12 +264,10 @@ def apply_verdicts(
if masked is not None:
_set_by_path(tools, path, masked)
for verdict in function_def_verdicts:
if verdict.action in ("MASK", "DETECT"):
logger.warning(
"Alice WonderFence (apply_guardrail): DETECT function definition guardrail=%s correlation_id=%s",
guardrail_name,
verdict.correlation_ids[0] if verdict.correlation_ids else None,
)
functions = (function_def_request_data or {}).get("functions") or []
for path, verdict in zip(function_def_paths, function_def_verdicts):
masked = _masked_value(verdict, guardrail_name, "function definition")
if masked is not None and functions:
_set_by_path(functions, path, masked)
return inputs

View file

@ -933,11 +933,40 @@ async def test_apply_guardrail_scans_legacy_functions_when_no_other_content(
@pytest.mark.asyncio
async def test_apply_guardrail_legacy_function_not_masked_only_detected(
async def test_apply_guardrail_legacy_function_detect_does_not_mutate(
guardrail_and_client, make_request_data
):
"""A non-BLOCK verdict on a function definition passes through without
mutating request_data['functions'] (detection only, no mask write-back)."""
"""A DETECT verdict on a function definition logs but does not rewrite it."""
guardrail, client = guardrail_and_client
def evaluate(prompt, **kwargs):
r = Mock()
r.action = "DETECT" if "watch" in prompt else "NO_ACTION"
r.action_text = None
r.detections = []
r.correlation_id = None
return r
client.evaluate_prompt.side_effect = evaluate
request_data = make_request_data(
functions=[_legacy_function(description="watch this")]
)
out = await guardrail.apply_guardrail(
inputs={"texts": ["hi"]},
request_data=request_data,
input_type="request",
)
assert out is not None
assert request_data["functions"][0]["description"] == "watch this"
@pytest.mark.asyncio
async def test_apply_guardrail_masks_legacy_function_description_in_place(
guardrail_and_client, make_request_data
):
"""A MASK verdict on a functions[] description must be written back into
request_data['functions'], not left as the original unredacted text."""
guardrail, client = guardrail_and_client
def evaluate(prompt, **kwargs):
@ -953,11 +982,9 @@ async def test_apply_guardrail_legacy_function_not_masked_only_detected(
request_data = make_request_data(
functions=[_legacy_function(description="contains secret stuff")]
)
out = await guardrail.apply_guardrail(
await guardrail.apply_guardrail(
inputs={"texts": ["hi"]},
request_data=request_data,
input_type="request",
)
assert out is not None
# functions left untouched (no mask write-back)
assert request_data["functions"][0]["description"] == "contains secret stuff"
assert request_data["functions"][0]["description"] == "[REDACTED]"

View file

@ -104,8 +104,9 @@ def test_tool_definition_segments_ignores_non_dict_tools_and_blank_descriptions(
# --------------- function_definition_segments (legacy functions[]) ---------------
def test_function_definition_segments_extracts_descriptions():
def test_function_definition_segments_extracts_descriptions_and_paths():
from litellm.proxy.guardrails.guardrail_hooks.alice_wonderfence.processing import (
_set_by_path,
function_definition_segments,
)
@ -125,7 +126,13 @@ def test_function_definition_segments_extracts_descriptions():
{"name": "f", "description": " "},
]
}
assert set(function_definition_segments(request_data)) == {"TOP_DESC", "PARAM_DESC"}
paths, segments = function_definition_segments(request_data)
assert set(segments) == {"TOP_DESC", "PARAM_DESC"}
for path, text in zip(paths, segments):
_set_by_path(request_data["functions"], path, f"<{text}>")
fn = request_data["functions"][0]
assert fn["description"] == "<TOP_DESC>"
assert fn["parameters"]["properties"]["city"]["description"] == "<PARAM_DESC>"
def test_function_definition_segments_empty_when_absent():
@ -133,4 +140,5 @@ def test_function_definition_segments_empty_when_absent():
function_definition_segments,
)
assert function_definition_segments({"model": "gpt-4"}) == []
paths, segs = function_definition_segments({"model": "gpt-4"})
assert paths == [] and segs == []