fix(credentials): gate PATCH on WIF fields resolved from model_id

The credential PATCH handler checked server-owned workload identity
federation fields only on the values the caller sent, while a body that
named a deployment through model_id had its credential values resolved
after that check. A non-admin could therefore copy a federated
deployment's WIF fields onto an ordinary credential. Resolve the incoming
values first and run the non-admin gate on them, matching the POST path
This commit is contained in:
mateo-berri 2026-10-03 15:47:50 -07:00
parent f43151a97d
commit df155d7911
2 changed files with 53 additions and 10 deletions

View file

@ -69,13 +69,14 @@ def _reject_non_admin_wif_fields(
)
def _incoming_wif_fields(credential: UpdateCredentialItem) -> tuple[str, ...]:
"""WIF fields the request payload itself touches: the ones it sets (to any value, ``None``
included, since the key alone is what the federation resolver reacts to), plus the ones it
def _incoming_wif_fields(incoming_values: Mapping[str, object], credential: UpdateCredentialItem) -> tuple[str, ...]:
"""WIF fields the request touches: the ones its values set (to any value, ``None`` included,
since the key alone is what the federation resolver reacts to), whether the caller sent them
or named a deployment through ``model_id`` for the proxy to copy them from, plus the ones it
names in ``credential_values_to_delete``, since dropping a federation field off the stored
credential breaks every deployment referencing it just as installing one would redirect them.
"""
return server_owned_wif_fields_named(credential.credential_values or ()) + server_owned_wif_fields_named(
return server_owned_wif_fields_named(incoming_values) + server_owned_wif_fields_named(
credential.credential_values_to_delete or ()
)
@ -515,7 +516,12 @@ async def update_credential(
try:
_reject_overlapping_credential_values(credential)
_reject_non_admin_wif_fields(_incoming_wif_fields(credential), user_api_key_dict)
incoming_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)
if prisma_client is None:
raise HTTPException(
status_code=500,
@ -533,11 +539,7 @@ async def update_credential(
patch: Final = CredentialItem(
credential_name=credential.credential_name,
credential_info=_CREDENTIAL_DICT_ADAPTER.validate_python(credential.credential_info),
credential_values=_CREDENTIAL_DICT_ADAPTER.validate_python(
_resolve_deployment_credentials(llm_router, credential.model_id)
if credential.model_id
else credential.credential_values or {}
),
credential_values=incoming_values,
credential_values_to_delete=credential.credential_values_to_delete,
)
merged_credential: Final = update_db_credential(db_credential, patch)

View file

@ -695,6 +695,47 @@ class TestNonAdminCannotPersistWifFieldsOnCredential:
assert response.status_code == 403, response.text
update_mock.assert_not_awaited()
def test_non_admin_cannot_patch_wif_fields_onto_a_credential_through_model_id(self, credential_store):
"""Regression: the PATCH gate read only the submitted ``credential_values``, so a non-admin
naming a federated deployment through ``model_id`` had its WIF fields copied onto an
ordinary credential unchecked, while POST already gated the resolved values."""
stored = CredentialItem(credential_name="existing", credential_values={"api_key": "sk-old"}, credential_info={})
update_by_name = AsyncMock(return_value=None)
router = MagicMock()
router.get_deployment.return_value = {"model_name": "claude-opus-5-5"}
router.get_deployment_credentials.return_value = {
"anthropic_keycloak_token_url": "https://keycloak.internal/token",
"anthropic_keycloak_client_secret_ref": "os.environ/KEYCLOAK_CLIENT_SECRET",
}
credential_store(find_by_name=AsyncMock(return_value=stored), update_by_name=update_by_name, llm_router=router)
response = _patch_credential(
"existing",
{"credential_name": "existing", "model_id": "federated-deployment", "credential_info": {}},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
update_by_name.assert_not_awaited()
def test_proxy_admin_can_patch_wif_fields_onto_a_credential_through_model_id(self, credential_store):
stored = CredentialItem(credential_name="existing", credential_values={"api_key": "sk-old"}, credential_info={})
update_by_name = AsyncMock(return_value=None)
router = MagicMock()
router.get_deployment.return_value = {"model_name": "claude-opus-5-5"}
router.get_deployment_credentials.return_value = {"anthropic_keycloak_token_url": "https://keycloak.internal/token"}
credential_store(find_by_name=AsyncMock(return_value=stored), update_by_name=update_by_name, llm_router=router)
response = _patch_credential(
"existing",
{"credential_name": "existing", "model_id": "federated-deployment", "credential_info": {}},
auth=_as_admin,
)
assert response.status_code == 200, response.text
written = json.loads(update_by_name.await_args.kwargs["data"]["credential_values"])
assert "anthropic_keycloak_token_url" in written, "the deployment's WIF field reaches the stored credential"
def test_proxy_admin_can_update_a_credential_to_add_a_wif_destination(self):
stored = CredentialItem(
credential_name="existing",