authentik.providers.saml.tests.test_auth_n_request

  1"""Test AuthN Request generator and parser"""
  2
  3from base64 import b64encode
  4
  5from defusedxml.lxml import fromstring
  6from django.http.request import QueryDict
  7from django.test import TestCase
  8from guardian.utils import get_anonymous_user
  9from lxml import etree  # nosec
 10
 11from authentik.blueprints.tests import apply_blueprint
 12from authentik.common.saml.constants import (
 13    NS_MAP,
 14    SAML_BINDING_POST,
 15    SAML_NAME_ID_FORMAT_EMAIL,
 16    SAML_NAME_ID_FORMAT_UNSPECIFIED,
 17)
 18from authentik.core.models import Application
 19from authentik.core.tests.utils import (
 20    RequestFactory,
 21    create_test_admin_user,
 22    create_test_cert,
 23    create_test_flow,
 24)
 25from authentik.crypto.models import CertificateKeyPair
 26from authentik.events.models import Event, EventAction
 27from authentik.lib.generators import generate_id
 28from authentik.lib.xml import lxml_from_string
 29from authentik.providers.saml.models import SAMLPropertyMapping, SAMLProvider
 30from authentik.providers.saml.processors.assertion import AssertionProcessor
 31from authentik.providers.saml.processors.authn_request_parser import AuthNRequestParser
 32from authentik.sources.saml.exceptions import MismatchedRequestID
 33from authentik.sources.saml.models import SAMLBindingTypes, SAMLSource
 34from authentik.sources.saml.processors.request import SESSION_KEY_REQUEST_ID, RequestProcessor
 35from authentik.sources.saml.processors.response import ResponseProcessor
 36
 37POST_REQUEST = (
 38    "PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiPz48c2FtbDJwOkF1dGhuUmVxdWVzdCB4bWxuczpzYW1sMn"
 39    "A9InVybjpvYXNpczpuYW1lczp0YzpTQU1MOjIuMDpwcm90b2NvbCIgQXNzZXJ0aW9uQ29uc3VtZXJTZXJ2aWNlVVJMPSJo"
 40    "dHRwczovL2V1LWNlbnRyYWwtMS5zaWduaW4uYXdzLmFtYXpvbi5jb20vcGxhdGZvcm0vc2FtbC9hY3MvMmQ3MzdmOTYtNT"
 41    "VmYi00MDM1LTk1M2UtNWUyNDEzNGViNzc4IiBEZXN0aW5hdGlvbj0iaHR0cHM6Ly9pZC5iZXJ5anUub3JnL2FwcGxpY2F0"
 42    "aW9uL3NhbWwvYXdzLXNzby9zc28vYmluZGluZy9wb3N0LyIgSUQ9ImF3c19MRHhMR2V1YnBjNWx4MTJneENnUzZ1UGJpeD"
 43    "F5ZDVyZSIgSXNzdWVJbnN0YW50PSIyMDIxLTA3LTA2VDE0OjIzOjA2LjM4OFoiIFByb3RvY29sQmluZGluZz0idXJuOm9h"
 44    "c2lzOm5hbWVzOnRjOlNBTUw6Mi4wOmJpbmRpbmdzOkhUVFAtUE9TVCIgVmVyc2lvbj0iMi4wIj48c2FtbDI6SXNzdWVyIH"
 45    "htbG5zOnNhbWwyPSJ1cm46b2FzaXM6bmFtZXM6dGM6U0FNTDoyLjA6YXNzZXJ0aW9uIj5odHRwczovL2V1LWNlbnRyYWwt"
 46    "MS5zaWduaW4uYXdzLmFtYXpvbi5jb20vcGxhdGZvcm0vc2FtbC9kLTk5NjcyZjgyNzg8L3NhbWwyOklzc3Vlcj48c2FtbD"
 47    "JwOk5hbWVJRFBvbGljeSBGb3JtYXQ9InVybjpvYXNpczpuYW1lczp0YzpTQU1MOjEuMTpuYW1laWQtZm9ybWF0OmVtYWls"
 48    "QWRkcmVzcyIvPjwvc2FtbDJwOkF1dGhuUmVxdWVzdD4="
 49)
 50REDIRECT_REQUEST = (
 51    "fZLNbsIwEIRfJfIdbKeFgEUipXAoEm0jSHvopTLJplhK7NTr9Oft6yRUKhekPdk73+yOdoWyqVuRdu6k9/DRAbrgu6k1iu"
 52    "EjJp3VwkhUKLRsAIUrxCF92IlwykRrjTOFqUmQIoJ1yui10dg1YA9gP1UBz/tdTE7OtSgo5WzKQzYditGeP8GW9rSQZk+H"
 53    "nAQbb6+07EGj7EI1j8SCeaVs21oVQ9dAoRqcf6OIhh6VLpV+pxZKZaFwlATbTUzeyqKazaqiDCO5WEQwZzKCagkwr8obWc"
 54    "qjb0PsYKvRSe1iErKQTTj3lYdc3HLBl68kyM4L340u19M5j4LiPs+zybjgC1gclvMNJFn104vB2P5L/TpW/kVNkqvBrug/"
 55    "+mjVikeP224y4/P7CdK6Nl9rC9JBTDihySi5vIbkFw=="
 56)
 57REDIRECT_SIGNATURE = (
 58    "UlOe1BItHVHM+io6rUZAenIqfibm7hM6wr9I1rcP5kPJ4N8cbkyqmAMh5LD2lUq3PDERJfjdO/oOKnvJmbD2y9MOObyR2d"
 59    "7Udv62KERrA0qM917Q+w8wrLX7w2nHY96EDvkXD4iAomR5EE9dHRuubDy7uRv2syEevc0gfoLi7W/5vp96vJgsaSqxnTp+"
 60    "QiYq49KyWyMtxRULF2yd+vYDnHCDME73mNSULEHfwCU71dvbKpnFaej78q7wS20gUk6ysOOXXtvDHbiVcpUb/9oyDgNAxU"
 61    "jVvPdh96AhBFj2HCuGZhP0CGotafTciu6YlsiwUpuBkIYgZmNWYa3FR9LS4Q=="
 62)
 63REDIRECT_SIG_ALG = "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256"
 64REDIRECT_RELAY_STATE = "ss:mem:7a054b4af44f34f89dd2d973f383c250b6b076e7f06cfa8276008a6504eaf3c7"
 65REDIRECT_CERT = """-----BEGIN CERTIFICATE-----
 66MIIDCDCCAfCgAwIBAgIRAM5s+bhOHk4ChSpPkGSh0NswDQYJKoZIhvcNAQELBQAw
 67KzEpMCcGA1UEAwwgcGFzc2Jvb2sgU2VsZi1zaWduZWQgQ2VydGlmaWNhdGUwHhcN
 68MjAxMTA3MjAzNDIxWhcNMjExMTA4MjAzNDIxWjBUMSkwJwYDVQQDDCBwYXNzYm9v
 69ayBTZWxmLXNpZ25lZCBDZXJ0aWZpY2F0ZTERMA8GA1UECgwIcGFzc2Jvb2sxFDAS
 70BgNVBAsMC1NlbGYtc2lnbmVkMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKC
 71AQEAuh+Bv6a/ogpic72X/sq86YiLzVjixnGqjc4wpsPPP00GX8jUAZJL4Tjo+sYK
 72IU2DF2/azlVqjkbLho4rGuuc8YkbFXBEXPYc5h3bseO2vk6sbbbWKV0mro1VFhBh
 73T59hBORuMMefmQdhFzsRNOGklIptQdg0quD8ET3+/uNfIT98S2ruZdYteFls46Sa
 74MokZFYVD6pWEYV4P2MKVAFqJX9bqBW0LfCCfFqHAOJjUZj9dtleg86d2WfedUOG2
 75LK0iLrydjhThbI0GUDhv0jWYkRlv04fdJ1WSRANYA3gBOnyw+Iigh2xNnYbVZMXT
 76I0BupIJ4UoODMc4QpD2GYJ6oGwIDAQABMA0GCSqGSIb3DQEBCwUAA4IBAQCCEF3e
 77Y99KxEBSR4H4/TvKbnh4QtHswOf7MaGdjtrld7l4u4Hc4NEklNdDn1XLKhZwnq3Z
 78LRsRlJutDzZ18SRmAJPXPbka7z7D+LA1mbNQElOgiKyQHD9rIJSBr6X5SM9As3CR
 797QUsb8dg7kc+Jn7WuLZIEVxxMtekt0buWEdMJiklF0tCS3LNsP083FaQk/H1K0z6
 803PWP26EFdwir3RyTKLY5CBLjKrUAo9O1l/WBVFYbdetnipbGGu5f6nk6nnxbwLLI
 81Dm52Vkq+xFDDUq9IqIoYvLaE86MDvtpMQEx65tIGU19vUf3fL/+sSfdRZ1HDzP4d
 82qNAZMq1DqpibfCBg
 83-----END CERTIFICATE-----"""
 84
 85
 86class TestAuthNRequest(TestCase):
 87    """Test AuthN Request generator and parser"""
 88
 89    @apply_blueprint("system/providers-saml.yaml")
 90    def setUp(self):
 91        self.request_factory = RequestFactory()
 92        self.cert = create_test_cert()
 93        self.provider: SAMLProvider = SAMLProvider.objects.create(
 94            authorization_flow=create_test_flow(),
 95            acs_url="http://testserver/source/saml/provider/acs/",
 96            signing_kp=self.cert,
 97            verification_kp=self.cert,
 98        )
 99        self.provider.property_mappings.set(SAMLPropertyMapping.objects.all())
100        self.provider.save()
101        Application.objects.create(
102            name="test-app",
103            slug="test-app",
104            provider=self.provider,
105        )
106        self.source = SAMLSource.objects.create(
107            slug="provider",
108            issuer="authentik",
109            pre_authentication_flow=create_test_flow(),
110            signing_kp=self.cert,
111            verification_kp=self.cert,
112            signed_assertion=True,
113            binding_type=SAMLBindingTypes.POST,
114        )
115
116    def test_signed_valid(self):
117        """Test generated AuthNRequest with valid signature"""
118        http_request = self.request_factory.get("/")
119
120        # First create an AuthNRequest
121        request_proc = RequestProcessor(self.source, http_request, "test_state")
122        auth_n = request_proc.get_auth_n()
123        self.assertEqual(auth_n.attrib["ProtocolBinding"], SAML_BINDING_POST)
124
125        request = request_proc.build_auth_n()
126        # Now we check the ID and signature
127        parsed_request = AuthNRequestParser(self.provider).parse(
128            b64encode(request.encode()).decode(), "test_state"
129        )
130        self.assertEqual(parsed_request.id, request_proc.request_id)
131        self.assertEqual(parsed_request.relay_state, "test_state")
132
133    def test_request_encrypt(self):
134        """Test full SAML Request/Response flow, fully encrypted"""
135        self.provider.encryption_kp = self.cert
136        self.provider.save()
137        self.source.encryption_kp = self.cert
138        self.source.save()
139        http_request = self.request_factory.get("/", user=get_anonymous_user())
140
141        # First create an AuthNRequest
142        request_proc = RequestProcessor(self.source, http_request, "test_state")
143        request = request_proc.build_auth_n()
144
145        # To get an assertion we need a parsed request (parsed by provider)
146        parsed_request = AuthNRequestParser(self.provider).parse(
147            b64encode(request.encode()).decode(), "test_state"
148        )
149        # Now create a response and convert it to string (provider)
150        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
151        response = response_proc.build_response()
152
153        # Now parse the response (source)
154        http_request.POST = QueryDict(mutable=True)
155        http_request.POST["SAMLResponse"] = b64encode(response.encode()).decode()
156
157        response_parser = ResponseProcessor(self.source, http_request)
158        response_parser.parse()
159
160    def test_request_encrypt_cert_only(self):
161        """Test SAML encryption with certificate-only keypair (no private key).
162
163        This tests the scenario where the IdP (provider) only has the SP's public
164        certificate for encryption, without a private key. This is the expected
165        real-world scenario since the SP would never share their private key.
166        """
167        # Create a full keypair for the source (SP) - it needs the private key to decrypt
168        full_keypair = create_test_cert()
169
170        # Create a certificate-only keypair for the provider (IdP)
171        # This simulates having only the SP's public certificate
172        cert_only = CertificateKeyPair.objects.create(
173            name=generate_id(),
174            certificate_data=full_keypair.certificate_data,
175            key_data="",  # No private key
176        )
177
178        self.provider.encryption_kp = cert_only
179        self.provider.save()
180        self.source.encryption_kp = full_keypair
181        self.source.save()
182        http_request = self.request_factory.get("/", user=get_anonymous_user())
183
184        # First create an AuthNRequest
185        request_proc = RequestProcessor(self.source, http_request, "test_state")
186        request = request_proc.build_auth_n()
187
188        # To get an assertion we need a parsed request (parsed by provider)
189        parsed_request = AuthNRequestParser(self.provider).parse(
190            b64encode(request.encode()).decode(), "test_state"
191        )
192        # Now create a response and convert it to string (provider)
193        # This should work with only the certificate (public key) for encryption
194        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
195        response = response_proc.build_response()
196
197        # Now parse the response (source) - decryption requires the private key
198        http_request.POST = QueryDict(mutable=True)
199        http_request.POST["SAMLResponse"] = b64encode(response.encode()).decode()
200
201        response_parser = ResponseProcessor(self.source, http_request)
202        response_parser.parse()
203
204    def test_request_sign_response_and_encrypt(self):
205        """Test SAML with sign_response enabled AND encryption.
206
207        This tests the fix for signature invalidation when encryption is enabled.
208        The response must be signed AFTER encryption, not before, because encryption
209        replaces the Assertion with EncryptedAssertion which changes the response content.
210        """
211        self.provider.sign_response = True
212        self.provider.sign_assertion = False
213        self.provider.encryption_kp = self.cert
214        self.provider.save()
215        self.source.encryption_kp = self.cert
216        self.source.signed_response = True
217        self.source.signed_assertion = False  # Only response is signed, not assertion
218        self.source.save()
219        http_request = self.request_factory.get("/", user=get_anonymous_user())
220
221        # First create an AuthNRequest
222        request_proc = RequestProcessor(self.source, http_request, "test_state")
223        request = request_proc.build_auth_n()
224
225        # To get an assertion we need a parsed request (parsed by provider)
226        parsed_request = AuthNRequestParser(self.provider).parse(
227            b64encode(request.encode()).decode(), "test_state"
228        )
229        # Now create a response and convert it to string (provider)
230        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
231        response = response_proc.build_response()
232
233        # Verify the response contains EncryptedAssertion and a signature
234        response_xml = fromstring(response)
235        self.assertEqual(len(response_xml.xpath("//saml:EncryptedAssertion", namespaces=NS_MAP)), 1)
236        self.assertEqual(
237            len(response_xml.xpath("//samlp:Response/ds:Signature", namespaces=NS_MAP)), 1
238        )
239
240        # Now parse the response (source) - this will verify the signature and decrypt
241        http_request.POST = QueryDict(mutable=True)
242        http_request.POST["SAMLResponse"] = b64encode(response.encode()).decode()
243
244        response_parser = ResponseProcessor(self.source, http_request)
245        response_parser.parse()
246
247    def test_request_sign_assertion_and_encrypt(self):
248        """Test SAML with sign_assertion enabled AND encryption.
249
250        The assertion signature should be inside the encrypted content and
251        remain valid after decryption.
252        """
253        self.provider.sign_response = False
254        self.provider.sign_assertion = True
255        self.provider.encryption_kp = self.cert
256        self.provider.save()
257        self.source.encryption_kp = self.cert
258        self.source.signed_assertion = True
259        self.source.save()
260        http_request = self.request_factory.get("/", user=get_anonymous_user())
261
262        # First create an AuthNRequest
263        request_proc = RequestProcessor(self.source, http_request, "test_state")
264        request = request_proc.build_auth_n()
265
266        # To get an assertion we need a parsed request (parsed by provider)
267        parsed_request = AuthNRequestParser(self.provider).parse(
268            b64encode(request.encode()).decode(), "test_state"
269        )
270        # Now create a response and convert it to string (provider)
271        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
272        response = response_proc.build_response()
273
274        # Verify the response contains EncryptedAssertion
275        response_xml = fromstring(response)
276        self.assertEqual(len(response_xml.xpath("//saml:EncryptedAssertion", namespaces=NS_MAP)), 1)
277
278        # Now parse the response (source) - this will decrypt and verify assertion signature
279        http_request.POST = QueryDict(mutable=True)
280        http_request.POST["SAMLResponse"] = b64encode(response.encode()).decode()
281
282        response_parser = ResponseProcessor(self.source, http_request)
283        response_parser.parse()
284
285    def test_request_sign_both_and_encrypt(self):
286        """Test SAML with both sign_assertion and sign_response enabled AND encryption.
287
288        This is the most complex scenario: assertion is signed, then encrypted,
289        then the response is signed. Both signatures should be valid.
290        """
291        self.provider.sign_response = True
292        self.provider.sign_assertion = True
293        self.provider.encryption_kp = self.cert
294        self.provider.save()
295        self.source.encryption_kp = self.cert
296        self.source.signed_assertion = True
297        self.source.signed_response = True
298        self.source.save()
299        http_request = self.request_factory.get("/", user=get_anonymous_user())
300
301        # First create an AuthNRequest
302        request_proc = RequestProcessor(self.source, http_request, "test_state")
303        request = request_proc.build_auth_n()
304
305        # To get an assertion we need a parsed request (parsed by provider)
306        parsed_request = AuthNRequestParser(self.provider).parse(
307            b64encode(request.encode()).decode(), "test_state"
308        )
309        # Now create a response and convert it to string (provider)
310        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
311        response = response_proc.build_response()
312
313        # Verify the response contains EncryptedAssertion and response signature
314        response_xml = fromstring(response)
315        self.assertEqual(len(response_xml.xpath("//saml:EncryptedAssertion", namespaces=NS_MAP)), 1)
316        self.assertEqual(
317            len(response_xml.xpath("//samlp:Response/ds:Signature", namespaces=NS_MAP)), 1
318        )
319
320        # Now parse the response (source) - this will verify response signature,
321        # decrypt, then verify assertion signature
322        http_request.POST = QueryDict(mutable=True)
323        http_request.POST["SAMLResponse"] = b64encode(response.encode()).decode()
324
325        response_parser = ResponseProcessor(self.source, http_request)
326        response_parser.parse()
327
328    def test_encrypted_assertion_namespace_preservation(self):
329        """Test that encrypted assertions include namespace declarations.
330
331        When an assertion is encrypted, the resulting decrypted XML must include
332        the necessary namespace declarations (xmlns:saml) since it's now a standalone
333        document fragment, no longer inheriting namespaces from the parent Response.
334        """
335        self.provider.encryption_kp = self.cert
336        self.provider.save()
337        self.source.encryption_kp = self.cert
338        self.source.save()
339        http_request = self.request_factory.get("/", user=get_anonymous_user())
340
341        # First create an AuthNRequest
342        request_proc = RequestProcessor(self.source, http_request, "test_state")
343        request = request_proc.build_auth_n()
344
345        # To get an assertion we need a parsed request (parsed by provider)
346        parsed_request = AuthNRequestParser(self.provider).parse(
347            b64encode(request.encode()).decode(), "test_state"
348        )
349        # Now create a response and convert it to string (provider)
350        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
351        response = response_proc.build_response()
352
353        # Parse the encrypted response
354        response_xml = fromstring(response)
355        encrypted_assertion = response_xml.xpath("//saml:EncryptedAssertion", namespaces=NS_MAP)[0]
356        encrypted_data = encrypted_assertion.xpath("//xenc:EncryptedData", namespaces=NS_MAP)[0]
357
358        # Decrypt the assertion manually to verify namespace is present
359        import xmlsec
360
361        manager = xmlsec.KeysManager()
362        key = xmlsec.Key.from_memory(self.cert.key_data, xmlsec.constants.KeyDataFormatPem, None)
363        manager.add_key(key)
364        enc_ctx = xmlsec.EncryptionContext(manager)
365        decrypted = enc_ctx.decrypt(encrypted_data)
366
367        # The decrypted assertion should have xmlns:saml namespace declaration
368        decrypted_str = etree.tostring(decrypted).decode()
369        self.assertIn("xmlns:saml", decrypted_str)
370
371        # Also verify full round-trip works (source can parse it)
372        http_request.POST = QueryDict(mutable=True)
373        http_request.POST["SAMLResponse"] = b64encode(response.encode()).decode()
374
375        response_parser = ResponseProcessor(self.source, http_request)
376        response_parser.parse()
377
378    def test_encrypted_response_schema_validation(self):
379        """Test that encrypted SAML responses validate against the SAML schema.
380
381        The response with EncryptedAssertion must be valid per saml-schema-protocol-2.0.xsd.
382        This ensures we don't have invalid elements like EncryptedData inside Assertion.
383        """
384        self.provider.encryption_kp = self.cert
385        self.provider.save()
386        http_request = self.request_factory.get("/", user=get_anonymous_user())
387
388        # First create an AuthNRequest
389        request_proc = RequestProcessor(self.source, http_request, "test_state")
390        request = request_proc.build_auth_n()
391
392        # To get an assertion we need a parsed request (parsed by provider)
393        parsed_request = AuthNRequestParser(self.provider).parse(
394            b64encode(request.encode()).decode(), "test_state"
395        )
396        # Now create a response and convert it to string (provider)
397        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
398        response = response_proc.build_response()
399
400        # Validate against SAML schema
401        schema = etree.XMLSchema(
402            etree.parse("schemas/saml-schema-protocol-2.0.xsd", parser=etree.XMLParser())  # nosec
403        )
404        self.assertTrue(schema.validate(lxml_from_string(response)))
405
406        # Verify structure: should have EncryptedAssertion, not Assertion with EncryptedData inside
407        response_xml = fromstring(response)
408        self.assertEqual(len(response_xml.xpath("//saml:EncryptedAssertion", namespaces=NS_MAP)), 1)
409        self.assertEqual(len(response_xml.xpath("//saml:Assertion", namespaces=NS_MAP)), 0)
410
411    def test_request_signed(self):
412        """Test full SAML Request/Response flow, fully signed"""
413        http_request = self.request_factory.get("/", user=get_anonymous_user())
414
415        # First create an AuthNRequest
416        request_proc = RequestProcessor(self.source, http_request, "test_state")
417        request = request_proc.build_auth_n()
418
419        # To get an assertion we need a parsed request (parsed by provider)
420        parsed_request = AuthNRequestParser(self.provider).parse(
421            b64encode(request.encode()).decode(), "test_state"
422        )
423        # Now create a response and convert it to string (provider)
424        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
425        response = response_proc.build_response()
426
427        # Now parse the response (source)
428        http_request.POST = QueryDict(mutable=True)
429        http_request.POST["SAMLResponse"] = b64encode(response.encode()).decode()
430
431        response_parser = ResponseProcessor(self.source, http_request)
432        response_parser.parse()
433
434    def test_request_signed_both(self):
435        """Test full SAML Request/Response flow, fully signed"""
436        self.provider.sign_assertion = True
437        self.provider.sign_response = True
438        self.provider.save()
439        self.source.signed_response = True
440        http_request = self.request_factory.get("/", user=get_anonymous_user())
441
442        # First create an AuthNRequest
443        request_proc = RequestProcessor(self.source, http_request, "test_state")
444        request = request_proc.build_auth_n()
445
446        # To get an assertion we need a parsed request (parsed by provider)
447        parsed_request = AuthNRequestParser(self.provider).parse(
448            b64encode(request.encode()).decode(), "test_state"
449        )
450        # Now create a response and convert it to string (provider)
451        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
452        response = response_proc.build_response()
453        # Ensure both response and assertion ID are in the response twice (once as ID attribute,
454        # once as ds:Reference URI)
455        self.assertEqual(response.count(response_proc._assertion_id), 2)
456        self.assertEqual(response.count(response_proc._response_id), 2)
457
458        schema = etree.XMLSchema(
459            etree.parse("schemas/saml-schema-protocol-2.0.xsd", parser=etree.XMLParser())  # nosec
460        )
461        self.assertTrue(schema.validate(lxml_from_string(response)))
462
463        response_xml = fromstring(response)
464        self.assertEqual(
465            len(response_xml.xpath("//saml:Assertion/ds:Signature", namespaces=NS_MAP)), 1
466        )
467        self.assertEqual(
468            len(response_xml.xpath("//samlp:Response/ds:Signature", namespaces=NS_MAP)), 1
469        )
470
471        # Now parse the response (source)
472        http_request.POST = QueryDict(mutable=True)
473        http_request.POST["SAMLResponse"] = b64encode(response.encode()).decode()
474
475        response_parser = ResponseProcessor(self.source, http_request)
476        response_parser.parse()
477
478    def test_request_id_invalid(self):
479        """Test generated AuthNRequest with invalid request ID"""
480        http_request = self.request_factory.get("/", user=get_anonymous_user())
481
482        # First create an AuthNRequest
483        request_proc = RequestProcessor(self.source, http_request, "test_state")
484        request = request_proc.build_auth_n()
485
486        # change the request ID
487        http_request.session[SESSION_KEY_REQUEST_ID] = "test"
488        http_request.session.save()
489
490        # To get an assertion we need a parsed request (parsed by provider)
491        parsed_request = AuthNRequestParser(self.provider).parse(
492            b64encode(request.encode()).decode(), "test_state"
493        )
494        # Now create a response and convert it to string (provider)
495        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
496        response = response_proc.build_response()
497
498        # Now parse the response (source)
499        http_request.POST = QueryDict(mutable=True)
500        http_request.POST["SAMLResponse"] = b64encode(response.encode()).decode()
501
502        response_parser = ResponseProcessor(self.source, http_request)
503
504        with self.assertRaises(MismatchedRequestID):
505            response_parser.parse()
506
507    def test_signed_valid_detached(self):
508        """Test generated AuthNRequest with valid signature (detached)"""
509        http_request = self.request_factory.get("/")
510
511        # First create an AuthNRequest
512        request_proc = RequestProcessor(self.source, http_request, "test_state")
513        params = request_proc.build_auth_n_detached()
514        # Now we check the ID and signature
515        parsed_request = AuthNRequestParser(self.provider).parse_detached(
516            params["SAMLRequest"],
517            params["RelayState"],
518            params["Signature"],
519            params["SigAlg"],
520        )
521        self.assertEqual(parsed_request.id, request_proc.request_id)
522        self.assertEqual(parsed_request.relay_state, "test_state")
523
524    def test_signed_detached_static(self):
525        """Test request with detached signature,
526        taken from https://www.samltool.com/generic_sso_req.php"""
527        static_keypair = CertificateKeyPair.objects.create(
528            name="samltool", certificate_data=REDIRECT_CERT
529        )
530        provider = SAMLProvider(
531            name="samltool",
532            authorization_flow=create_test_flow(),
533            acs_url="https://10.120.20.200/saml-sp/SAML2/POST",
534            audience="https://10.120.20.200/saml-sp/SAML2/POST",
535            issuer_override="https://10.120.20.200/saml-sp/SAML2/POST",
536            signing_kp=static_keypair,
537            verification_kp=static_keypair,
538        )
539        parsed_request = AuthNRequestParser(provider).parse_detached(
540            REDIRECT_REQUEST, REDIRECT_RELAY_STATE, REDIRECT_SIGNATURE, REDIRECT_SIG_ALG
541        )
542        self.assertEqual(parsed_request.id, "_dcf55fcd27a887e60a7ef9ee6fd3adab")
543        self.assertEqual(parsed_request.name_id_policy, SAML_NAME_ID_FORMAT_UNSPECIFIED)
544        self.assertEqual(parsed_request.relay_state, REDIRECT_RELAY_STATE)
545
546    def test_signed_static(self):
547        """Test post request with static request"""
548        provider = SAMLProvider(
549            name="aws",
550            authorization_flow=create_test_flow(),
551            acs_url=(
552                "https://eu-central-1.signin.aws.amazon.com/platform/"
553                "saml/acs/2d737f96-55fb-4035-953e-5e24134eb778"
554            ),
555            audience="https://10.120.20.200/saml-sp/SAML2/POST",
556            issuer_override="https://10.120.20.200/saml-sp/SAML2/POST",
557            signing_kp=create_test_cert(),
558        )
559        parsed_request = AuthNRequestParser(provider).parse(POST_REQUEST)
560        self.assertEqual(parsed_request.id, "aws_LDxLGeubpc5lx12gxCgS6uPbix1yd5re")
561        self.assertEqual(parsed_request.name_id_policy, SAML_NAME_ID_FORMAT_EMAIL)
562
563    def test_authn_context_class_ref_mapping(self):
564        """Test custom authn_context_class_ref"""
565        authn_context_class_ref = generate_id()
566        mapping = SAMLPropertyMapping.objects.create(
567            name=generate_id(), expression=f"""return '{authn_context_class_ref}'"""
568        )
569        self.provider.authn_context_class_ref_mapping = mapping
570        self.provider.save()
571        user = create_test_admin_user()
572        http_request = self.request_factory.get("/", user=user)
573
574        # First create an AuthNRequest
575        request_proc = RequestProcessor(self.source, http_request, "test_state")
576        request = request_proc.build_auth_n()
577
578        # To get an assertion we need a parsed request (parsed by provider)
579        parsed_request = AuthNRequestParser(self.provider).parse(
580            b64encode(request.encode()).decode(), "test_state"
581        )
582        # Now create a response and convert it to string (provider)
583        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
584        response = response_proc.build_response()
585        self.assertIn(user.username, response)
586        self.assertIn(authn_context_class_ref, response)
587
588    def test_authn_context_class_ref_mapping_invalid(self):
589        """Test custom authn_context_class_ref (invalid)"""
590        mapping = SAMLPropertyMapping.objects.create(name=generate_id(), expression="q")
591        self.provider.authn_context_class_ref_mapping = mapping
592        self.provider.save()
593        user = create_test_admin_user()
594        http_request = self.request_factory.get("/", user=user)
595
596        # First create an AuthNRequest
597        request_proc = RequestProcessor(self.source, http_request, "test_state")
598        request = request_proc.build_auth_n()
599
600        # To get an assertion we need a parsed request (parsed by provider)
601        parsed_request = AuthNRequestParser(self.provider).parse(
602            b64encode(request.encode()).decode(), "test_state"
603        )
604        # Now create a response and convert it to string (provider)
605        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
606        response = response_proc.build_response()
607        self.assertIn(user.username, response)
608
609        events = Event.objects.filter(
610            action=EventAction.CONFIGURATION_ERROR,
611        )
612        self.assertTrue(events.exists())
613        self.assertEqual(
614            events.first().context["message"],
615            f"Failed to evaluate property-mapping: '{mapping.name}'",
616        )
617
618    def test_request_attributes(self):
619        """Test full SAML Request/Response flow, fully signed"""
620        user = create_test_admin_user()
621        http_request = self.request_factory.get("/", user=user)
622
623        # First create an AuthNRequest
624        request_proc = RequestProcessor(self.source, http_request, "test_state")
625        request = request_proc.build_auth_n()
626
627        # To get an assertion we need a parsed request (parsed by provider)
628        parsed_request = AuthNRequestParser(self.provider).parse(
629            b64encode(request.encode()).decode(), "test_state"
630        )
631        # Now create a response and convert it to string (provider)
632        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
633        self.assertIn(user.username, response_proc.build_response())
634
635    def test_request_attributes_invalid(self):
636        """Test full SAML Request/Response flow, fully signed"""
637        user = create_test_admin_user()
638        http_request = self.request_factory.get("/", user=user)
639
640        # First create an AuthNRequest
641        request_proc = RequestProcessor(self.source, http_request, "test_state")
642        request = request_proc.build_auth_n()
643
644        # Create invalid PropertyMapping
645        mapping = SAMLPropertyMapping.objects.create(
646            name=generate_id(), saml_name="test", expression="q"
647        )
648        self.provider.property_mappings.add(mapping)
649
650        # To get an assertion we need a parsed request (parsed by provider)
651        parsed_request = AuthNRequestParser(self.provider).parse(
652            b64encode(request.encode()).decode(), "test_state"
653        )
654        # Now create a response and convert it to string (provider)
655        response_proc = AssertionProcessor(self.provider, http_request, parsed_request)
656        self.assertIn(user.username, response_proc.build_response())
657
658        events = Event.objects.filter(
659            action=EventAction.CONFIGURATION_ERROR,
660        )
661        self.assertTrue(events.exists())
662        self.assertEqual(
663            events.first().context["message"],
664            f"Failed to evaluate property-mapping: '{mapping.name}'",
665        )
666
667    def test_idp_initiated(self):
668        """Test IDP-initiated login"""
669        self.provider.default_relay_state = generate_id()
670        request = AuthNRequestParser(self.provider).idp_initiated()
671        self.assertEqual(request.id, None)
672        self.assertEqual(request.relay_state, self.provider.default_relay_state)