diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 50b87264f3f..1c8608305a2 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -835,6 +835,14 @@ class LiteLLMRoutes(enum.Enum): "/team/daily/activity/aggregated", "/team/spend/by_user", "/team/{team_id}/members/me", + # POST/GET the team's logging callbacks, and DELETE one of them. Every + # handler calls _verify_team_access, which admits only a proxy admin, an + # org admin for the team, or an admin of this team. + # + # team_id is a free-form string, so it spells these with the same path + # converter the router uses; the gate matches that converter. + "/team/{team_id:path}/callback", + "/team/{team_id:path}/callback/{callback_name}", "/model/new", "/model/update", "/model/delete", diff --git a/litellm/proxy/auth/route_checks.py b/litellm/proxy/auth/route_checks.py index 4dba2497bb9..953e3cf3e88 100644 --- a/litellm/proxy/auth/route_checks.py +++ b/litellm/proxy/auth/route_checks.py @@ -497,10 +497,22 @@ class RouteChecks: def _placeholder_to_regex(match: re.Match) -> str: placeholder: Final = match.group(0).strip("{}") - if placeholder.endswith(":path"): - # allow "/" in the placeholder value, but don't eat the route suffix after ":" - return r"[^:]+" - return r"[^/]+" + if not placeholder.endswith(":path"): + return r"[^/]+" + # A ":path" placeholder takes whatever the router's own path + # converter takes, slashes and colons alike, so an id spelled with + # either (or both) still matches the template it was mounted under. + # + # Unless the template puts a ":" literal of its own after the + # placeholder: the Google routes end in ":generateContent" and + # friends, and there the value has to stop before that suffix + # rather than swallow it and match a different verb. + # + # "[\s\S]" rather than ".", because "." stops at a newline and the + # path converter does not: a %0A anywhere in the value would leave + # the route unmatched here while still reaching the handler, which + # turns this gate into a bypass for the lists built on it. + return r"[^:]+" if ":" in match.string[match.end() :] else r"[\s\S]+" pattern = re.sub(r"\{[^}]+\}", _placeholder_to_regex, pattern) # Anchor the pattern to match the entire string diff --git a/litellm/proxy/common_utils/callback_config_validation.py b/litellm/proxy/common_utils/callback_config_validation.py index 7ee3bd8d829..c9d97068313 100644 --- a/litellm/proxy/common_utils/callback_config_validation.py +++ b/litellm/proxy/common_utils/callback_config_validation.py @@ -44,6 +44,91 @@ def _langfuse_environment_error(callback_vars: Mapping[str, str]) -> str | None: return None +# Which credential family a dynamic variable belongs to. The families are the +# integrations that share one account: every langfuse_* variable configures the +# same Langfuse project whether it rides the classic callback or the OTel one, +# and every dd_* variable configures the same Datadog account. +_VAR_FAMILIES: Final[Mapping[str, str]] = MappingProxyType( + { + "arize_": "Arize", + "dd_": "Datadog", + "gcs_": "GCS", + "humanloop_": "Humanloop", + "langfuse_": "Langfuse", + "langsmith_": "LangSmith", + "newrelic_": "New Relic", + "posthog_": "PostHog", + "wandb_": "Weights & Biases", + "weave_": "Weights & Biases", + } +) + + +def _family_of(var: str) -> str | None: + """The credential family ``var`` configures, or ``None`` if it configures none. + + ``turn_off_message_logging`` and friends belong to no backend, so they carry + no credentials anyone could redirect. + """ + return next((family for prefix, family in _VAR_FAMILIES.items() if var.startswith(prefix)), None) + + +def cross_entry_family_error( + callback_vars: Mapping[str, str] | None, + stored_vars_by_entry: Sequence[Mapping[str, str]], +) -> str | None: + """Reject an entry that changes what a family another entry holds resolves to. + + Every stored entry's variables are flattened into one dict before a request + reads them, and the flattened dict is what the exporter authenticates and + addresses with. So an entry naming only a destination is enough to redirect + credentials that were written somewhere else: a host on a second entry pairs + with the key from the first, and the request carries that key to the new + host. + + Two rules together keep the flattened dict out of the caller's hands. A + variable the family already configures has to keep the value it has, so + nothing already in use can be moved. A variable the family does not yet + configure may only carry a value the family already holds, which is what lets + the same credential go in under its other spelling (``langfuse_secret`` and + ``langfuse_secret_key`` are one key) without anything here having to list the + spellings. Between them, no value the caller chose can enter the family, and + repeating the family as it stands is still allowed -- that is how one + integration gets registered for both the success and the failure event. + + A team admin who does want to move a family deletes the entry holding it + first, which reveals nothing. + + Only the writers this endpoint newly admits are held to this, because a proxy + admin already holds every credential the proxy has. + + ``stored_vars_by_entry`` has to arrive decrypted; the credential values are + encrypted at rest and ciphertext never equals the plaintext coming in. + """ + if not callback_vars: + return None + stored_by_var: Final = { + var: value for entry in stored_vars_by_entry for var, value in entry.items() if _family_of(var) is not None + } + family_values: Final = frozenset( + (family, value) + for entry in stored_vars_by_entry + for var, value in entry.items() + if (family := _family_of(var)) is not None + ) + held_families: Final = frozenset(family for family, _ in family_values) + return next( + ( + f"{family} is already configured by another callback entry on this team. " + f"Remove that entry before setting {var} here." + for var, value, family in ((v, callback_vars[v], _family_of(v)) for v in callback_vars) + if family in held_families + and (stored_by_var[var] != value if var in stored_by_var else (family, value) not in family_values) + ), + None, + ) + + def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None: """Validate every ``logging`` entry of a team/key metadata payload.""" if not metadata: diff --git a/litellm/proxy/management_endpoints/team_callback_endpoints.py b/litellm/proxy/management_endpoints/team_callback_endpoints.py index fe658a13c24..1932e89717b 100644 --- a/litellm/proxy/management_endpoints/team_callback_endpoints.py +++ b/litellm/proxy/management_endpoints/team_callback_endpoints.py @@ -20,6 +20,7 @@ from litellm.proxy._types import ( LiteLLM_AuditLogs, LiteLLM_TeamTable, LitellmTableNames, + LitellmUserRoles, ProxyErrorTypes, ProxyException, TeamCallbackDeleteResponse, @@ -28,7 +29,10 @@ from litellm.proxy._types import ( UserAPIKeyAuth, ) from litellm.proxy.auth.user_api_key_auth import user_api_key_auth -from litellm.proxy.common_utils.callback_config_validation import callback_config_error +from litellm.proxy.common_utils.callback_config_validation import ( + callback_config_error, + cross_entry_family_error, +) from litellm.proxy.common_utils.callback_utils import ( _CALLBACK_VAR_ENCRYPTED_PREFIX, decrypt_callback_vars, @@ -230,6 +234,22 @@ def _callback_error(status_code: int, message: str) -> HTTPException: ) +def _unknown_team_error(team_id: str, user_api_key_dict: UserAPIKeyAuth, status_code: int) -> HTTPException: + """Report an unknown team without telling an unauthorized caller that it is unknown. + + These routes are reachable by any authenticated caller so that a team admin can + get as far as _verify_team_access. A distinct "does not exist" would therefore let + any valid key probe which team ids exist, so a caller who could not have managed + the team either way gets the same 403 body _verify_team_access raises. + """ + if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN: + return _callback_error(status_code, f"Team id = {team_id} does not exist.") + return HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="You do not have access to this team", + ) + + @router.post( "/team/{team_id:path}/callback", tags=["team management"], @@ -304,10 +324,7 @@ async def add_team_callbacks( # Check if team_id exists already _existing_team = await prisma_client.get_data(team_id=team_id, table_name="team", query_type="find_unique") if _existing_team is None: - raise HTTPException( - status_code=400, - detail={"error": f"Team id = {team_id} does not exist. Please use a different team id."}, - ) + raise _unknown_team_error(team_id, user_api_key_dict, status.HTTP_400_BAD_REQUEST) # IDOR guard: only proxy admins / org admins / team admins of THIS # team may write callback credentials. Without this, any @@ -326,6 +343,28 @@ async def add_team_callbacks( if team_callback_settings is None or not isinstance(team_callback_settings, list): team_callback_settings = [] + # One entry has to own a credential family end to end. The entries are + # flattened into one dict before a request reads them, so an entry + # naming only a destination would pair with a key written on another + # entry and carry it to that destination -- a key a team admin can read + # back nowhere. Repeating a value the owning entry already stores is + # fine, which is how one integration covers both events. Proxy admins + # are exempt: they already hold every credential the proxy has. + if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: + # Decrypted, because the check compares the incoming values against + # the stored ones and the credentials are encrypted at rest. + decrypted_logging: Final = decrypt_callback_vars(team_metadata).get("logging") + stored_entries: Final = decrypted_logging if isinstance(decrypted_logging, list) else () + stored_entry_vars: Final = [ # mutable-ok: read-only input to the check, never stored + entry.get("callback_vars") or {} for entry in stored_entries + ] + family_error: Final = cross_entry_family_error(data.callback_vars, stored_entry_vars) + if family_error is not None: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=family_error, + ) + ## check if it already exists, for the same callback event for callback in team_callback_settings: if ( @@ -452,7 +491,7 @@ async def delete_team_callback( team_id=team_id, table_name="team", query_type="find_unique" ) if _existing_team is None: - raise _callback_error(404, f"Team id = {team_id} does not exist.") + raise _unknown_team_error(team_id, user_api_key_dict, status.HTTP_404_NOT_FOUND) # IDOR guard: only proxy admins / org admins / team admins of THIS team may # deregister its callbacks, otherwise any authenticated key holder could @@ -726,10 +765,7 @@ async def get_team_callbacks( # Check if team_id exists _existing_team = await prisma_client.get_data(team_id=team_id, table_name="team", query_type="find_unique") if _existing_team is None: - raise HTTPException( - status_code=404, - detail={"error": f"Team id = {team_id} does not exist."}, - ) + raise _unknown_team_error(team_id, user_api_key_dict, status.HTTP_404_NOT_FOUND) # IDOR guard: callback metadata holds third-party API credentials # (Langfuse / Langsmith / GCS). Only proxy admins / org admins / diff --git a/tests/test_litellm/proxy/auth/test_route_checks.py b/tests/test_litellm/proxy/auth/test_route_checks.py index 5b15d4a7d5e..0fd19f518d9 100644 --- a/tests/test_litellm/proxy/auth/test_route_checks.py +++ b/tests/test_litellm/proxy/auth/test_route_checks.py @@ -3638,3 +3638,191 @@ def test_agent_registry_route_gate_open_to_non_admin_roles(user_role, method, ro valid_token=valid_token, request_data={}, ) +TEAM_CALLBACK_ROUTES = ( + "/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/callback", + "/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/callback/langfuse", + # the routes register team_id with the :path converter, so a team id may + # contain a slash + "/team/tenant/06bda574/callback", + "/team/tenant/06bda574/callback/langfuse", + # team_id is a free-form string, so it may also contain a colon + "/team/tenant:06bda574/callback", + "/team/tenant:06bda574/callback/langfuse", + # or both, which is the shape neither a "[^:]+" nor a "[^/]+" expansion + # of the placeholder reaches on its own + "/team/tenant:acme/prod/callback", + "/team/tenant:acme/prod/callback/langfuse", +) + + +def _gate(route, role) -> str: + """Drive the real route gate for a non-proxy-admin caller. + + Reports "allowed" when the gate lets the request through to its handler, and + the denial message otherwise, so a caller asserts the verdict as a value + instead of on whether an exception escaped. + """ + user_obj = LiteLLM_UserTable( + user_id="team_admin_user", + user_email="team-admin@example.com", + user_role=role, + ) + request = MagicMock(spec=Request) + request.query_params = {} + try: + RouteChecks.non_proxy_admin_allowed_routes_check( + user_obj=user_obj, + _user_role=role, + route=route, + request=request, + valid_token=UserAPIKeyAuth(user_id="team_admin_user", user_role=role), + request_data={}, + ) + except Exception as exc: + return f"denied: {exc}" + return "allowed" + + +def test_team_callback_routes_are_self_managed(): + """The grant has to come from self_managed_routes specifically. + + That list is the one whose entries carry no role predicate, so the handler + decides. Granting the same paths through internal_user_routes instead would + look identical for an internal_user while silently denying the org admins and + view-only roles that list does not cover. + """ + for template in ( + "/team/{team_id:path}/callback", + "/team/{team_id:path}/callback/{callback_name}", + ): + assert template in LiteLLMRoutes.self_managed_routes.value + + +@pytest.mark.parametrize("route", TEAM_CALLBACK_ROUTES) +@pytest.mark.parametrize( + "role", + [ + LitellmUserRoles.INTERNAL_USER.value, + LitellmUserRoles.INTERNAL_USER_VIEW_ONLY.value, + LitellmUserRoles.ORG_ADMIN.value, + ], +) +def test_team_callback_routes_reach_their_handler_for_non_admins(route, role): + """A team admin manages their own team's logging callbacks, so the route gate + must let a non-proxy-admin through to the handler. + + The handler is what authorizes: every team callback endpoint calls + _verify_team_access, which admits only a proxy admin, an org admin for the + team, or an admin of that team, and 403s everyone else. Before this, the gate + rejected the team admin with a 401 naming proxy admin, so the handler's own + check was unreachable for them. + """ + assert _gate(route, role) == "allowed" + + +@pytest.mark.parametrize( + "pattern, route, matches", + [ + # a :path placeholder takes what the router's path converter takes + ("/team/{team_id:path}/callback", "/team/plain/callback", True), + ("/team/{team_id:path}/callback", "/team/tenant/acme/callback", True), + ("/team/{team_id:path}/callback", "/team/tenant:acme/callback", True), + ("/team/{team_id:path}/callback", "/team/tenant:acme/prod/callback", True), + # and still has to reach the template's own suffix + ("/team/{team_id:path}/callback", "/team/tenant:acme/disable_logging", False), + # a template with a ":" literal after the placeholder keeps the suffix + ( + "/v1beta/models/{model_name:path}:generateContent", + "/v1beta/models/gemini-2.5-flash:generateContent", + True, + ), + ( + "/v1beta/models/{model_name:path}:generateContent", + "/v1beta/models/publishers/google/gemini-2.5-flash:generateContent", + True, + ), + # the value must not swallow that suffix and match a different verb + ( + "/v1beta/models/{model_name:path}:generateContent", + "/v1beta/models/gemini-2.5-flash:countTokens", + False, + ), + # a %0A in the value reaches the handler through the path converter, so + # the gate has to see it too or DISABLE_ADMIN_ENDPOINTS is bypassable + ("/v1/mcp/server/{path:path}", "/v1/mcp/server/abc\ndef", True), + ("/team/{team_id:path}/callback", "/team/ten\nant/callback", True), + ("/v1beta/models/{model_name:path}:generateContent", "/v1beta/models/gem\nini:generateContent", True), + # an ordinary placeholder stays one segment + ("/team/{team_id}/members/me", "/team/abc/members/me", True), + ("/team/{team_id}/members/me", "/team/tenant/abc/members/me", False), + ("/team/{team_id}/members/me", "/team/ab\nc/members/me", True), + ], +) +def test_path_placeholder_matches_what_the_router_accepts(pattern, route, matches): + """The gate's placeholder expansion has to agree with the router's. + + A team id may carry a slash, a colon, or both, and the router mounted these + paths with the same :path converter, so an id the router routes must not be + an id the gate fails to recognize. The one narrowing that stays is a template + whose own suffix begins with a colon: there the value stops before it, or + ":generateContent" would also match a ":countTokens" request. + """ + assert RouteChecks._route_matches_pattern(route=route, pattern=pattern) is matches + + +# Every other route the proxy mounts under /team/{team_id}, spelled the way it +# is registered. None of them takes a path converter, so none can be reached by +# a URL that ends in the callback suffix. +PROTECTED_TEAM_ROUTES = ( + "/team/06bda574-5ca9-43d3-beb8-3b23c2f17112", + "/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/disable_logging", + "/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/members/me", + "/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/member/u-1/reset_spend", + # the same routes with the callback suffix spliced in, which is the shape a + # caller would craft to make a protected route look self-managed + "/team/06bda574/callback/disable_logging/x", + "/team/06bda574/callback/member/u-1/reset_spend", + "/team/06bda574/callback/members/me", +) + + +@pytest.mark.parametrize("route", PROTECTED_TEAM_ROUTES) +def test_the_callback_grant_does_not_reach_another_team_route(route): + """Widening the callback templates must not hand out any neighbouring route. + + The grant is two templates ending in the callback suffix. Every other team + route registers an ordinary single-segment placeholder, so no URL the router + sends to one of them can end in "/callback" or "/callback/" -- and the + gate must agree, or a crafted team id would carry a caller into a handler + the grant never covered. + """ + for template in ( + "/team/{team_id:path}/callback", + "/team/{team_id:path}/callback/{callback_name}", + ): + assert RouteChecks._route_matches_pattern(route=route, pattern=template) is False + + +def test_team_disable_logging_stays_proxy_admin_only(): + """disable_logging was left out of the grant, so it must still be rejected at + the gate. It is the one team callback route a team admin cannot reach.""" + verdict = _gate( + "/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/disable_logging", + LitellmUserRoles.INTERNAL_USER.value, + ) + + assert "Only proxy admin" in verdict + assert "disable_logging" in verdict + + +@pytest.mark.parametrize( + "route", + [ + "/team/06bda574-5ca9-43d3-beb8-3b23c2f17112", + "/team/update", + "/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/model/add", + ], +) +def test_neighbouring_team_routes_stay_closed(route): + """The grant is the callback paths and nothing else on the team namespace.""" + assert "Only proxy admin" in _gate(route, LitellmUserRoles.INTERNAL_USER.value) diff --git a/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py index 08e931e6405..bdc12dad4bc 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py @@ -19,6 +19,7 @@ from litellm.proxy._types import ( LitellmUserRoles, UserAPIKeyAuth, ) +from litellm.proxy.common_utils.callback_config_validation import cross_entry_family_error from litellm.proxy.management_endpoints.team_callback_endpoints import ( add_team_callbacks, delete_team_callback, @@ -1443,3 +1444,118 @@ async def test_delete_team_callback_route_accepts_team_ids_containing_slashes(): assert response.json()["data"]["success_callbacks"] == ["langsmith"] written = json.loads(mock_prisma.db.litellm_teamtable.update.await_args.kwargs["data"]["metadata"]) assert [entry["callback_name"] for entry in written["logging"]] == ["langsmith"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "call_handler", + [ + lambda caller: add_team_callbacks( + data=AddTeamCallback( + callback_name="langfuse", + callback_type="success", + callback_vars={"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, + ), + http_request=Mock(spec=Request), + team_id="team-does-not-exist", + user_api_key_dict=caller, + ), + lambda caller: get_team_callbacks( + http_request=Mock(spec=Request), + team_id="team-does-not-exist", + user_api_key_dict=caller, + ), + lambda caller: delete_team_callback( + http_request=Mock(spec=Request), + team_id="team-does-not-exist", + callback_name="langfuse", + user_api_key_dict=caller, + ), + ], + ids=["add", "get", "delete"], +) +async def test_unknown_team_is_indistinguishable_from_no_access(call_handler, unauthorized_caller): + """An unauthorized caller must not learn whether a team id exists. + + These routes are reachable by any authenticated caller so a team admin can get + as far as the access check, so a distinct "does not exist" would turn them into + a probe for valid team ids. The unknown-team response has to match the + no-access one exactly, status and body. + """ + with patch("litellm.proxy.proxy_server.prisma_client") as mock_client: # test-quality-ok: the handler imports prisma_client from proxy_server at call time, so there is no seam to inject through + mock_client.get_data = AsyncMock(return_value=None) + with pytest.raises(HTTPException) as unknown_team: + await call_handler(unauthorized_caller) + + with patch("litellm.proxy.proxy_server.prisma_client") as mock_client: # test-quality-ok: the handler imports prisma_client from proxy_server at call time, so there is no seam to inject through + mock_client.get_data = AsyncMock(return_value=_team_row()) + mock_client.db.litellm_teamtable.update = AsyncMock() + with patch( # test-quality-ok: _verify_team_access calls this module-level helper directly, so there is no seam to inject through + "litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team", + new_callable=AsyncMock, + return_value=False, + ): + with pytest.raises(HTTPException) as no_access: + await call_handler(unauthorized_caller) + + assert unknown_team.value.status_code == no_access.value.status_code == 403 + assert unknown_team.value.detail == no_access.value.detail + assert "does not exist" not in str(unknown_team.value.detail) + + +@pytest.mark.asyncio +async def test_proxy_admin_still_told_the_team_is_unknown(): + """The masking is only for callers who could not have managed the team; a proxy + admin keeps the diagnosable error.""" + admin = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, user_id="admin", api_key="sk-admin") + with patch("litellm.proxy.proxy_server.prisma_client") as mock_client: # test-quality-ok: the handler imports prisma_client from proxy_server at call time, so there is no seam to inject through + mock_client.get_data = AsyncMock(return_value=None) + with pytest.raises(HTTPException) as exc: + await get_team_callbacks( + http_request=Mock(spec=Request), + team_id="team-does-not-exist", + user_api_key_dict=admin, + ) + + assert exc.value.status_code == 404 + assert "does not exist" in str(exc.value.detail) + + +@pytest.mark.parametrize( + "new_vars, stored, rejected", + [ + # the redirect, in every carrier a caller could pick: an entry naming + # only a host, pairing with a key pair written on another entry + ({"langfuse_host": "http://attacker.invalid"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], True), + # the sibling carrier -- langfuse and langfuse_otel are one account + ({"langfuse_host": "http://attacker.invalid"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_secret_key": "sk"}], True), + # a destination variable no integration registry lists + ({"dd_agent_host": "attacker.invalid"}, [{"dd_api_key": "k", "dd_site": "us5.datadoghq.com"}], True), + # one entry owning its family end to end is the feature + ({"langfuse_host": "https://eu.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, [], False), + # a different family alongside an existing one stays fine + ({"gcs_bucket_name": "bucket"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], False), + ({"langsmith_api_key": "k"}, [{"dd_api_key": "k"}], False), + # variables that configure no backend carry nothing to redirect + ({"turn_off_message_logging": "true"}, [{"langfuse_secret_key": "sk"}], False), + # the same integration registered for a second event: identical values + # flatten to the identical dict, so there is nothing to redirect + ({"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], False), + # the same credential under its other spelling is the same credential + ({"langfuse_secret": "sk"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], False), + # a value the family already holds cannot be moved into another of its + # variables either; the exporter would address or authenticate with it + ({"langfuse_host": "pk"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk"}], True), + # the same shape with one value moved is the redirect again + ({"langfuse_host": "http://attacker.invalid", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], True), + ], +) +def test_one_entry_owns_a_credential_family(new_vars, stored, rejected): + """A team admin must not be able to redirect a credential they cannot read. + + The stored entries are flattened into one dict before a request reads them, + so an entry naming only a destination pairs with a key written elsewhere and + carries it to that destination. + """ + error = cross_entry_family_error(new_vars, stored) + assert (error is not None) is rejected