chore(proxy): also scrub guardrail callbacks / module paths from DB overlay

A guardrail entry's ``callbacks`` list (v1: ``{name: {callbacks:[...]}}``,
v2: ``{guardrail_name, litellm_params: {callbacks: [...], guardrail:
"module.path"}}``) is iterated during config load and threaded through
``get_instance_fn``. A PROXY_ADMIN persisting
``litellm_settings.guardrails[*].callbacks: ["s3://..."]`` or
``litellm_settings.guardrails[*].litellm_params.guardrail: "s3://..."``
via ``/config/update`` was not covered by the previous scrub matrix.

Walk both v1 and v2 entry shapes and null out remote-URL callbacks /
module-path values before the merge. Adds four regression tests.
This commit is contained in:
user 2026-05-14 01:24:51 +00:00
parent 3aef0642a8
commit aced4295bb
No known key found for this signature in database
2 changed files with 110 additions and 0 deletions

View File

@ -3139,6 +3139,28 @@ def _is_remote_module_url(value: Any) -> bool:
)
def _scrub_guardrail_inner(inner: Dict[str, Any]) -> None:
"""Strip remote-URL entries from a guardrail's ``callbacks`` list
and ``guardrail`` (v2 module-path) field. Mutates in place."""
cbs = inner.get("callbacks")
if isinstance(cbs, list):
cleaned = [c for c in cbs if not _is_remote_module_url(c)]
if len(cleaned) != len(cbs):
verbose_proxy_logger.warning(
"Refused %d remote-URL entries from DB-overlay "
"litellm_settings.guardrails[...].callbacks",
len(cbs) - len(cleaned),
)
inner["callbacks"] = cleaned
if _is_remote_module_url(inner.get("guardrail")):
verbose_proxy_logger.warning(
"Refused remote-URL guardrail module from DB-overlay "
"litellm_settings.guardrails[...].guardrail: %r",
inner.get("guardrail"),
)
inner["guardrail"] = None
def _scrub_db_overlay_remote_module_loads(section: str, db_value: Any) -> Any:
"""Strip ``s3://`` / ``gcs://`` entries from the DB-overlay value for
fields whose contents reach ``get_instance_fn``. The same scheme is
@ -3192,6 +3214,26 @@ def _scrub_db_overlay_remote_module_loads(section: str, db_value: Any) -> Any:
item.get("custom_handler"),
)
item["custom_handler"] = None
# ``litellm_settings.guardrails`` is a list of single-key dicts in
# v1 ({guardrail_name: {callbacks: [...], default_on: bool}}) or a
# list of v2 entries ({guardrail_name, litellm_params: {guardrail:
# "module.path", callbacks: [...]}}). Both shapes terminate in
# ``callbacks`` (a list) or ``guardrail`` (a single dotted name)
# that flow into ``get_instance_fn`` during config load.
if section == "litellm_settings":
guardrails = sanitized.get("guardrails")
if isinstance(guardrails, list):
for entry in guardrails:
if not isinstance(entry, dict):
continue
for inner in entry.values():
if not isinstance(inner, dict):
continue
_scrub_guardrail_inner(inner)
lp = entry.get("litellm_params")
if isinstance(lp, dict):
_scrub_guardrail_inner(lp)
# ``general_settings.litellm_jwtauth.custom_validate`` is a nested
# string field.
if section == "general_settings":

View File

@ -71,6 +71,74 @@ def test_custom_provider_map_custom_handler_stripped():
assert cleaned["custom_provider_map"][1]["custom_handler"] is None
def test_litellm_settings_guardrails_v1_callbacks_stripped():
# v1 guardrail shape: {guardrail_name: {callbacks: [...], default_on: bool}}
overlay = {
"guardrails": [
{
"prompt_injection": {
"default_on": True,
"callbacks": [
"lakera_prompt_injection",
"s3://attacker/m.i",
"gcs://attacker/m.i",
],
}
}
]
}
cleaned = _scrub_db_overlay_remote_module_loads("litellm_settings", overlay)
assert cleaned["guardrails"][0]["prompt_injection"]["callbacks"] == [
"lakera_prompt_injection"
]
def test_litellm_settings_guardrails_v2_callbacks_and_guardrail_stripped():
# v2 shape: {guardrail_name, litellm_params: {guardrail: "module.path", callbacks: [...]}}
overlay = {
"guardrails": [
{
"guardrail_name": "custom",
"litellm_params": {
"guardrail": "s3://attacker/m.i",
"mode": "pre_call",
"callbacks": ["lakera", "s3://attacker/cb.i"],
},
}
]
}
cleaned = _scrub_db_overlay_remote_module_loads("litellm_settings", overlay)
lp = cleaned["guardrails"][0]["litellm_params"]
assert lp["guardrail"] is None
assert lp["callbacks"] == ["lakera"]
assert lp["mode"] == "pre_call"
def test_litellm_settings_guardrails_local_dotted_name_preserved():
overlay = {
"guardrails": [
{
"guardrail_name": "custom",
"litellm_params": {
"guardrail": "custom_module.MyGuardrail",
"callbacks": ["my_module.cb", "langfuse"],
},
}
]
}
cleaned = _scrub_db_overlay_remote_module_loads("litellm_settings", overlay)
lp = cleaned["guardrails"][0]["litellm_params"]
assert lp["guardrail"] == "custom_module.MyGuardrail"
assert lp["callbacks"] == ["my_module.cb", "langfuse"]
def test_litellm_settings_guardrails_non_list_passthrough():
cleaned = _scrub_db_overlay_remote_module_loads(
"litellm_settings", {"guardrails": "not-a-list"}
)
assert cleaned["guardrails"] == "not-a-list"
def test_pass_through_endpoints_target_stripped():
overlay = {
"pass_through_endpoints": [