mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-30 06:43:27 +00:00
fix(credentials): keep the stored secret when PATCH echoes its masked read-back (#48950)
* test(credentials): pin that PATCH keeps the stored secret when it echoes the masked read-back * fix(credentials): keep the stored secret when PATCH echoes its masked read-back * fix(credentials): only treat a mask on a key GET masks as a placeholder PATCH now skips mask-shaped values only on the keys get_masked_values masks, using the masker's own keyword check, so a value like api_base ab****cd is saved as sent again. Also drops the new test docstrings and tightens the new test helper types --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
tessellark
GitHub
parent
a66dc50efb
commit
b7da9a2755
3 changed files with 353 additions and 30 deletions
|
|
@ -5052,6 +5052,23 @@ class Logging(LiteLLMLoggingBaseClass):
|
|||
return result_copy
|
||||
|
||||
|
||||
_SENSITIVE_KEYWORDS: Final = (
|
||||
"authorization",
|
||||
"token",
|
||||
"key",
|
||||
"secret",
|
||||
"vertex_credentials",
|
||||
"credentials",
|
||||
"password",
|
||||
"passwd",
|
||||
)
|
||||
|
||||
|
||||
def has_sensitive_keyword(key: str) -> bool:
|
||||
lowered: Final = key.lower()
|
||||
return any(keyword in lowered for keyword in _SENSITIVE_KEYWORDS)
|
||||
|
||||
|
||||
def get_masked_values(
|
||||
sensitive_object: Mapping[str, object],
|
||||
ignore_sensitive_values: bool = False,
|
||||
|
|
@ -5070,16 +5087,6 @@ def get_masked_values(
|
|||
masked_length: Optional length for the masked portion (number of *). If set, will use exactly this many *
|
||||
regardless of original string length. The total length will be unmasked_length + masked_length.
|
||||
"""
|
||||
sensitive_keywords: Final = (
|
||||
"authorization",
|
||||
"token",
|
||||
"key",
|
||||
"secret",
|
||||
"vertex_credentials",
|
||||
"credentials",
|
||||
"password",
|
||||
"passwd",
|
||||
)
|
||||
|
||||
def _mask_value(v: object) -> object:
|
||||
if isinstance(v, dict):
|
||||
|
|
@ -5106,12 +5113,7 @@ def get_masked_values(
|
|||
return v[: unmasked_length // 2] + "*" * (len(v) - unmasked_length) + v[-unmasked_length // 2 :]
|
||||
|
||||
return {
|
||||
k: (
|
||||
v
|
||||
if ignore_sensitive_values
|
||||
or not any(sensitive_keyword in k.lower() for sensitive_keyword in sensitive_keywords)
|
||||
else _mask_value(v)
|
||||
)
|
||||
k: (v if ignore_sensitive_values or not has_sensitive_keyword(k) else _mask_value(v))
|
||||
for k, v in sensitive_object.items()
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ CRUD endpoints for storing reusable credentials.
|
|||
import time
|
||||
import unicodedata
|
||||
from collections.abc import Mapping
|
||||
from types import MappingProxyType
|
||||
from typing import (
|
||||
Annotated,
|
||||
Final,
|
||||
|
|
@ -23,7 +24,7 @@ from litellm.constants import (
|
|||
GITHUB_COPILOT_PER_USER_AUTH_TYPE,
|
||||
)
|
||||
from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
|
||||
from litellm.litellm_core_utils.litellm_logging import get_masked_values
|
||||
from litellm.litellm_core_utils.litellm_logging import get_masked_values, has_sensitive_keyword
|
||||
from litellm.llms.anthropic.wif import (
|
||||
ExportedJwks,
|
||||
NotAnInternalIssuerCredential,
|
||||
|
|
@ -83,6 +84,11 @@ _CREDENTIAL_DICT_ADAPTER: Final = TypeAdapter(dict[str, object])
|
|||
_GITHUB_COPILOT_PROVIDER: Final = "github_copilot"
|
||||
_DISPLAY_NAME_MAX_LENGTH: Final = 255
|
||||
_DISPLAY_NAME_FORBIDDEN_CATEGORIES: Final = frozenset({"Cc", "Cf", "Cs", "Zl", "Zp"})
|
||||
_MASK_UNMASKED_LENGTH: Final = 4
|
||||
_MASK_ASTERISKS: Final = 4
|
||||
_SHORT_SECRET_MASK: Final = "*****"
|
||||
_MASKED_READ_BACK_DEPTH: Final = 20
|
||||
_NO_FIELDS: Final[Mapping[str, object]] = MappingProxyType({})
|
||||
|
||||
|
||||
def _normalized_github_copilot_values(
|
||||
|
|
@ -395,7 +401,14 @@ async def get_credentials(
|
|||
"""
|
||||
try:
|
||||
masked_credentials: Final = [
|
||||
_credential_view(credential, get_masked_values(credential.credential_values))
|
||||
_credential_view(
|
||||
credential,
|
||||
get_masked_values(
|
||||
credential.credential_values,
|
||||
unmasked_length=_MASK_UNMASKED_LENGTH,
|
||||
number_of_asterisks=_MASK_ASTERISKS,
|
||||
),
|
||||
)
|
||||
for credential in litellm.credential_list
|
||||
]
|
||||
return {"success": True, "credentials": masked_credentials}
|
||||
|
|
@ -423,7 +436,11 @@ async def get_credential_by_name(
|
|||
if credential.credential_name == credential_name:
|
||||
return _credential_view(
|
||||
credential,
|
||||
get_masked_values(credential.credential_values, unmasked_length=4, number_of_asterisks=4),
|
||||
get_masked_values(
|
||||
credential.credential_values,
|
||||
unmasked_length=_MASK_UNMASKED_LENGTH,
|
||||
number_of_asterisks=_MASK_ASTERISKS,
|
||||
),
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
|
|
@ -521,8 +538,8 @@ async def get_credential_by_model(
|
|||
raise HTTPException(status_code=404, detail="Model not found")
|
||||
masked_credential_values: Final = get_masked_values(
|
||||
credential_values,
|
||||
unmasked_length=4,
|
||||
number_of_asterisks=4,
|
||||
unmasked_length=_MASK_UNMASKED_LENGTH,
|
||||
number_of_asterisks=_MASK_ASTERISKS,
|
||||
)
|
||||
credential: Final = CredentialItem(
|
||||
credential_name=f"{model.model_name}-credential-{model_id}",
|
||||
|
|
@ -908,6 +925,48 @@ def update_db_credential(
|
|||
return merged_credential
|
||||
|
||||
|
||||
def _is_masked_read_back(value: object) -> bool:
|
||||
if not isinstance(value, str):
|
||||
return False
|
||||
if value == _SHORT_SECRET_MASK:
|
||||
return True
|
||||
shown: Final = _MASK_UNMASKED_LENGTH // 2
|
||||
return len(value) == _MASK_UNMASKED_LENGTH + _MASK_ASTERISKS and value[shown:-shown] == "*" * _MASK_ASTERISKS
|
||||
|
||||
|
||||
def _as_fields(value: object) -> Mapping[str, object] | None:
|
||||
try:
|
||||
return _CREDENTIAL_DICT_ADAPTER.validate_python(value, strict=True)
|
||||
except ValidationError:
|
||||
return None
|
||||
|
||||
|
||||
def _kept_value(key: str, value: object, stored: Mapping[str, object], depth: int) -> tuple[str, object] | None:
|
||||
if not has_sensitive_keyword(key):
|
||||
return key, value
|
||||
fields: Final = _as_fields(value)
|
||||
if fields is not None and depth < _MASKED_READ_BACK_DEPTH:
|
||||
return key, _without_masked_read_backs(fields, _as_fields(stored.get(key)) or _NO_FIELDS, depth + 1)
|
||||
if not _is_masked_read_back(value):
|
||||
return key, value
|
||||
if key in stored:
|
||||
return key, stored[key]
|
||||
return None
|
||||
|
||||
|
||||
def _without_masked_read_backs(
|
||||
requested: Mapping[str, object], stored: Mapping[str, object], depth: int = 0
|
||||
) -> Mapping[str, object]:
|
||||
kept: Final = (_kept_value(key, value, stored, depth) for key, value in requested.items())
|
||||
return _CREDENTIAL_DICT_ADAPTER.validate_python(
|
||||
MappingProxyType(dict(entry for entry in kept if entry is not None))
|
||||
)
|
||||
|
||||
|
||||
def _stored_objects(stored_values: Mapping[str, object]) -> Mapping[str, object]:
|
||||
return MappingProxyType({key: value for key, value in stored_values.items() if _as_fields(value) is not None})
|
||||
|
||||
|
||||
@router.patch(
|
||||
"/credentials/{credential_name:path}",
|
||||
dependencies=[Depends(user_api_key_auth)],
|
||||
|
|
@ -936,12 +995,12 @@ async def update_credential(
|
|||
)
|
||||
requested_display_name: Final = _normalized_display_name(credential.display_name)
|
||||
_reject_overlapping_credential_values(credential)
|
||||
incoming_values: Final = _CREDENTIAL_DICT_ADAPTER.validate_python(
|
||||
requested_values: Final = _CREDENTIAL_DICT_ADAPTER.validate_python(
|
||||
_resolve_deployment_credentials(llm_router, credential.model_id)
|
||||
if credential.model_id
|
||||
else credential.credential_values or {}
|
||||
)
|
||||
_reject_non_admin_wif_fields(_incoming_wif_fields(incoming_values, credential), user_api_key_dict)
|
||||
_reject_non_admin_wif_fields(_incoming_wif_fields(requested_values, credential), user_api_key_dict)
|
||||
if prisma_client is None:
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
|
|
@ -952,19 +1011,20 @@ async def update_credential(
|
|||
if db_credential is None:
|
||||
raise _not_found_unless_config_defined(credential_name, "Credential not found in DB.")
|
||||
_reject_non_admin_wif_fields(_stored_wif_fields(db_credential), user_api_key_dict)
|
||||
decrypted_stored_credential: Final = decrypted_credential(db_credential)
|
||||
decrypted_values: Final = _CREDENTIAL_DICT_ADAPTER.validate_python(
|
||||
decrypted_stored_credential.credential_values
|
||||
)
|
||||
incoming_values: Final = _without_masked_read_backs(requested_values, _stored_objects(decrypted_values))
|
||||
patch: Final = CredentialItem(
|
||||
credential_name=credential_name,
|
||||
display_name=(
|
||||
requested_display_name if "display_name" in credential.model_fields_set else db_credential.display_name
|
||||
),
|
||||
credential_info=_CREDENTIAL_DICT_ADAPTER.validate_python(credential.credential_info),
|
||||
credential_values=incoming_values,
|
||||
credential_values=_CREDENTIAL_DICT_ADAPTER.validate_python(incoming_values),
|
||||
credential_values_to_delete=credential.credential_values_to_delete,
|
||||
)
|
||||
decrypted_stored_credential: Final = decrypted_credential(db_credential)
|
||||
decrypted_values: Final = _CREDENTIAL_DICT_ADAPTER.validate_python(
|
||||
decrypted_stored_credential.credential_values
|
||||
)
|
||||
stored_info: Final = _CREDENTIAL_DICT_ADAPTER.validate_python(
|
||||
db_credential.credential_info # pyright: ignore[reportUnknownMemberType] # untyped credential field
|
||||
)
|
||||
|
|
|
|||
|
|
@ -2,8 +2,9 @@
|
|||
|
||||
import json
|
||||
from contextlib import contextmanager
|
||||
from functools import reduce
|
||||
from types import SimpleNamespace
|
||||
from typing import Final
|
||||
from typing import Final, Protocol
|
||||
from unittest.mock import AsyncMock, MagicMock, call, patch
|
||||
|
||||
import httpx
|
||||
|
|
@ -12,14 +13,16 @@ from cryptography.hazmat.primitives import serialization
|
|||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from fastapi import HTTPException, Request
|
||||
from fastapi.testclient import TestClient
|
||||
from pydantic import TypeAdapter
|
||||
|
||||
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 +463,264 @@ def test_update_credential_never_syncs_a_null_into_the_in_memory_credential(rest
|
|||
assert synced.credential_values["api_key"] == "sk-rotated"
|
||||
|
||||
|
||||
_WRITTEN_JSON: Final = TypeAdapter(dict[str, object])
|
||||
|
||||
|
||||
class _InstallCredentialStore(Protocol):
|
||||
def __call__(
|
||||
self,
|
||||
*,
|
||||
connected: bool = True,
|
||||
in_memory: tuple[object, ...] = (),
|
||||
llm_router: object | None = None,
|
||||
**repository_calls: AsyncMock,
|
||||
) -> None: ...
|
||||
|
||||
|
||||
def _encrypted_row(served: CredentialItem) -> CredentialItem:
|
||||
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[str, object]:
|
||||
written: Final = _WRITTEN_JSON.validate_json(update_by_name.await_args.kwargs["data"]["credential_values"])
|
||||
return {
|
||||
key: decrypt_value_helper(value, key) if isinstance(value, str) else value for key, value in written.items()
|
||||
}
|
||||
|
||||
|
||||
def _written_info(update_by_name: AsyncMock) -> dict[str, object]:
|
||||
return _WRITTEN_JSON.validate_json(update_by_name.await_args.kwargs["data"]["credential_info"])
|
||||
|
||||
|
||||
def _serve_credential(
|
||||
credential_store: _InstallCredentialStore,
|
||||
served: CredentialItem,
|
||||
in_memory: tuple[CredentialItem, ...] | None = None,
|
||||
) -> AsyncMock:
|
||||
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[str, dict[str, str]]:
|
||||
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: _InstallCredentialStore,
|
||||
):
|
||||
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: _InstallCredentialStore,
|
||||
):
|
||||
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: _InstallCredentialStore,
|
||||
):
|
||||
served = CredentialItem(credential_name="existing", credential_values={"api_key": "abc"}, credential_info={})
|
||||
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"] != "abc"
|
||||
|
||||
response = _patch_credential(
|
||||
"existing", {"credential_values": read_back["credential_values"], "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: _InstallCredentialStore, rotated_values: dict[str, object]
|
||||
):
|
||||
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}
|
||||
|
||||
|
||||
def test_update_credential_saves_a_mask_shaped_value_on_a_key_get_never_masks(
|
||||
credential_store: _InstallCredentialStore,
|
||||
):
|
||||
served = CredentialItem(
|
||||
credential_name="existing",
|
||||
credential_values={
|
||||
"api_key": "sk-real-secret-1234",
|
||||
"api_base": "https://api.example.com",
|
||||
"extra_headers": {"x-team": "platform"},
|
||||
},
|
||||
credential_info={},
|
||||
)
|
||||
update_by_name = _serve_credential(credential_store, served)
|
||||
sent: Final = {"api_base": "ab****cd", "organization": "*****", "extra_headers": {"x-team": "*****"}}
|
||||
assert get_masked_values(sent, unmasked_length=4, number_of_asterisks=4) == sent
|
||||
|
||||
response = _patch_credential("existing", {"credential_values": sent, "credential_info": {}})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert _written_values(update_by_name) == {"api_key": "sk-real-secret-1234", **sent}
|
||||
|
||||
|
||||
@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: _InstallCredentialStore, client_email: str
|
||||
):
|
||||
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: _InstallCredentialStore,
|
||||
):
|
||||
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: _InstallCredentialStore, in_memory: tuple[CredentialItem, ...]
|
||||
):
|
||||
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: _InstallCredentialStore, levels: int
|
||||
):
|
||||
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(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue