scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 200

test(credentials): pin that PATCH keeps the stored secret when it echoes the masked read-back

Browse source
This commit is contained in:
mateo-berri 2026-10-20 21:09:29 -07:00
parent 4ecc5fd13a
commit c61c6b6c19
1 changed files with 226 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

227
tests/unit/proxy/credential_endpoints/test_endpoints.py
View file

@ -2,6 +2,7 @@
import json
from contextlib import contextmanager
from functools import reduce
from types import SimpleNamespace
from typing import Final
from unittest.mock import AsyncMock, MagicMock, call, patch
@ -15,11 +16,12 @@ from fastapi.testclient import TestClient
import litellm
from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
from litellm.litellm_core_utils.litellm_logging import get_masked_values
from litellm.llms.github_copilot.per_user_auth import _SESSION_CACHE, _session_cache_key
from litellm.models.credentials import CredentialSource, UserConnectionFlowHandle
from litellm.proxy._types import LiteLLM_UserTable, LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.encrypt_decrypt_utils import encrypt_value_helper
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value_helper, encrypt_value_helper
from litellm.proxy.credential_endpoints.endpoints import _decode_flow_handle, get_llm_router
from litellm.proxy.proxy_server import app
from litellm.types.utils import CredentialItem
@ -460,6 +462,229 @@ def test_update_credential_never_syncs_a_null_into_the_in_memory_credential(rest
assert synced.credential_values["api_key"] == "sk-rotated"
def _encrypted_row(served: CredentialItem) -> CredentialItem:
"""The table row behind a credential the proxy serves: every string value encrypted at rest."""
return CredentialItem(
credential_name=served.credential_name,
credential_values={key: encrypt_value_helper(value) for key, value in served.credential_values.items()},
credential_info=dict(served.credential_info),
)
def _written_values(update_by_name: AsyncMock) -> dict:
written = json.loads(update_by_name.await_args.kwargs["data"]["credential_values"])
return {key: decrypt_value_helper(value, key) for key, value in written.items()}
def _written_info(update_by_name: AsyncMock) -> dict:
return json.loads(update_by_name.await_args.kwargs["data"]["credential_info"])
def _serve_credential(
credential_store, served: CredentialItem, in_memory: tuple[CredentialItem, ...] | None = None
) -> AsyncMock:
"""Serves ``served`` from the table; memory holds the same credential unless ``in_memory`` says
otherwise (a replica that has not resynced yet, or one that never loaded the credential)."""
update_by_name = AsyncMock(return_value=None)
credential_store(
in_memory=(served,) if in_memory is None else in_memory,
find_by_name=AsyncMock(side_effect=lambda _name: _encrypted_row(served)),
update_by_name=update_by_name,
)
return update_by_name
def _service_account(private_key: str = "pk-1234567890", client_email: str = "svc@example.com") -> dict:
return {"vertex_credentials": {"private_key": private_key, "client_email": client_email}}
def test_update_credential_keeps_the_stored_secret_when_the_patch_echoes_its_masked_read_back(credential_store):
"""Regression for #28906: ``GET /credentials/by_name/{name}`` renders ``sk-real-secret-1234`` as
``sk****34``, and a caller editing an unrelated field sends that rendering back. The handler
stored the placeholder as the secret, so every later provider call failed with 401 until the
key was re-entered."""
served = CredentialItem(
credential_name="existing",
credential_values={"api_key": "sk-real-secret-1234"},
credential_info={"custom_llm_provider": "openai", "description": "original"},
)
update_by_name = _serve_credential(credential_store, served)
response = _patch_credential(
"existing",
{
"credential_values": {"api_key": "sk****34"},
"credential_info": {"custom_llm_provider": "openai", "description": "edited"},
},
)
assert response.status_code == 200, response.text
assert _written_values(update_by_name) == {"api_key": "sk-real-secret-1234"}
assert _written_info(update_by_name)["description"] == "edited"
assert CredentialAccessor.get_credential_values("existing") == {"api_key": "sk-real-secret-1234"}
assert litellm.credential_list[0].credential_info["description"] == "edited"
def test_update_credential_keeps_the_secret_when_the_patch_echoes_what_get_by_name_served(credential_store):
"""The #28906 flow through both endpoints: whatever ``GET /credentials/by_name`` renders today,
sending it back with one field edited leaves the secret alone, so the placeholder check stays
tied to the masker's real output rather than to a rendering spelled out by hand."""
served = CredentialItem(
credential_name="existing",
credential_values={"api_key": "sk-real-secret-1234", "api_base": "https://api.example.com"},
credential_info={"custom_llm_provider": "openai", "description": "original"},
)
update_by_name = _serve_credential(credential_store, served)
read_back = _call_as("GET", "/credentials/by_name/existing").json()
assert read_back["credential_values"]["api_key"] != "sk-real-secret-1234"
response = _patch_credential(
"existing", {**read_back, "credential_info": {**read_back["credential_info"], "description": "edited"}}
)
assert response.status_code == 200, response.text
assert _written_values(update_by_name) == {"api_key": "sk-real-secret-1234", "api_base": "https://api.example.com"}
assert _written_info(update_by_name)["description"] == "edited"
def test_update_credential_keeps_a_short_secret_whose_read_back_is_all_asterisks(credential_store):
"""Secrets of four characters or fewer render as ``*****`` with no prefix or suffix, so the
placeholder check has to match that rendering too, not only the ``ab****yz`` shape."""
served = CredentialItem(credential_name="existing", credential_values={"api_key": "abc"}, credential_info={})
update_by_name = _serve_credential(credential_store, served)
response = _patch_credential("existing", {"credential_values": {"api_key": "*****"}, "credential_info": {}})
assert response.status_code == 200, response.text
assert _written_values(update_by_name) == {"api_key": "abc"}
assert CredentialAccessor.get_credential_values("existing") == {"api_key": "abc"}
@pytest.mark.parametrize(
"rotated_values",
(
{"api_key": "sk-rotated-key-5678"},
{"api_key": "new**secret"},
_service_account(private_key="pk**rotated"),
),
)
def test_update_credential_still_rotates_a_secret_sent_in_full(credential_store, rotated_values):
"""Dropping every sensitive field would also pass the regressions above. A real new value must
still replace the stored one in the table and in memory, even when it carries asterisks of its
own, since only the exact rendering GET serves is a placeholder."""
served = CredentialItem(
credential_name="existing",
credential_values={"api_key": "sk-old-key-000011", **_service_account()},
credential_info={},
)
update_by_name = _serve_credential(credential_store, served)
response = _patch_credential("existing", {"credential_values": rotated_values, "credential_info": {}})
assert response.status_code == 200, response.text
assert _written_values(update_by_name) == {**served.credential_values, **rotated_values}
assert CredentialAccessor.get_credential_values("existing") == {**served.credential_values, **rotated_values}
@pytest.mark.parametrize("client_email", ("svc@example.com", "new@example.com"))
def test_update_credential_keeps_a_masked_secret_inside_a_stored_object(credential_store, client_email):
"""Service-account style credentials are stored as an object whose sensitive fields are masked
one level down. The object sent back replaces the stored one whole, so a masked leaf inside it
takes the stored leaf while a sibling field edited beside it still updates."""
served = CredentialItem(credential_name="existing", credential_values=_service_account(), credential_info={})
update_by_name = _serve_credential(credential_store, served)
response = _patch_credential(
"existing",
{
"credential_values": _service_account(private_key="pk****90", client_email=client_email),
"credential_info": {},
},
)
assert response.status_code == 200, response.text
assert _written_values(update_by_name) == _service_account(client_email=client_email)
assert CredentialAccessor.get_credential_values("existing") == _service_account(client_email=client_email)
def test_update_credential_drops_a_masked_leaf_the_stored_object_no_longer_has(credential_store):
"""A read-back taken before a field was removed from a service-account object still carries
that field masked. There is no stored leaf to put back, so the placeholder is dropped rather
than written."""
served = CredentialItem(credential_name="existing", credential_values=_service_account(), credential_info={})
update_by_name = _serve_credential(credential_store, served)
response = _patch_credential(
"existing",
{
"credential_values": {
"vertex_credentials": {
"private_key": "pk****90",
"refresh_token": "rt****89",
"client_email": "svc@example.com",
}
},
"credential_info": {},
},
)
assert response.status_code == 200, response.text
assert _written_values(update_by_name) == _service_account()
@pytest.mark.parametrize(
"in_memory",
(
pytest.param(
(
CredentialItem(
credential_name="existing",
credential_values={"api_key": "sk-old-key-000011"},
credential_info={"description": "original"},
),
),
id="replica_not_resynced",
),
pytest.param((), id="replica_never_loaded_it"),
),
)
def test_update_credential_keeps_the_table_secret_whatever_this_replica_last_rendered(credential_store, in_memory):
"""Reads are served from memory, which lags the table after another replica rotates the key, and
a script may cache a read-back from before the rotation. An echo of that older rendering is still
a placeholder, and the check is judged on the table row, so the row keeps the rotated key."""
served = CredentialItem(
credential_name="existing",
credential_values={"api_key": "sk-rotated-key-000099"},
credential_info={"description": "original"},
)
update_by_name = _serve_credential(credential_store, served, in_memory=in_memory)
response = _patch_credential(
"existing",
{"credential_values": {"api_key": "sk****11"}, "credential_info": {"description": "edited"}},
)
assert response.status_code == 200, response.text
assert _written_values(update_by_name) == {"api_key": "sk-rotated-key-000099"}
assert _written_info(update_by_name)["description"] == "edited"
@pytest.mark.parametrize("levels", (20, 21))
def test_update_credential_keeps_a_secret_exactly_as_deep_as_the_masker_masks(credential_store, levels):
"""The masker stops masking at a fixed depth, and the placeholder check has to stop at the same
one: at the last masked level the echo is a placeholder that must take the stored leaf, and one
level past it the echo is the secret itself. Either way the write must hold the real secret."""
nested = reduce(lambda inner, _level: {"credentials": inner}, range(levels), {"api_key": "leaf-secret-1234"})
served = CredentialItem(credential_name="existing", credential_values=nested, credential_info={})
update_by_name = _serve_credential(credential_store, served)
read_back = get_masked_values(served.credential_values, unmasked_length=4, number_of_asterisks=4)
response = _patch_credential("existing", {"credential_values": read_back, "credential_info": {}})
assert response.status_code == 200, response.text
assert _written_values(update_by_name) == nested
def _generate_es256_pem() -> str:
key = ec.generate_private_key(ec.SECP256R1())
return key.private_bytes(