sources/saml: Basic support for EncryptedAssertion element.

authentik/providers/radius/api/providers.py

@@ -2,6 +2,7 @@
 
 from base64 import b64encode
 
+from django.conf import settings
 from django.shortcuts import get_object_or_404
 from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import OpenApiParameter, extend_schema
@@ -100,7 +101,16 @@ def get_attributes(self, provider: RadiusProvider):
             RadiusProviderPropertyMapping,
             ["packet"],
         )
-        dict = Dictionary("authentik/providers/radius/dictionaries/dictionary")
+        dict = Dictionary(
+            str(
+                settings.BASE_DIR
+                / "authentik"
+                / "providers"
+                / "radius"
+                / "dictionaries"
+                / "dictionary"
+            )
+        )
 
         packet = AuthPacket()
         packet.secret = provider.shared_secret

authentik/providers/saml/processors/assertion.py

@@ -18,7 +18,7 @@
 from authentik.providers.saml.utils import get_random_id
 from authentik.providers.saml.utils.time import get_time_string
 from authentik.sources.ldap.auth import LDAP_DISTINGUISHED_NAME
-from authentik.sources.saml.exceptions import UnsupportedNameIDFormat
+from authentik.sources.saml.exceptions import InvalidSignature, UnsupportedNameIDFormat
 from authentik.sources.saml.processors.constants import (
     DIGEST_ALGORITHM_TRANSLATION_MAP,
     NS_MAP,
@@ -318,6 +318,9 @@
                 xmlsec.constants.KeyDataFormatCertPem,
             )
             ctx.key = key
-            ctx.sign(signature_node)
+            try:
+                ctx.sign(signature_node)
+            except xmlsec.Error as exc:
+                raise InvalidSignature() from exc
 
         return etree.tostring(root_response).decode("utf-8")  # nosec

authentik/sources/saml/api/source.py

@@ -33,6 +33,7 @@ class Meta:
             "digest_algorithm",
             "signature_algorithm",
             "temporary_user_delete_after",
+            "encryption_kp",
         ]
 
 

authentik/sources/saml/exceptions.py

@@ -19,5 +19,9 @@ class MismatchedRequestID(SAMLException):
     """Exception raised when the returned request ID doesn't match the saved ID."""
 
 
+class InvalidEncryption(SAMLException):
+    """Encryption of XML Object is either missing or invalid"""
+
+
 class InvalidSignature(SAMLException):
     """Signature of XML Object is either missing or invalid"""

authentik/sources/saml/migrations/0016_samlsource_encryption_kp.py

@@ -0,0 +1,29 @@
+# Generated by Django 5.0.8 on 2024-08-07 17:33
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_crypto", "0004_alter_certificatekeypair_name"),
+        ("authentik_sources_saml", "0015_groupsamlsourceconnection_samlsourcepropertymapping"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="samlsource",
+            name="encryption_kp",
+            field=models.ForeignKey(
+                blank=True,
+                default=None,
+                help_text="When selected, incoming assertions are encrypted by the IdP using the public key of the encryption keypair. The assertion is decrypted by the SP using the the private key.",
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="+",
+                to="authentik_crypto.certificatekeypair",
+                verbose_name="Encryption Keypair",
+            ),
+        ),
+    ]

authentik/sources/saml/models.py

@@ -156,6 +156,20 @@ class SAMLSource(Source):
         on_delete=models.SET_NULL,
         verbose_name=_("Signing Keypair"),
     )
+    encryption_kp = models.ForeignKey(
+        CertificateKeyPair,
+        default=None,
+        null=True,
+        blank=True,
+        help_text=_(
+            "When selected, incoming assertions are encrypted by the IdP using the public "
+            "key of the encryption keypair. The assertion is decrypted by the SP using the "
+            "the private key."
+        ),
+        on_delete=models.SET_NULL,
+        verbose_name=_("Encryption Keypair"),
+        related_name="+",
+    )
 
     digest_algorithm = models.TextField(
         choices=(

authentik/sources/saml/processors/metadata.py

@@ -46,6 +46,20 @@
             return key_descriptor
         return None
 
+    def get_encryption_key_descriptor(self) -> Optional[Element]:  # noqa: UP007
+        """Get Encryption KeyDescriptor, if enabled for the source"""
+        if self.source.encryption_kp:
+            key_descriptor = Element(f"{{{NS_SAML_METADATA}}}KeyDescriptor")
+            key_descriptor.attrib["use"] = "encryption"
+            key_info = SubElement(key_descriptor, f"{{{NS_SIGNATURE}}}KeyInfo")
+            x509_data = SubElement(key_info, f"{{{NS_SIGNATURE}}}X509Data")
+            x509_certificate = SubElement(x509_data, f"{{{NS_SIGNATURE}}}X509Certificate")
+            x509_certificate.text = strip_pem_header(
+                self.source.encryption_kp.certificate_data.replace("\r", "")
+            ).replace("\n", "")
+            return key_descriptor
+        return None
+
     def get_name_id_formats(self) -> Iterator[Element]:
         """Get compatible NameID Formats"""
         formats = [
@@ -74,6 +88,10 @@
         if signing_descriptor is not None:
             sp_sso_descriptor.append(signing_descriptor)
 
+        encryption_descriptor = self.get_encryption_key_descriptor()
+        if encryption_descriptor is not None:
+            sp_sso_descriptor.append(encryption_descriptor)
+
         for name_id_format in self.get_name_id_formats():
             sp_sso_descriptor.append(name_id_format)
 

authentik/sources/saml/processors/response.py

@@ -23,6 +23,7 @@
 from authentik.core.sources.flow_manager import SourceFlowManager
 from authentik.lib.utils.time import timedelta_from_string
 from authentik.sources.saml.exceptions import (
+    InvalidEncryption,
     InvalidSignature,
     MismatchedRequestID,
     MissingSAMLResponse,
@@ -76,11 +77,43 @@
         self._root_xml = b64decode(raw_response.encode())
         self._root = fromstring(self._root_xml)
 
+        if self._source.encryption_kp:
+            self._decrypt_response()
+
         if self._source.verification_kp:
             self._verify_signed()
         self._verify_request_id()
         self._verify_status()
 
+    def _decrypt_response(self):
+        """Decrypt SAMLResponse EncryptedAssertion Element"""
+        manager = xmlsec.KeysManager()
+        key = xmlsec.Key.from_memory(
+            self._source.encryption_kp.key_data,
+            xmlsec.constants.KeyDataFormatPem,
+        )
+
+        manager.add_key(key)
+        encryption_context = xmlsec.EncryptionContext(manager)
+
+        encrypted_assertion = self._root.find(f".//{{{NS_SAML_ASSERTION}}}EncryptedAssertion")
+        if encrypted_assertion is None:
+            raise InvalidEncryption()
+        encrypted_data = xmlsec.tree.find_child(
+            encrypted_assertion, "EncryptedData", xmlsec.constants.EncNs
+        )
+        try:
+            decrypted_assertion = encryption_context.decrypt(encrypted_data)
+        except xmlsec.Error as exc:
+            raise InvalidEncryption() from exc
+
+        index_of = self._root.index(encrypted_assertion)
+        self._root.remove(encrypted_assertion)
+        self._root.insert(
+            index_of,
+            decrypted_assertion,
+        )
+
     def _verify_signed(self):
         """Verify SAML Response's Signature"""
         signature_nodes = self._root.xpath(
@@ -101,9 +134,9 @@
         ctx.set_enabled_key_data([xmlsec.constants.KeyDataX509])
         try:
             ctx.verify(signature_node)
-        except (xmlsec.InternalError, xmlsec.VerificationError) as exc:
-            raise InvalidSignature from exc
-        LOGGER.debug("Successfully verified signautre")
+        except xmlsec.Error as exc:
+            raise InvalidSignature() from exc
+        LOGGER.debug("Successfully verified signature")
 
     def _verify_request_id(self):
         if self._source.allow_idp_initiated:

authentik/sources/saml/tests/fixtures/encrypted-key.pem

@@ -0,0 +1,51 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIJKQIBAAKCAgEAqdjTNNuHV8I13gHYx3S4vjGdMaL8+B18OmA/iK9DV2OhW9T6
+zL2tXpG5Iw2mZi8OhIgKC4if3wL314NwwKoU++nEMn/uyYUG1c/YpvttpjhCTzwh
+rqDjZYhyae/Ef4pB68UUMVvcCZpNuqZbYkeF1gZMRSv3oiq9fbIndT7Yc7f7nXug
+qzO/sqpQdwRXBJ3zoC5abJg2q+iYslC2IiFe/43XlW1GZFPt5910kx2lfhnJRYQD
+BiSOIxwOPSeh7qhgpkxKxHUjlW757kdmNjIpL5v51JG/CZpAHYHWx61gPwyMuqT/
+xKxAL+J9K4gIRZP8ViHBFw1FVIe/UI8/Yf19L8IMIdLmS2d2HH/bRLinig5yJXko
+78KKNeWMBICmvJVQ1VpyBtwFyPw0x6zzZVSCEZ8CpJgnnaJ96YcyTg1eaXEBoxRb
+j795D/k899hVn9RxovDzg2yUH5WaHqiWjHMrGrkVvLWj5ojC2lgzrLsEGN+FS49F
+wS2zwHoQTrbcJL4W029m0BfvAjdKrtuTGyM4hK6thfyilCQTCEJmZ0gkEYfnLhbx
+QmT19jnWTof+2MBrj1vdlvh0CJxIXxD9BtI5Q9Zf4xkMJv8LmCuRXADkLduyd/Jz
+Q2P3aCt0AV/C1doR/yd0LtNY7skyOV6YOjPTNW5AMbX889Gw77TcYmCVn/cCAwEA
+AQKCAgBOG4bhf3VJv+fazTmeXAibeqCCE6THC3Q2Ok3tc0ACP7CUVSjzH+VLILOl
+saDMzCYef5sy+6UdvzUv2GPxTiYxRSszWA79gJ4IlLla7TRbJPMlkg8hSh7Y8fs/
+yYIxbujq3mpvWoGhruLBC8DpvN+I8cOAafxLCOG0nMm1iu2qpbjiDtjv8m/dX6J6
+YTYNSwAfMUHnP8agnuod0q03m+YemuHB94tQFyLIpth10UPqbjxXqiJj4Eq3Ta8k
+o4W+BZPQ1jPqDb6L+YmZcR9JnB7BpLaq8U2LwnJqv2uAzzP8Oq67JKb0kIxCGSOb
+8cZwDOKVz5cHHVS9T2IFT6MD0rmPDZxUNl7e/T8cNjF92/+fsai7LOnMYzgBL5KG
+DYzI4kEW1eqeKkTH6domAAvfva0rLH2JhyWWyvV7o14xjBL+hvhyu6ba0KKPUENz
+xFkQrFDCa3Xch6GeWHtgT6l+Tjy9pwg7WoS1twAHuVl33Hz/xDZVC7Hf7DGEcJFv
+sqD3kYvl2TgCbqw5jb72Vrvd6kGM3X1SPiChWtc+7N7LR3/b6ugf2Cqx9QVNve2U
+nkqNW0TNsQIBUwk9bUM0vWZ2z9jT+mcayXjk1Comptj9fgOpNn0yxMrCLQaSi3X8
+L/5ZArzPppkDXUa7MwVeSyJnYCaA2OGw4p5lMDwM01gkij7c0QKCAQEA06KepO4d
+H/ZmjMjChDxEdKgwY0oGsbOM6l9d/0YBe/kAQYFIsJ7U9u2d1Qx5g0ELDsCHPzIX
+zcatng4fDOHvZWaFiE+vtgH4+8H0q5yvQ18WrDb2EcjtsXNDItgPb5Oo1lc7MlM5
+iu7w/u49l53d17DaxAc96RFhOQNGvWa3U3HvBlkB3SCl4NnCWeVh/G46bYF9Q9g5
+Jg7d1djcTlONXBlRVGCDnHro3rS0IxFCYla2F8CEh6FepthvWCgUxQ+WZTkHluCY
+J6xflufeormLlrMwjcgYcaapikCelbBnEGqfzqklRQHfLhMPeYFh3KaBxr1J+Xzc
+n4w2TpAveJnwMQKCAQEAzXOkpiF8EC0DKadeGtbRiw51p8qbXrNlxmg8T7BKpSB1
+X3aVgCtwB4UYZz3Jvz9LStWDTzCZkiLydpzBDCk6sTdJW98KClzFbl6NdwNu1kdj
+SWj/9izmEDi9SHXvo+RnC37k+QNrdSWWzLV7heglXmjY/+IHHhNinOCsL7sARXLa
+sS2/Fl+cyXsngDQAUpyVCVWW7kmY9QQR7Q798guj63x/0bObud9xImnNfvchFzn0
+oahZ/ZY+3FGq5+8pKsfV0jJGtB9dyYoZ0+h3auxkKvE13rUoOMWiyAxfA44/S97C
+YWv3nBdcCcLkw/XjR843q8D7ctQXMMYcqatFL7zwpwKCAQBNWnkF64p1rkgZWR/P
+2X9j7D2TbPE5blkpKSZgMaRFPePcDXcWJ1fL0VoJDwAy+0khYTmN3a9ZpS68QIkU
+2lf4Bhr0kbu1mM76pg/Z0fE1fMH6vDQAmCJY47o8OCCcNapWfZfDcyvrHh6z7zxP
++IGnXpr3X3Y/g/y3K/1lKPAE7fXhqhLGUjKPFsi0tuSzsU5lzBiO/a8VvAVVLmiH
+sH5QlWhmoMg6H6qSDBZzYtGSxALWd6V5NYA1F5LK9AtzY5ki8k9V1E2I4rYloCZ9
+77eXo3Mxv1s/3xzEzY2pRMrG81Hp5WUb7e03F/xl+uZcEfgJPhKVwA+buVH4MTdI
+q2thAoIBAQCjZAzVclvQIXwabFiSz7Tl+iHnx2G49sNB/zO3zGQQ3rd5rD1JKUJ3
+OIon0SPZTOT8JsG/AM+hQNnDKvb8TO24cleNENxTUWRSWi/3Lmu/ThbQEwk9Jofw
+7q7aKbDjjonEwq4mu2mCSNqdAtexruXJJ2ksVv2CFbifOq61ZurYUHdL4S3PBUsT
+kTXg53o6OPzt53uZFj7m3M3E0d9z134NkX21sDlwoRrAW5RqHO/cIONEjTbETfDA
+FtLskW8T7slF2WYRacCUv5e6x23xQv6GiD5nV3sda1AB+JS3pzD/jbDY+Zx6Lrmr
+qat1jN+sA3ySw2816yZmS6gP532mcYSRAoIBAQCAkIU6fwLcNL262Ty8a231x74J
+vqMTg8y8lZdC/nhwF7qBxhb43CekSFNSi+s17voN+ko6Gt0uRXIQ5GueiiVWFPoG
+arM6bnPNu1uZ566+vXPfwQ73WZ5uG0cw/z1NRkHWDGsoX0M7b8u/PvAkN0KY5PwV
+Xy4XHamfizQAg4Bh9PnBWyXQXSgGhzRaia7YnorFZPrXB+zDsicX2DkhjquPSIfS
+pvv0aeDqx9EfhSymJlaIsp6o3jL6pYiQtvKPmQm3a4suf7/rhoMn7gIe/Btypzs6
+y2cEqNlvBYi4s2d/nVsXirXDiGdBwbDQhRm4w39Yv2si2/8zMDlhapf+KHWE
+-----END RSA PRIVATE KEY-----

authentik/sources/saml/tests/test_metadata.py

@@ -20,6 +20,7 @@ def setUp(self):
             slug=generate_id(),
             issuer="authentik",
             signing_kp=create_test_cert(),
+            encryption_kp=create_test_cert(),
             pre_authentication_flow=create_test_flow(),
         )
 
@@ -46,7 +47,7 @@ def test_metadata(self):
         metadata = ElementTree.fromstring(xml)
         self.assertEqual(metadata.attrib["entityID"], "authentik")
 
-    def test_metadata_without_signautre(self):
+    def test_metadata_without_signature(self):
         """Test Metadata generation being valid"""
         self.source.signing_kp = None
         self.source.save()

authentik/sources/saml/tests/test_response.py

@@ -5,9 +5,11 @@
 from django.contrib.sessions.middleware import SessionMiddleware
 from django.test import RequestFactory, TestCase
 
-from authentik.core.tests.utils import create_test_flow
+from authentik.core.tests.utils import create_test_cert, create_test_flow
+from authentik.crypto.models import CertificateKeyPair
 from authentik.lib.generators import generate_id
 from authentik.lib.tests.utils import dummy_get_response, load_fixture
+from authentik.sources.saml.exceptions import InvalidEncryption
 from authentik.sources.saml.models import SAMLSource
 from authentik.sources.saml.processors.response import ResponseProcessor
 
@@ -77,3 +79,48 @@ def test_success(self):
                 "path": self.source.get_user_path(),
             },
         )
+
+    def test_encrypted_correct(self):
+        """Test encrypted"""
+        key = load_fixture("fixtures/encrypted-key.pem")
+        kp = CertificateKeyPair.objects.create(
+            name=generate_id(),
+            key_data=key,
+        )
+        self.source.encryption_kp = kp
+        request = self.factory.post(
+            "/",
+            data={
+                "SAMLResponse": b64encode(
+                    load_fixture("fixtures/response_encrypted.xml").encode()
+                ).decode()
+            },
+        )
+
+        middleware = SessionMiddleware(dummy_get_response)
+        middleware.process_request(request)
+        request.session.save()
+
+        parser = ResponseProcessor(self.source, request)
+        parser.parse()
+
+    def test_encrypted_incorrect_key(self):
+        """Test encrypted"""
+        kp = create_test_cert()
+        self.source.encryption_kp = kp
+        request = self.factory.post(
+            "/",
+            data={
+                "SAMLResponse": b64encode(
+                    load_fixture("fixtures/response_encrypted.xml").encode()
+                ).decode()
+            },
+        )
+
+        middleware = SessionMiddleware(dummy_get_response)
+        middleware.process_request(request)
+        request.session.save()
+
+        parser = ResponseProcessor(self.source, request)
+        with self.assertRaises(InvalidEncryption):
+            parser.parse()

blueprints/schema.json

@@ -7407,6 +7407,12 @@
                     "minLength": 1,
                     "title": "Delete temporary users after",
                     "description": "Time offset when temporary users should be deleted. This only applies if your IDP uses the NameID Format 'transient', and the user doesn't log out manually. (Format: hours=1;minutes=2;seconds=3)."
+                },
+                "encryption_kp": {
+                    "type": "string",
+                    "format": "uuid",
+                    "title": "Encryption Keypair",
+                    "description": "When selected, incoming assertions are encrypted by the IdP using the public key of the encryption keypair. The assertion is decrypted by the SP using the the private key."
                 }
             },
             "required": []