From af97aa5bc634dc4cb866228be1ef229ff99ac198 Mon Sep 17 00:00:00 2001 From: user <70670632+stuxf@users.noreply.github.com> Date: Sun, 31 May 2026 07:58:29 +0000 Subject: [PATCH] chore(proxy): require credential re-supply when a non-admin changes a model endpoint Clearing an inherited credential on a destination change is not enough: an empty or absent api_key falls back to the proxy's environment secret (e.g. OPENAI_API_KEY) for OpenAI-compatible providers, so the operator's key would still be sent to the caller-chosen endpoint. A team admin could also defeat the per-field clear by supplying an unrelated empty/non-empty credential field. Instead of clearing, non-admin endpoint changes now require every credential the model already had to be re-provided with a non-empty value, otherwise the update is rejected; proxy admins are unaffected. The connection-test redirect guard applies the same non-empty value check so an empty request credential no longer counts as supplied. Adds direct regression tests for re-supply required, empty re-supply, partial re-supply, unchanged destination, and keyless models. --- .../health_endpoints/_health_endpoints.py | 4 +- .../model_management_endpoints.py | 62 ++++---- .../test_model_management_endpoints.py | 133 +++++++----------- 3 files changed, 84 insertions(+), 115 deletions(-) diff --git a/litellm/proxy/health_endpoints/_health_endpoints.py b/litellm/proxy/health_endpoints/_health_endpoints.py index 6282d3894f7..d29b9acea5f 100644 --- a/litellm/proxy/health_endpoints/_health_endpoints.py +++ b/litellm/proxy/health_endpoints/_health_endpoints.py @@ -109,11 +109,11 @@ def _reject_inherited_credential_redirect( destination override. """ inherited_credential = any( - field in config_litellm_params and field not in request_litellm_params + config_litellm_params.get(field) and not request_litellm_params.get(field) for field in _HEALTH_CREDENTIAL_FIELDS ) overrides_destination = any( - field in request_litellm_params for field in _HEALTH_DESTINATION_FIELDS + request_litellm_params.get(field) for field in _HEALTH_DESTINATION_FIELDS ) if inherited_credential and overrides_destination: raise HTTPException( diff --git a/litellm/proxy/management_endpoints/model_management_endpoints.py b/litellm/proxy/management_endpoints/model_management_endpoints.py index 713d386de5b..3ce666804da 100644 --- a/litellm/proxy/management_endpoints/model_management_endpoints.py +++ b/litellm/proxy/management_endpoints/model_management_endpoints.py @@ -130,16 +130,17 @@ def _decrypted_param(value: object) -> object: ) -def _strip_credentials_on_destination_change( - merged_litellm_params: dict, +def _assert_credential_resupplied_on_destination_change( patch_plaintext: dict, db_plaintext: "Callable[[str], object]", ) -> None: - """When a destination field (api_base/base_url/provider endpoint) changes, - drop every INHERITED credential — one the patch did not itself supply — so a - stored secret is never silently re-pointed at a new (possibly attacker- - controlled) endpoint. Cleared per field: supplying one credential must not - preserve the others.""" + """Changing a model's destination (api_base/base_url/provider endpoint) must + not send an inherited credential to the new endpoint. Clearing the credential + is NOT sufficient: an empty/absent key falls back to the proxy's environment + secret (e.g. OPENAI_API_KEY) and would still be sent there. So require the + caller to re-supply, with a non-empty value, every credential the model + already had; otherwise reject. A non-empty re-supplied value is the caller's + own credential going to their own endpoint, which is allowed.""" destination_changed = any( field in patch_plaintext and patch_plaintext[field] != db_plaintext(field) for field in _DESTINATION_LITELLM_PARAMS @@ -147,8 +148,17 @@ def _strip_credentials_on_destination_change( if not destination_changed: return for field in _CREDENTIAL_LITELLM_PARAMS: - if field not in patch_plaintext: - merged_litellm_params.pop(field, None) + if db_plaintext(field) and not patch_plaintext.get(field): + raise ProxyException( + message=( + f"Re-provide {field} (non-empty) when changing the model " + "endpoint; a stored credential cannot be reused with, or " + "cleared in favor of an environment fallback at, a new destination." + ), + type=ProxyErrorTypes.validation_error.value, + code=status.HTTP_400_BAD_REQUEST, + param=field, + ) def _is_pricing_field(field: str) -> bool: @@ -356,18 +366,6 @@ def update_db_model( merged_deployment_dict["model_info"].pop(field, None) # type: ignore merged_deployment_dict.get("litellm_params", {}).pop(field, None) # type: ignore - # Refuse to silently re-point a stored credential at a new endpoint: if the - # destination changed without a fresh credential, drop the inherited secret - # so it must be re-entered rather than forwarded to the new destination. - if updated_patch.litellm_params: - _strip_credentials_on_destination_change( - merged_deployment_dict["litellm_params"], # type: ignore - patch_litellm_plaintext, - lambda field: _decrypted_param( - getattr(db_model.litellm_params, field, None) - ), - ) - # convert to prisma compatible format prisma_compatible_model_dict = PrismaCompatibleUpdateDBModel() @@ -492,6 +490,14 @@ async def patch_model( user_api_key_dict=user_api_key_dict, ) + if patch_data.litellm_params and not is_proxy_admin(user_api_key_dict): + _assert_credential_resupplied_on_destination_change( + patch_data.litellm_params.model_dump(exclude_none=True), + lambda field: _decrypted_param( + getattr(db_model.litellm_params, field, None) + ), + ) + # Handle team model updates with proper alias management update_data = await _update_team_model_in_db( db_model=db_model, @@ -1519,13 +1525,13 @@ async def update_model( pass # Don't silently re-point an inherited credential at a new endpoint. - _strip_credentials_on_destination_change( - merged_dictionary, - _new_litellm_params_dict, - lambda field: _decrypted_param( - _existing_litellm_params_dict.get(field) - ), - ) + if not is_proxy_admin(user_api_key_dict): + _assert_credential_resupplied_on_destination_change( + _new_litellm_params_dict, + lambda field: _decrypted_param( + _existing_litellm_params_dict.get(field) + ), + ) _data: dict = { "litellm_params": json.dumps(merged_dictionary), # type: ignore diff --git a/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py index 867002d6192..68ab1e969d4 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py @@ -1956,81 +1956,50 @@ class TestModelMgmtAuthzHardening: info = json.loads(result["model_info"]) assert info["id"] == "real-id" - def test_api_base_change_clears_inherited_credential(self): - import litellm + @staticmethod + def _assert_resupply(patch_plaintext, db): from litellm.proxy.management_endpoints.model_management_endpoints import ( - update_db_model, + _assert_credential_resupplied_on_destination_change, ) - from litellm.types.router import updateLiteLLMParams - with ( - patch.object(litellm, "user_url_validation", False), - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", - side_effect=lambda value, **kwargs: value, - ), - ): - result = update_db_model( - db_model=self._db_model_with_secret(), - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams( - api_base="https://attacker.example.com" - ) - ), - ) - params = json.loads(result["litellm_params"]) - assert "api_key" not in params # stored secret must not ride to new base - - def test_api_base_change_with_new_key_keeps_credential(self): - import litellm - from litellm.proxy.management_endpoints.model_management_endpoints import ( - update_db_model, + _assert_credential_resupplied_on_destination_change( + patch_plaintext, lambda field: db.get(field) ) - from litellm.types.router import updateLiteLLMParams - with ( - patch.object(litellm, "user_url_validation", False), - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", - side_effect=lambda value, **kwargs: value, - ), - ): - result = update_db_model( - db_model=self._db_model_with_secret(), - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams( - api_base="https://attacker.example.com", api_key="sk-new" - ) - ), + def test_destination_change_without_resupply_is_rejected(self): + from litellm.proxy._types import ProxyException + + db = {"api_base": "https://real.example.com", "api_key": "sk-stored"} + # Redirecting api_base without re-supplying the stored key must be + # rejected; clearing it would fall back to the proxy env key instead. + with pytest.raises(ProxyException) as e: + self._assert_resupply({"api_base": "https://attacker.example.com"}, db) + assert str(e.value.code) == "400" + + def test_empty_credential_resupply_is_rejected(self): + from litellm.proxy._types import ProxyException + + db = {"api_base": "https://real.example.com", "api_key": "sk-stored"} + # An empty key is not a fresh credential: it resolves to OPENAI_API_KEY. + with pytest.raises(ProxyException): + self._assert_resupply( + {"api_base": "https://attacker.example.com", "api_key": ""}, db ) - params = json.loads(result["litellm_params"]) - assert "api_key" in params # caller re-supplied a key, so it is kept - def test_unchanged_destination_keeps_credential(self): - import litellm - from litellm.proxy.management_endpoints.model_management_endpoints import ( - update_db_model, + def test_destination_change_with_resupply_is_allowed(self): + db = {"api_base": "https://real.example.com", "api_key": "sk-stored"} + self._assert_resupply( + {"api_base": "https://attacker.example.com", "api_key": "sk-new"}, db ) - from litellm.types.router import updateLiteLLMParams - # Re-saving the same api_base (e.g. a UI resend) must NOT clear the key. - with ( - patch.object(litellm, "user_url_validation", False), - patch( - "litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper", - side_effect=lambda value, **kwargs: value, - ), - ): - result = update_db_model( - db_model=self._db_model_with_secret(), - updated_patch=updateDeployment( - litellm_params=updateLiteLLMParams( - api_base="https://real.example.com" - ) - ), - ) - params = json.loads(result["litellm_params"]) - assert "api_key" in params + def test_unchanged_destination_is_allowed(self): + db = {"api_base": "https://real.example.com", "api_key": "sk-stored"} + self._assert_resupply({"api_base": "https://real.example.com"}, db) + + def test_keyless_model_destination_change_is_allowed(self): + # A model that never carried a credential has nothing to leak. + db = {"api_base": "https://real.example.com"} + self._assert_resupply({"api_base": "https://attacker.example.com"}, db) def test_validate_model_url_params_blocks_internal_ip(self): import litellm @@ -2481,25 +2450,19 @@ class TestModelMgmtAuthzHardening: user_api_key_dict=self._admin(), ) - def test_destination_change_clears_inherited_credential_per_field(self): - from litellm.proxy.management_endpoints.model_management_endpoints import ( - _strip_credentials_on_destination_change, - ) + def test_destination_change_partial_resupply_is_rejected(self): + from litellm.proxy._types import ProxyException # Patch redirects api_base AND supplies an UNRELATED credential - # (aws_secret_access_key); the inherited api_key must still be dropped. - merged = { - "api_key": "sk-inherited", - "aws_secret_access_key": "supplied", - "api_base": "https://attacker.example", - } - patch_plaintext = { - "api_base": "https://attacker.example", - "aws_secret_access_key": "supplied", - } + # (aws_secret_access_key); the inherited api_key is still not re-supplied, + # so the change must be rejected rather than letting api_key ride along. db = {"api_base": "https://real.example", "api_key": "sk-inherited"} - _strip_credentials_on_destination_change( - merged, patch_plaintext, lambda f: db.get(f) - ) - assert "api_key" not in merged # inherited, not re-supplied -> dropped - assert merged.get("aws_secret_access_key") == "supplied" # supplied -> kept + with pytest.raises(ProxyException) as e: + self._assert_resupply( + { + "api_base": "https://attacker.example", + "aws_secret_access_key": "supplied", + }, + db, + ) + assert e.value.param == "api_key"