mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-16 23:41:43 +00:00
test(auth_v2): cover SCIM scim:write guard and SAML RelayState redirect
Follow the auth module updates: SCIM routes now require the scim:write scope,
and the SAML ACS/login flow redirects to a validated RelayState instead of
returning JSON.
- scim: authenticate every request with a scoped key, and pin the guard
directly: no credential -> 401 with WWW-Authenticate, an authenticated
principal without scim:write -> 403 insufficient_scope
- saml: assert ACS returns 303 to a safe RelayState ("/dashboard") and falls
back to default_redirect_path for an absolute/"//host" RelayState; assert
GET /login threads ?next= through as a validated RelayState; add a garbage
SAMLResponse -> 401 case
A mutation spot-check confirmed the open-redirect tests fail when the
_safe_relay_state guard is bypassed.
This commit is contained in:
parent
d3878ee9a9
commit
7cf35ccc0a
2 changed files with 111 additions and 5 deletions
|
|
@ -287,6 +287,67 @@ def test_acs_missing_response_is_rejected(saml_env):
|
|||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_acs_rejects_garbage_response(saml_env):
|
||||
app, store = _build_app(saml_env)
|
||||
client = TestClient(app)
|
||||
response = client.post(
|
||||
"/auth/saml/acs",
|
||||
data={"SAMLResponse": "this-is-not-a-saml-response"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert response.status_code == 401
|
||||
assert store._users == {}
|
||||
|
||||
|
||||
def test_acs_redirects_to_safe_relay_state(saml_env):
|
||||
app, _ = _build_app(saml_env)
|
||||
client = TestClient(app)
|
||||
acs = client.post(
|
||||
"/auth/saml/acs",
|
||||
data={"SAMLResponse": saml_env.mint_response(), "RelayState": "/dashboard"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert acs.status_code == 303
|
||||
assert acs.headers["location"] == "/dashboard"
|
||||
|
||||
|
||||
def test_acs_rejects_open_redirect_relay_state(saml_env):
|
||||
app, _ = _build_app(saml_env)
|
||||
client = TestClient(app)
|
||||
acs = client.post(
|
||||
"/auth/saml/acs",
|
||||
data={
|
||||
"SAMLResponse": saml_env.mint_response(),
|
||||
"RelayState": "https://evil.example.com/phish",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert acs.status_code == 303
|
||||
# unsafe RelayState falls back to default_redirect_path, never the attacker URL
|
||||
assert acs.headers["location"] == "/"
|
||||
|
||||
|
||||
def test_login_threads_safe_next_as_relay_state(saml_env):
|
||||
app, _ = _build_app(saml_env)
|
||||
client = TestClient(app)
|
||||
response = client.get("/auth/saml/login?next=/dashboard", follow_redirects=False)
|
||||
assert response.status_code == 303
|
||||
assert "RelayState=%2Fdashboard" in response.headers["location"]
|
||||
|
||||
|
||||
def test_login_rejects_open_redirect_next(saml_env):
|
||||
app, _ = _build_app(saml_env)
|
||||
client = TestClient(app)
|
||||
response = client.get(
|
||||
"/auth/saml/login?next=https://evil.example.com", follow_redirects=False
|
||||
)
|
||||
assert response.status_code == 303
|
||||
location = response.headers["location"]
|
||||
assert "evil.example.com" not in location
|
||||
# falls back to default_redirect_path ("/") as the RelayState
|
||||
assert "RelayState=%2F&" in location or location.endswith("RelayState=%2F")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Pure helpers (no xmlsec1 required) - attribute mapping + open-redirect guard
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
|
|
|||
|
|
@ -5,26 +5,49 @@ from fastapi import FastAPI
|
|||
from fastapi.testclient import TestClient
|
||||
|
||||
from litellm.auth_v2.config import AuthConfig
|
||||
from litellm.auth_v2.resolver import InMemoryIdentityStore
|
||||
from litellm.auth_v2.models import AuthMethod, Principal, PrincipalType
|
||||
from litellm.auth_v2.resolver import InMemoryIdentityStore, _hash_api_key
|
||||
from litellm.auth_v2.security import install_auth
|
||||
|
||||
USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User"
|
||||
GROUP_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:Group"
|
||||
ERROR_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:Error"
|
||||
|
||||
SCIM_KEY = "sk-scim-writer"
|
||||
NOSCOPE_KEY = "sk-no-scim-scope"
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> TestClient:
|
||||
|
||||
def _principal(subject: str, scopes: list) -> Principal:
|
||||
return Principal(
|
||||
principal_type=PrincipalType.HUMAN,
|
||||
subject=subject,
|
||||
auth_method=AuthMethod.API_KEY,
|
||||
scopes=scopes,
|
||||
)
|
||||
|
||||
|
||||
def _app() -> FastAPI:
|
||||
app = FastAPI()
|
||||
install_auth(
|
||||
app,
|
||||
AuthConfig(),
|
||||
InMemoryIdentityStore(),
|
||||
InMemoryIdentityStore(
|
||||
api_keys={
|
||||
_hash_api_key(SCIM_KEY): _principal("scim-writer", ["scim:write"]),
|
||||
_hash_api_key(NOSCOPE_KEY): _principal("no-scope", []),
|
||||
}
|
||||
),
|
||||
mount_scim=True,
|
||||
mount_oidc=False,
|
||||
mount_saml=False,
|
||||
)
|
||||
return TestClient(app)
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> TestClient:
|
||||
# SCIM routes require scim:write; authenticate every request with a scoped key
|
||||
return TestClient(_app(), headers={"x-litellm-api-key": SCIM_KEY})
|
||||
|
||||
|
||||
def _create_user(client: TestClient, user_name="alice@example.com", display="Alice"):
|
||||
|
|
@ -144,3 +167,25 @@ def test_schemas_endpoint_returns_user_and_group(client):
|
|||
response = client.get("/scim/v2/Schemas")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["totalResults"] == 2
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# SCIM routes are gated by scim:write (design section 11)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_scim_requires_authentication():
|
||||
unauth = TestClient(_app())
|
||||
response = unauth.post(
|
||||
"/scim/v2/Users",
|
||||
json={"schemas": [USER_SCHEMA], "userName": "x@example.com"},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
assert "WWW-Authenticate" in response.headers
|
||||
|
||||
|
||||
def test_scim_requires_scim_write_scope():
|
||||
underscoped = TestClient(_app(), headers={"x-litellm-api-key": NOSCOPE_KEY})
|
||||
response = underscoped.get("/scim/v2/Users")
|
||||
assert response.status_code == 403
|
||||
assert "insufficient_scope" in response.headers.get("WWW-Authenticate", "")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue