fix(credentials): reject unknown access keys and fail closed on malformed stored access

Addresses two linked Greptile P1 findings. validate_credential_access now
rejects access keys outside {global,teams,orgs} at write time, matching the
strict CredentialAccess read model, so a destination can never be stored in a
shape the model later refuses to parse. _authorize_credential_patch guards the
stored-credential parse and fails closed to proxy-admin-only on a ValidationError
instead of letting a malformed legacy row 500 every PATCH.
This commit is contained in:
yucheng-berriai 2026-06-29 12:47:54 -07:00
parent e8020da01e
commit 190190cce6
4 changed files with 68 additions and 1 deletions

View file

@ -521,7 +521,15 @@ async def _authorize_credential_patch(
except ValidationError as ve:
raise HTTPException(status_code=400, detail={"error": _summarize_validation_error(ve)})
assert existing is not None # narrowed by existing_is_logging_gated
existing_info_typed = CredentialInfo.model_validate(existing.credential_info)
try:
existing_info_typed = CredentialInfo.model_validate(existing.credential_info)
except ValidationError:
# Stored info the strict access model can't parse (e.g. a legacy access key)
# can't be run through the field-level decider. Fail closed: only the proxy
# admin may patch such a row, instead of 500-ing every caller.
if not is_admin:
raise HTTPException(status_code=403, detail={"error": OPAQUE_DENY_REASON})
return
decision = decide_credential_patch(
is_proxy_admin=is_admin,
caller_team_admin_ids=team_admin_ids,

View file

@ -63,6 +63,12 @@ def validate_credential_access(credential_info: Optional[dict]) -> None:
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": f"access.{field} must be a list of strings"},
)
unknown = set(access) - {"global", "teams", "orgs"}
if unknown:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": f"access contains unknown field(s): {sorted(unknown)}"},
)
def _logging_credentials_by_name() -> dict[str, dict]:

View file

@ -762,3 +762,46 @@ async def test_get_credentials_returns_all_for_proxy_admin(monkeypatch):
)
names = sorted(c["credential_name"] for c in response["credentials"])
assert names == ["openai", "poc-langfuse"]
@pytest.mark.asyncio
async def test_authorize_patch_malformed_stored_access_does_not_500(
_patch_team_admin_lookup,
):
"""A stored logging destination whose ``access`` carries a key the strict model
forbids (e.g. legacy data, or data written before the write gate rejected unknown
keys) must not 500 every PATCH. Fail closed: a non-admin gets 403, the proxy admin
is still allowed through. Pre-fix the unguarded ``CredentialInfo.model_validate``
on the stored row raised an uncaught ``ValidationError`` for all callers."""
malformed = CredentialItem(
credential_name="legacy-dest",
credential_values={"langfuse_host": "h"},
credential_info={
"credential_type": "logging",
"access": {"global": True, "legacy_field": "x"},
},
)
patch = UpdateCredentialItem(credential_info={"access": {"teams": ["team-T"]}})
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
with pytest.raises(HTTPException) as exc:
await endpoints._authorize_credential_patch(
credential_name="legacy-dest",
patch=patch,
existing=malformed,
user_api_key_dict=_team_admin_of(["team-T"]),
prisma_client=MagicMock(),
)
assert exc.value.status_code == 403
assert exc.value.detail == {"error": OPAQUE_DENY_REASON}
assert (
await endpoints._authorize_credential_patch(
credential_name="legacy-dest",
patch=patch,
existing=malformed,
user_api_key_dict=_admin(),
prisma_client=MagicMock(),
)
is None
)

View file

@ -383,3 +383,13 @@ def test_validate_credential_access_rejects_bad_shape(access):
with pytest.raises(HTTPException) as exc:
validate_credential_access({"access": access})
assert exc.value.status_code == 400
def test_validate_credential_access_rejects_unknown_field():
"""Unknown access keys must be rejected at write time so a destination can never
be stored in a shape the strict ``CredentialAccess`` read model later refuses to
parse (which would 500 every subsequent PATCH)."""
with pytest.raises(HTTPException) as exc:
validate_credential_access({"access": {"global": True, "legacy_field": "x"}})
assert exc.value.status_code == 400
assert "legacy_field" in exc.value.detail["error"]