Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/_experimental/mcp_server/outbound_credentials/envelope.py: 53%
209 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""Client-held sealed envelope for the oauth_delegate DCR bridge.
3A DCR-bridge client holds ONE bearer that must carry BOTH a litellm identity and the
4upstream OAuth grant, with zero server-side storage. The gateway token endpoint mints a
5litellm-signed envelope (:func:`mint_envelope`); the MCP edge validates it, recovers the
6identity claims and the inner upstream grant (:func:`open_envelope`), and forwards the
7inner access token upstream. This module is pure and unwired: it imports nothing from
8endpoint or edge code, reads no proxy globals, and takes all key material and the clock
9as explicit parameters.
11Wire shape: ``llm_env_`` + an HS256 JWT (same signing approach as the BYOK session
12bearer in ``byok_oauth_endpoints.py``). Registered claims are ``iss``/``iat``/``exp``;
13custom claims are ``server_id``, ``key_hash``, and ``grant``, where ``grant`` is the
14upstream token grant serialized to JSON, encrypted with the repo's symmetric
15encryption helpers (``encrypt_value``/``decrypt_value`` from
16``encrypt_decrypt_utils`` — the same family ``encrypt_value_helper`` applies to
17persisted DCR credentials), and base64url-encoded, so the inner token never appears
18in plaintext anywhere in the envelope.
20Failures are values: :func:`open_envelope` returns one of the frozen
21``EnvelopeOpenError`` variants (discriminated on ``tag``) for invalid, expired,
22tampered, or undecryptable input, and :func:`mint_envelope` returns a typed error
23for oversized grants or an unrepresentable provider lifetime. Error values carry
24tags and metadata only, never token material.
26The pydantic input models reject programmer errors at construction (e.g. a
27non-positive ``expires_in`` or an empty required field). :func:`open_envelope` is
28additionally total over hostile, attacker-controlled input: it never raises, only
29returns an ``EnvelopeOpenError``. :func:`mint_envelope` operates on a
30gateway-supplied grant (an upstream IdP's UTF-8 JSON token response), so it does not
31defend against non-UTF-8 field content that cannot survive JSON parsing.
32"""
34from __future__ import annotations
36import base64
37from datetime import datetime, timedelta
38from typing import Final, Literal, TypeAlias
40import jwt
41from pydantic import BaseModel, ConfigDict, Field, SecretStr, ValidationError
43from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value, encrypt_value
45ENVELOPE_PREFIX: Final = "llm_env_"
46"""Marker prefix on every serialized ACCESS envelope so the edge can cheaply tell an envelope
47from a raw upstream token before doing any cryptography."""
49REFRESH_ENVELOPE_PREFIX: Final = "llm_refresh_"
50"""Marker prefix on every serialized REFRESH envelope. A distinct prefix keeps the two credentials
51routable without crypto and, together with the signed ``kind`` claim, stops one from being presented
52where the other is expected: a refresh envelope carries a long-lived upstream refresh token and is only
53ever presented back to the token endpoint, never forwarded upstream on a tool call."""
55ENVELOPE_ISSUER: Final = "litellm-mcp-bridge"
56"""``iss`` claim stamped into every envelope and required back on open."""
58MAX_ENVELOPE_TTL_SECONDS: Final = 3600
59"""Fallback ACCESS envelope lifetime when the upstream omits ``expires_in``.
61The historical exported name is retained for import compatibility. When the upstream
62reports a positive lifetime, the envelope matches it so a renewal does not consume a
63still-valid provider refresh grant."""
65MAX_REFRESH_ENVELOPE_TTL_SECONDS: Final = 1209600
66"""Hard ceiling on REFRESH envelope lifetime (14 days). A refresh envelope only renews the short-lived
67access envelope, and each renewal re-validates the sealed litellm key (revocation gates it) and is
68re-minted with a fresh window, so the practical bound is idle time, not a fixed session. ``exp`` is
69``min(upstream refresh_expires_in, this cap)`` (the cap alone when the upstream omits it); if the
70upstream refresh token dies first, the next renewal simply fails at the upstream and the client
71re-authenticates. The value is deliberately far shorter than a typical upstream refresh-token lifetime
72so a leaked refresh envelope is bounded even if the upstream would have honoured it for longer."""
74MAX_ENVELOPE_BYTES: Final = 12288
75"""Size cap on the final serialized envelope (prefix + JWT, in bytes). Upstream JWTs
76commonly run 2-4KB; base64 plus encryption overhead roughly doubles that inside the
77envelope, and common proxy/server header limits sit around 16KB total. 12288 leaves
78comfortable headroom for a large upstream token while keeping the envelope safely
79transmittable as a single Authorization header. Oversized grants are rejected with a
80typed error, never truncated."""
82_ENVELOPE_JWT_ALGORITHM: Final = "HS256"
84EnvelopeKind = Literal["access", "refresh"]
85"""Which credential an envelope is. Stamped into the signed claims and required to match on open, so a
86signature-valid envelope of one kind cannot be replayed as the other even if its wire prefix is swapped
87(the prefix is not part of the signed payload; this claim is)."""
90EnvelopeSubjectType: TypeAlias = Literal["key_hash", "user_id"]
91"""Discriminator for what litellm principal the envelope binds the grant to.
93``key_hash`` is a hashed virtual key (the scripted two-header client mints under the key it
94presents at the token endpoint); ``user_id`` is a litellm user subject (the interactive DCR
95client mints under the SSO-authenticated user, which is the only identity that browser login
96yields). Admission reloads a key record for the first and a user record for the second, then
97runs both through the same live-policy gate, so team/org/budget/revocation enforcement is
98identical either way."""
101class EnvelopeIdentity(BaseModel):
102 """The litellm principal the envelope binds the inner grant to.
104 ``subject`` is the principal identifier and ``subject_type`` says how to resolve it: a
105 hashed litellm key (``key_hash``) or a litellm user id (``user_id``), never a raw
106 credential (and the edge rejects a bare hash or id presented as a bearer). Admission
107 reloads the live record by it, so the principal's current team/org restrictions and its
108 revocation state are enforced at use time rather than frozen at mint time. ``server_id``
109 binds the envelope to one MCP server so it cannot be replayed across a server boundary.
110 """
112 model_config = ConfigDict(frozen=True)
113 server_id: str = Field(min_length=1)
114 subject_type: EnvelopeSubjectType
115 subject: str = Field(min_length=1)
118def key_hash_identity(server_id: str, key_hash: str) -> EnvelopeIdentity:
119 """The identity for the scripted client that mints under a presented virtual key."""
120 return EnvelopeIdentity(server_id=server_id, subject_type="key_hash", subject=key_hash)
123def user_identity(server_id: str, user_id: str) -> EnvelopeIdentity:
124 """The identity for the interactive DCR client that mints under its SSO user subject."""
125 return EnvelopeIdentity(server_id=server_id, subject_type="user_id", subject=user_id)
128class UpstreamTokenGrant(BaseModel):
129 """The upstream OAuth token response fields sealed inside the envelope.
131 ``expires_in`` must be positive when present; a non-positive value is a programmer
132 error rejected at construction. Token fields are ``SecretStr`` so reprs never leak
133 them.
134 """
136 model_config = ConfigDict(frozen=True)
137 access_token: SecretStr = Field(min_length=1)
138 token_type: str = Field(min_length=1)
139 refresh_token: SecretStr | None = None
140 scope: str | None = None
141 expires_in: int | None = Field(default=None, gt=0)
144class RefreshCredential(BaseModel):
145 """The upstream refresh grant sealed inside a refresh envelope.
147 Only the refresh token (plus the scope to re-request and the refresh token's own lifetime, when the
148 upstream reports it) is sealed; the access token is never in a refresh envelope. ``refresh_token`` is
149 a ``SecretStr`` so reprs never leak it, and ``expires_in`` (the refresh token's lifetime, not the
150 access token's) must be positive when present.
151 """
153 model_config = ConfigDict(frozen=True)
154 refresh_token: SecretStr = Field(min_length=1)
155 scope: str | None = None
156 expires_in: int | None = Field(default=None, gt=0)
159class EnvelopeKeys(BaseModel):
160 """Injected key material: the HS256 signing key and the symmetric encryption key.
162 ``signing_key`` must be at least 32 bytes: HS256's HMAC-SHA256 has a 256-bit
163 security level, RFC 7518 requires a key of at least that size, and a shorter key
164 makes PyJWT emit ``InsecureKeyLengthWarning``.
165 """
167 model_config = ConfigDict(frozen=True)
168 signing_key: SecretStr = Field(min_length=32)
169 encryption_key: SecretStr = Field(min_length=1)
172class SealedEnvelope(BaseModel):
173 """A minted envelope: the client-held bearer value and when it expires."""
175 model_config = ConfigDict(frozen=True)
176 token: SecretStr
177 expires_at: datetime
180class OpenedEnvelope(BaseModel):
181 """A validated access envelope: the identity it was minted for and the recovered grant."""
183 model_config = ConfigDict(frozen=True)
184 identity: EnvelopeIdentity
185 grant: UpstreamTokenGrant
188class OpenedRefreshEnvelope(BaseModel):
189 """A validated refresh envelope: the identity it was minted for and the recovered refresh grant."""
191 model_config = ConfigDict(frozen=True)
192 identity: EnvelopeIdentity
193 refresh: RefreshCredential
196class EnvelopeTooLarge(BaseModel):
197 """The serialized envelope exceeded ``MAX_ENVELOPE_BYTES``; carries sizes only."""
199 model_config = ConfigDict(frozen=True)
200 tag: Literal["envelope_too_large"] = "envelope_too_large"
201 size_bytes: int
202 max_bytes: int
205class EnvelopeLifetimeUnrepresentable(BaseModel):
206 """A positive provider lifetime cannot be represented as a Python datetime."""
208 model_config = ConfigDict(frozen=True)
209 tag: Literal["envelope_lifetime_unrepresentable"] = "envelope_lifetime_unrepresentable"
210 expires_in: int
213EnvelopeMintError: TypeAlias = EnvelopeTooLarge | EnvelopeLifetimeUnrepresentable
216class NotAnEnvelope(BaseModel):
217 """The candidate does not carry the envelope prefix."""
219 model_config = ConfigDict(frozen=True)
220 tag: Literal["not_an_envelope"] = "not_an_envelope"
223class BadSignature(BaseModel):
224 """The JWT signature does not verify under the provided signing key."""
226 model_config = ConfigDict(frozen=True)
227 tag: Literal["bad_signature"] = "bad_signature"
230class Expired(BaseModel):
231 """The envelope's ``exp`` is not in the future relative to the provided ``now``."""
233 model_config = ConfigDict(frozen=True)
234 tag: Literal["expired"] = "expired"
237class MalformedPayload(BaseModel):
238 """The token is not a well-formed envelope: undecodable JWT, wrong issuer, missing
239 or mistyped claims, or a decrypted grant that fails validation."""
241 model_config = ConfigDict(frozen=True)
242 tag: Literal["malformed_payload"] = "malformed_payload"
245class DecryptFailed(BaseModel):
246 """The signed ``grant`` blob could not be decrypted under the provided key."""
248 model_config = ConfigDict(frozen=True)
249 tag: Literal["decrypt_failed"] = "decrypt_failed"
252EnvelopeOpenError: TypeAlias = NotAnEnvelope | BadSignature | Expired | MalformedPayload | DecryptFailed
255class _EnvelopeClaims(BaseModel):
256 """Decoded-claims boundary that pins the exact shape :func:`mint_envelope` emits.
258 ``server_id``/``key_hash`` mirror the ``min_length`` constraints of
259 :class:`EnvelopeIdentity` so any claim set that validates here also constructs an
260 identity, keeping :func:`open_envelope` raise-free: a correctly signed JWT with an
261 empty identity claim fails here and maps to ``MalformedPayload``.
263 ``strict`` rejects coerced types (``exp: "123"``, ``exp: 123.0``) rather than opening
264 on them, and ``extra="forbid"`` rejects any claim the gateway never mints (a hostile
265 ``nbf``/``aud``/... rides along on a re-signed token). Since PyJWT's own ``iat``/
266 ``nbf``/``exp`` validators are disabled at decode (they raise on hostile claim types
267 and, for ``iat``/``nbf``, compare against the wall clock rather than the injected
268 ``now``), this model is the sole, total type gate for every registered claim.
269 """
271 model_config = ConfigDict(frozen=True, strict=True, extra="forbid")
272 iss: str
273 iat: int
274 exp: int
275 kind: EnvelopeKind
276 server_id: str = Field(min_length=1)
277 subject_type: EnvelopeSubjectType
278 subject: str = Field(min_length=1)
279 grant: str = Field(min_length=1)
282class _GrantWire(BaseModel):
283 model_config = ConfigDict(frozen=True)
284 access_token: str
285 token_type: str
286 refresh_token: str | None = None
287 scope: str | None = None
288 expires_in: int | None = None
291class _RefreshWire(BaseModel):
292 model_config = ConfigDict(frozen=True)
293 refresh_token: str
294 scope: str | None = None
295 expires_in: int | None = None
298def is_envelope(candidate: str) -> bool:
299 """Cheap prefix check for an ACCESS envelope so the edge can route envelopes vs raw tokens without
300 crypto. A refresh envelope has a different prefix and is not an access envelope."""
301 return candidate.startswith(ENVELOPE_PREFIX)
304def is_refresh_envelope(candidate: str) -> bool:
305 """Cheap prefix check for a REFRESH envelope so the token endpoint can route a refresh grant that
306 carries an envelope vs a raw upstream refresh token without crypto."""
307 return candidate.startswith(REFRESH_ENVELOPE_PREFIX)
310def mint_envelope(
311 identity: EnvelopeIdentity,
312 grant: UpstreamTokenGrant,
313 keys: EnvelopeKeys,
314 now: datetime,
315) -> SealedEnvelope | EnvelopeMintError:
316 """Seal ``grant`` for ``identity`` into a client-held envelope.
318 ``exp`` is ``grant.expires_in`` seconds from ``now`` when the upstream reports a
319 lifetime, or ``MAX_ENVELOPE_TTL_SECONDS`` when it does not. Returns
320 ``EnvelopeLifetimeUnrepresentable`` when that positive lifetime cannot be represented
321 as a Python datetime, or ``EnvelopeTooLarge`` when the serialized envelope exceeds
322 ``MAX_ENVELOPE_BYTES``.
323 """
324 ttl_seconds: Final = _envelope_ttl_seconds(grant.expires_in)
325 try:
326 expires_at: Final = now + timedelta(seconds=ttl_seconds)
327 except OverflowError:
328 return EnvelopeLifetimeUnrepresentable(expires_in=ttl_seconds)
329 return _seal(
330 kind="access",
331 prefix=ENVELOPE_PREFIX,
332 identity=identity,
333 grant_blob=_encrypt_grant_blob(_grant_plaintext(grant), keys.encryption_key),
334 expires_at=expires_at,
335 signing_key=keys.signing_key,
336 now=now,
337 )
340def open_envelope(
341 candidate: str,
342 keys: EnvelopeKeys,
343 now: datetime,
344) -> OpenedEnvelope | EnvelopeOpenError:
345 """Validate ``candidate`` and recover the identity and inner grant.
347 Never raises for bad input: every invalid, expired, tampered, or undecryptable
348 candidate maps to a distinct ``EnvelopeOpenError`` variant. The recovered
349 ``grant.expires_in`` is the value the upstream reported at mint time and is not
350 re-derived, so it is stale by up to the envelope's lifetime; callers that need a
351 live remaining lifetime should use ``now`` against the upstream, not this field.
352 """
353 claims: Final = _open_claims(candidate, prefix=ENVELOPE_PREFIX, expected_kind="access", keys=keys, now=now)
354 if not isinstance(claims, _EnvelopeClaims):
355 return claims
356 grant: Final = _decrypt_grant(claims.grant, keys.encryption_key)
357 if not isinstance(grant, UpstreamTokenGrant):
358 return grant
359 return OpenedEnvelope(
360 identity=EnvelopeIdentity(server_id=claims.server_id, subject_type=claims.subject_type, subject=claims.subject),
361 grant=grant,
362 )
365def mint_refresh_envelope(
366 identity: EnvelopeIdentity,
367 refresh: RefreshCredential,
368 keys: EnvelopeKeys,
369 now: datetime,
370) -> SealedEnvelope | EnvelopeMintError:
371 """Seal ``refresh`` for ``identity`` into a long-lived, client-held refresh envelope.
373 ``exp`` is ``min(refresh.expires_in, MAX_REFRESH_ENVELOPE_TTL_SECONDS)`` seconds from ``now`` (the
374 cap alone when the upstream omits the refresh lifetime). Sealing a distinct ``kind="refresh"`` claim
375 is what keeps a refresh envelope from ever opening as an access credential at the MCP edge. Returns
376 ``EnvelopeTooLarge`` when the serialized envelope exceeds ``MAX_ENVELOPE_BYTES``.
377 """
378 expires_at: Final = now + timedelta(seconds=_refresh_ttl_seconds(refresh.expires_in))
379 return _seal(
380 kind="refresh",
381 prefix=REFRESH_ENVELOPE_PREFIX,
382 identity=identity,
383 grant_blob=_encrypt_grant_blob(_refresh_plaintext(refresh), keys.encryption_key),
384 expires_at=expires_at,
385 signing_key=keys.signing_key,
386 now=now,
387 )
390def open_refresh_envelope(
391 candidate: str,
392 keys: EnvelopeKeys,
393 now: datetime,
394) -> OpenedRefreshEnvelope | EnvelopeOpenError:
395 """Validate a refresh ``candidate`` and recover the identity and inner refresh grant.
397 Total over hostile input exactly like :func:`open_envelope`: every invalid, expired, tampered,
398 wrong-kind, or undecryptable candidate maps to a distinct ``EnvelopeOpenError`` variant, never a
399 raise. The ``kind="refresh"`` claim is required, so an access envelope re-prefixed as a refresh one
400 is rejected as ``MalformedPayload``.
401 """
402 claims: Final = _open_claims(candidate, prefix=REFRESH_ENVELOPE_PREFIX, expected_kind="refresh", keys=keys, now=now)
403 if not isinstance(claims, _EnvelopeClaims):
404 return claims
405 refresh: Final = _decrypt_refresh(claims.grant, keys.encryption_key)
406 if not isinstance(refresh, RefreshCredential):
407 return refresh
408 return OpenedRefreshEnvelope(
409 identity=EnvelopeIdentity(server_id=claims.server_id, subject_type=claims.subject_type, subject=claims.subject),
410 refresh=refresh,
411 )
414def _seal(
415 kind: EnvelopeKind,
416 prefix: str,
417 identity: EnvelopeIdentity,
418 grant_blob: str,
419 expires_at: datetime,
420 signing_key: SecretStr,
421 now: datetime,
422) -> SealedEnvelope | EnvelopeTooLarge:
423 """Sign the claims for either envelope kind and enforce the size cap. Shared by both mints so the
424 JWT shape, issuer, and size guard cannot drift between access and refresh envelopes."""
425 claims: Final = _EnvelopeClaims(
426 iss=ENVELOPE_ISSUER,
427 iat=int(now.timestamp()),
428 exp=int(expires_at.timestamp()),
429 kind=kind,
430 server_id=identity.server_id,
431 subject_type=identity.subject_type,
432 subject=identity.subject,
433 grant=grant_blob,
434 )
435 token = prefix + jwt.encode(claims.model_dump(), signing_key.get_secret_value(), algorithm=_ENVELOPE_JWT_ALGORITHM)
436 size_bytes: Final = len(token.encode("utf-8"))
437 if size_bytes > MAX_ENVELOPE_BYTES:
438 return EnvelopeTooLarge(size_bytes=size_bytes, max_bytes=MAX_ENVELOPE_BYTES)
439 return SealedEnvelope(token=SecretStr(token), expires_at=expires_at)
442def _open_claims(
443 candidate: str,
444 prefix: str,
445 expected_kind: EnvelopeKind,
446 keys: EnvelopeKeys,
447 now: datetime,
448) -> _EnvelopeClaims | EnvelopeOpenError:
449 """Prefix-route, size-bound, signature-verify, kind-check, and expiry-check an attacker-controlled
450 candidate, shared by both openers so the security gate is identical for access and refresh. Returns
451 the validated claims or a distinct ``EnvelopeOpenError``; never raises."""
452 if not candidate.startswith(prefix):
453 return NotAnEnvelope()
454 # UTF-8 byte length is never below character length, so a character count already over the cap
455 # rejects an oversize candidate in O(1) without encoding it; the exact byte check then runs only on
456 # candidates already bounded to <= MAX_ENVELOPE_BYTES characters.
457 if len(candidate) > MAX_ENVELOPE_BYTES:
458 return MalformedPayload()
459 if len(candidate.encode("utf-8", "surrogatepass")) > MAX_ENVELOPE_BYTES:
460 return MalformedPayload()
461 claims: Final = _decode_claims(candidate.removeprefix(prefix), keys.signing_key)
462 if not isinstance(claims, _EnvelopeClaims):
463 return claims
464 if claims.kind != expected_kind:
465 return MalformedPayload()
466 if now.timestamp() >= claims.exp:
467 return Expired()
468 return claims
471def _envelope_ttl_seconds(upstream_expires_in: int | None) -> int:
472 if upstream_expires_in is None:
473 return MAX_ENVELOPE_TTL_SECONDS
474 return upstream_expires_in
477def _refresh_ttl_seconds(upstream_refresh_expires_in: int | None) -> int:
478 if upstream_refresh_expires_in is None:
479 return MAX_REFRESH_ENVELOPE_TTL_SECONDS
480 return min(upstream_refresh_expires_in, MAX_REFRESH_ENVELOPE_TTL_SECONDS)
483def _grant_plaintext(grant: UpstreamTokenGrant) -> str:
484 wire: Final = _GrantWire(
485 access_token=grant.access_token.get_secret_value(),
486 token_type=grant.token_type,
487 refresh_token=None if grant.refresh_token is None else grant.refresh_token.get_secret_value(),
488 scope=grant.scope,
489 expires_in=grant.expires_in,
490 )
491 return wire.model_dump_json(exclude_none=True)
494def _refresh_plaintext(refresh: RefreshCredential) -> str:
495 wire: Final = _RefreshWire(
496 refresh_token=refresh.refresh_token.get_secret_value(),
497 scope=refresh.scope,
498 expires_in=refresh.expires_in,
499 )
500 return wire.model_dump_json(exclude_none=True)
503def _decode_claims(
504 compact: str,
505 signing_key: SecretStr,
506) -> _EnvelopeClaims | BadSignature | MalformedPayload:
507 """Verify the HS256 signature and shape of an attacker-controlled compact JWT.
509 ``compact`` is fully hostile and bounded to ``MAX_ENVELOPE_BYTES`` by the caller.
510 PyJWT's ``iat``/``nbf``/``exp`` validators are disabled: they raise on hostile claim
511 types and, for ``iat``/``nbf``, compare against the wall clock rather than the
512 injected ``now`` (``exp`` is checked by the caller against ``now``). Apart from a
513 signature mismatch (``BadSignature``), every decode failure is ``MalformedPayload``:
514 a non-UTF-8 candidate surfaces as ``UnicodeEncodeError`` (a ``ValueError``), a
515 non-string registered claim such as ``iss`` as a ``TypeError`` from PyJWT's claim
516 validators, and a wrong issuer or structurally invalid token as an
517 ``InvalidTokenError``. ``_EnvelopeClaims`` is the total type gate for the payload.
518 """
519 try:
520 payload: Final = jwt.decode(
521 compact,
522 signing_key.get_secret_value(),
523 algorithms=[_ENVELOPE_JWT_ALGORITHM],
524 issuer=ENVELOPE_ISSUER,
525 options={
526 "verify_exp": False,
527 "verify_iat": False,
528 "verify_nbf": False,
529 "require": ["iss", "iat", "exp"],
530 },
531 )
532 except jwt.InvalidSignatureError:
533 return BadSignature()
534 except (jwt.InvalidTokenError, ValueError, TypeError):
535 return MalformedPayload()
536 try:
537 return _EnvelopeClaims.model_validate(payload)
538 except ValidationError:
539 return MalformedPayload()
542def _encrypt_grant_blob(plaintext: str, encryption_key: SecretStr) -> str:
543 ciphertext: Final = bytes(encrypt_value(value=plaintext, signing_key=encryption_key.get_secret_value()))
544 return base64.urlsafe_b64encode(ciphertext).decode("ascii")
547def _decrypt_grant(
548 blob: str,
549 encryption_key: SecretStr,
550) -> UpstreamTokenGrant | DecryptFailed | MalformedPayload:
551 from nacl.exceptions import CryptoError
553 try:
554 plaintext: Final = decrypt_value(
555 value=base64.urlsafe_b64decode(blob),
556 signing_key=encryption_key.get_secret_value(),
557 )
558 except (CryptoError, ValueError):
559 return DecryptFailed()
560 try:
561 return UpstreamTokenGrant.model_validate_json(plaintext)
562 except ValidationError:
563 return MalformedPayload()
566def _decrypt_refresh(
567 blob: str,
568 encryption_key: SecretStr,
569) -> RefreshCredential | DecryptFailed | MalformedPayload:
570 from nacl.exceptions import CryptoError
572 try:
573 plaintext: Final = decrypt_value(
574 value=base64.urlsafe_b64decode(blob),
575 signing_key=encryption_key.get_secret_value(),
576 )
577 except (CryptoError, ValueError):
578 return DecryptFailed()
579 try:
580 return RefreshCredential.model_validate_json(plaintext)
581 except ValidationError:
582 return MalformedPayload()