mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-12 03:48:56 +00:00
40a9b959a6
* test(e2e): add enum values, auto-discovering label gates and secret hiding for e2e metadata * test(e2e): tag a2a, access_control, other, secret_manager and migrations tests with Subject metadata * docs(e2e): name every markerless harness test file that carries no Subject * test(e2e): keep the step discovery comprehensions to one for clause
107 lines
4.3 KiB
Python
107 lines
4.3 KiB
Python
"""Live e2e: UI/CLI session tokens are accepted only while valid and only when minted as session tokens.
|
|
|
|
The runner mints its own session tokens under the proxy's salt key, so the valid and expired cases run in
|
|
seconds instead of waiting out a real login's expiry.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Final
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
from e2e_config import MASTER_KEY, unique_marker
|
|
from e2e_http import UnauthorizedError, unwrap
|
|
from lifecycle import ResourceManager
|
|
from models import KeyGenerateBody, KeyLoggingCallback, KeyLoggingCallbackVars, KeyMetadata
|
|
from other_client import OtherClient
|
|
from e2e_metadata import Domain, Subject, meta
|
|
|
|
pytestmark = pytest.mark.e2e
|
|
|
|
SALT_KEY: Final = os.environ.get("LITELLM_SALT_KEY") or MASTER_KEY
|
|
SESSION_TOKEN_PREFIX: Final = "litellm_login_"
|
|
ENCRYPTED_PREFIX: Final = "litellm_enc::"
|
|
|
|
|
|
def _admin_session_token(expires_at: datetime) -> str:
|
|
claims: Final = json.dumps(
|
|
{
|
|
"token": f"ui-token-{unique_marker()}",
|
|
"user_id": f"e2e-session-{unique_marker()}",
|
|
"user_role": "proxy_admin",
|
|
"team_id": "litellm-dashboard",
|
|
"expires": expires_at.isoformat(),
|
|
}
|
|
)
|
|
nonce: Final = os.urandom(12)
|
|
sealed: Final = AESGCM(hashlib.sha256(SALT_KEY.encode()).digest()).encrypt(
|
|
nonce, claims.encode(), SESSION_TOKEN_PREFIX.encode()
|
|
)
|
|
return SESSION_TOKEN_PREFIX + base64.urlsafe_b64encode(nonce + sealed).decode().rstrip("=")
|
|
|
|
|
|
class TestSessionToken:
|
|
@pytest.mark.covers("other.auth.session_token.valid_allows")
|
|
@meta(
|
|
Subject(
|
|
domain=Domain.PROXY_AUTH,
|
|
)
|
|
)
|
|
def test_unexpired_session_token_reaches_admin_route(self, client: OtherClient) -> None:
|
|
token: Final = _admin_session_token(datetime.now(timezone.utc) + timedelta(minutes=10))
|
|
listing: Final = unwrap(client.list_users_as(token))
|
|
assert listing.total >= 0, f"an unexpired admin session token did not reach /user/list: {listing}"
|
|
|
|
@pytest.mark.covers("other.auth.session_token.expired_denied")
|
|
@meta(
|
|
Subject(
|
|
domain=Domain.PROXY_AUTH,
|
|
)
|
|
)
|
|
def test_expired_session_token_is_denied(self, client: OtherClient) -> None:
|
|
token: Final = _admin_session_token(datetime.now(timezone.utc) - timedelta(minutes=1))
|
|
result: Final = client.list_users_as(token)
|
|
assert isinstance(result, UnauthorizedError), f"an expired session token must get 401, got {result}"
|
|
assert "expired" in result.body.lower(), f"expected the expired-key error, got {result.body[:300]}"
|
|
|
|
@pytest.mark.covers("other.auth.session_token.encrypted_value_denied")
|
|
@meta(
|
|
Subject(
|
|
domain=Domain.PROXY_AUTH,
|
|
)
|
|
)
|
|
def test_encrypted_stored_value_is_not_a_bearer_token(
|
|
self, client: OtherClient, resources: ResourceManager
|
|
) -> None:
|
|
stored_value: Final = f'{{"token": "{unique_marker()}", "user_role": "proxy_admin"}}'
|
|
key: Final = client.proxy.generate_key(
|
|
KeyGenerateBody(
|
|
key_alias=f"e2e-session-{unique_marker()}",
|
|
metadata=KeyMetadata(
|
|
logging=[
|
|
KeyLoggingCallback(
|
|
callback_name="langfuse",
|
|
callback_vars=KeyLoggingCallbackVars(langfuse_secret_key=stored_value),
|
|
)
|
|
]
|
|
),
|
|
)
|
|
)
|
|
resources.defer(lambda: client.proxy.delete_key(key))
|
|
|
|
metadata: Final = client.proxy.key_info(key).metadata
|
|
assert metadata is not None and metadata.logging, f"/key/info dropped the logging metadata: {metadata}"
|
|
encrypted: Final = metadata.logging[0].callback_vars.langfuse_secret_key
|
|
assert encrypted is not None and encrypted.startswith(ENCRYPTED_PREFIX), (
|
|
f"expected /key/info to return the stored secret encrypted, got {encrypted!r}"
|
|
)
|
|
|
|
for bearer in (encrypted.removeprefix(ENCRYPTED_PREFIX), encrypted):
|
|
result = client.list_users_as(bearer)
|
|
assert isinstance(result, UnauthorizedError), f"an encrypted stored value must get 401, got {result}"
|