From a7f9d5d6b988531930a1cd5756c6cee32246bb42 Mon Sep 17 00:00:00 2001 From: Yassin Kortam Date: Wed, 10 Jun 2026 20:29:03 -0700 Subject: [PATCH] test(auth_v2): pin the mTLS forwarded-DN gate against XFF spoofing The forwarded subject-DN trust gate keys on the raw socket peer and prefers a verified TLS-layer cert: - an untrusted peer cannot smuggle a forged DN by claiming a trusted address via X-Forwarded-For (the gate ignores XFF) - a verified client cert from the ASGI TLS extension wins over a proxy-forwarded DN header Full auth_v2 suite: 175 passing. --- .../proxy/auth_v2/test_authenticators.py | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tests/test_litellm/proxy/auth_v2/test_authenticators.py b/tests/test_litellm/proxy/auth_v2/test_authenticators.py index 1a635cf0bdc..7950649c06a 100644 --- a/tests/test_litellm/proxy/auth_v2/test_authenticators.py +++ b/tests/test_litellm/proxy/auth_v2/test_authenticators.py @@ -436,6 +436,30 @@ async def test_mtls_forwarded_header_from_untrusted_peer_is_ignored(): assert await auth.authenticate(request) is None +async def test_mtls_forwarded_header_gate_ignores_spoofed_xff(): + # the gate keys on the raw socket peer, not X-Forwarded-For: an untrusted peer + # cannot claim a trusted address via XFF to smuggle a forged DN header + auth = _mtls(MutualTLSConfig(enabled=True, forwarded_subject_header="x-client-dn")) + request = make_request( + headers={"x-client-dn": "CN=attacker", "x-forwarded-for": "10.0.0.5"}, + client=("8.8.8.8", 4444), + ) + assert await auth.authenticate(request) is None + + +async def test_mtls_prefers_verified_asgi_cert_over_forwarded_header(): + # a genuinely verified client cert from the TLS layer wins over a proxy header + auth = _mtls(MutualTLSConfig(enabled=True, forwarded_subject_header="x-client-dn")) + request = make_request( + headers={"x-client-dn": "CN=from-header"}, + client=("10.0.0.9", 1), + scope_extra={"extensions": {"tls": {"client_cert_name": "CN=from-tls"}}}, + ) + credential = await auth.authenticate(request) + assert credential is not None + assert credential.subject == "CN=from-tls" + + async def test_mtls_forwarded_header_absent_returns_none(): auth = _mtls(MutualTLSConfig(enabled=True, forwarded_subject_header="x-client-dn")) assert await auth.authenticate(make_request()) is None