fix(proxy): correct password.link share URL format and add route coverage

Use password.link's documented query-string link format ({base}/?{id}#{public})
instead of a path-style URL, drop the undocumented domain field, post via the
raw httpx client so non-201 responses surface as typed errors, classify
/key/share as a self-managed route so team/org admins reach its own authz
check, and pin the endpoint with a behavior-suite scenario

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Mubashir Osmani 2026-07-10 21:43:59 +00:00
parent 23ab246e92
commit 7ba50e1e39
5 changed files with 75 additions and 9 deletions

View file

@ -744,6 +744,7 @@ class LiteLLMRoutes(enum.Enum):
"/team/permissions_update",
"/team/daily/activity",
"/team/{team_id}/members/me",
"/key/share",
"/model/new",
"/model/update",
"/model/delete",

View file

@ -5957,7 +5957,7 @@ async def share_key_via_password_link(
client = get_async_httpx_client(llm_provider=httpxSpecialProvider.SecretManager)
async def _poster(url: str, headers: dict[str, str], body: dict[str, object]) -> HttpResponse:
return await client.post(url=url, headers=headers, json=body)
return await client.client.post(url, headers=headers, json=body)
result = await create_password_link_secret(
secret=data.key,

View file

@ -2,7 +2,7 @@ import base64
import json
import secrets
import string
from typing import Awaitable, Callable, Literal, Optional, Protocol, Union
from typing import Awaitable, Callable, Literal, Protocol, Union
import httpx
from cryptography.hazmat.primitives import hashes
@ -38,7 +38,6 @@ PasswordLinkResult = Union[PasswordLinkShare, PasswordLinkError]
class _SecretData(BaseModel):
id: str
domain: Optional[str] = None
class _CreateSecretResponse(BaseModel):
@ -124,7 +123,6 @@ async def create_password_link_secret(
except (ValueError, ValidationError) as exc:
return PasswordLinkError(message=f"Unexpected password.link response: {exc}")
domain = (parsed.data.domain or base).rstrip("/")
public_b64 = _b64(public_part.encode("utf-8"))
share_link = f"{domain}/{parsed.data.id}/#{public_b64}"
share_link = f"{base}/?{parsed.data.id}#{public_b64}"
return PasswordLinkShare(share_link=share_link, secret_id=parsed.data.id)

View file

@ -0,0 +1,67 @@
import os
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /key/share is admin-only (proxy admin, the key's team admin, or that
# team's org admin) and needs PASSWORD_LINK_API_KEY set on the proxy. The
# behavior suite runs with that env var unset, so an authorized caller on an
# existing key stops at the config gate (400) rather than hitting password.link
# over the network. These pin the auth, authz, not-found, and config gates.
@pytest.fixture(autouse=True)
def _unconfigured_password_link(monkeypatch):
monkeypatch.delenv("PASSWORD_LINK_API_KEY", raising=False)
async def test_key_share_requires_auth(proxy_client, world):
resp = await proxy_client.post(
"/key/share",
json={"key": world.keys[Actor.OWNER].cleartext},
)
assert resp.status_code == 401, resp.text
async def test_key_share_rejects_malformed_key(proxy_client, world):
resp = await proxy_client.post(
"/key/share",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"key": "not a valid key"},
)
assert resp.status_code == 400, resp.text
assert "invalid key format" in resp.text.lower()
async def test_key_share_unknown_key_is_not_found(proxy_client, world):
resp = await proxy_client.post(
"/key/share",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"key": "sk-does-not-exist-000000000000"},
)
assert resp.status_code == 404, resp.text
async def test_key_share_denies_non_admin(proxy_client, world):
resp = await proxy_client.post(
"/key/share",
headers={"Authorization": f"Bearer {world.keys[Actor.INTERNAL_USER].cleartext}"},
json={"key": world.keys[Actor.OWNER].cleartext},
)
assert resp.status_code == 403, resp.text
@pytest.mark.parametrize("actor", [Actor.PROXY_ADMIN, Actor.TEAM_ADMIN])
async def test_key_share_admin_reaches_config_gate(actor, proxy_client, world):
assert "PASSWORD_LINK_API_KEY" not in os.environ
resp = await proxy_client.post(
"/key/share",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
json={"key": world.keys[Actor.OWNER].cleartext},
)
assert resp.status_code == 400, resp.text
assert "password.link" in resp.text.lower()

View file

@ -54,7 +54,7 @@ async def test_creates_decryptable_one_time_link() -> None:
captured["url"] = url
captured["headers"] = headers
captured["body"] = body
return _FakeResponse(201, {"data": {"id": "abc123", "domain": "https://password.link"}})
return _FakeResponse(201, {"data": {"id": "abc123"}})
result = await create_password_link_secret(
secret=secret,
@ -77,7 +77,7 @@ async def test_creates_decryptable_one_time_link() -> None:
assert body["expiration"] == 12
assert body["max_views"] == 1
assert result.share_link.startswith("https://password.link/abc123/#")
assert result.share_link.startswith("https://password.link/?abc123#")
public_part = base64.b64decode(result.share_link.split("#", 1)[1]).decode("utf-8")
private_part = base64.b64decode(str(body["password_part_private"])).decode("utf-8")
@ -109,7 +109,7 @@ async def test_ciphertext_does_not_leak_plaintext() -> None:
@pytest.mark.asyncio
async def test_uses_api_base_when_response_has_no_domain() -> None:
async def test_link_uses_configured_api_base() -> None:
async def poster(url: str, headers: dict[str, str], body: dict[str, object]) -> _FakeResponse:
return _FakeResponse(201, {"data": {"id": "xyz"}})
@ -121,7 +121,7 @@ async def test_uses_api_base_when_response_has_no_domain() -> None:
)
assert isinstance(result, PasswordLinkShare)
assert result.share_link.startswith("https://vault.example.com/xyz/#")
assert result.share_link.startswith("https://vault.example.com/?xyz#")
@pytest.mark.asyncio