feat: configurable encryption algorithm types

Author: xpavlic

docs/howto/config.rst

@@ -656,6 +656,42 @@ Example::
 
     "verify_encrypt_cert_assertion": verify_encrypt_cert
 
+encrypt_assertion_session_key_algs
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+List of block encryption algorithms which can be used to encrypt assertion.
+Values order is from highest to lowest priority. Default value is ["http://www.w3.org/2001/04/xmlenc#tripledes-cbc"]
+
+Valid values are:
+    - "http://www.w3.org/2001/04/xmlenc#tripledes-cbc"
+    - "http://www.w3.org/2001/04/xmlenc#aes128-cbc"
+    - "http://www.w3.org/2001/04/xmlenc#aes192-cbc"
+    - "http://www.w3.org/2001/04/xmlenc#aes256-cbc"
+    - "http://www.w3.org/2009/xmlenc11#aes128-gcm"
+    - "http://www.w3.org/2009/xmlenc11#aes192-gcm"
+    - "http://www.w3.org/2009/xmlenc11#aes256-gcm"
+
+Example::
+
+    "encrypt_assertion_session_key_algs" : [
+        "http://www.w3.org/2009/xmlenc11#aes256-gcm",
+        "http://www.w3.org/2001/04/xmlenc#tripledes-cbc"
+    ]
+
+encrypt_assertion_cert_key_algs
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+List of key transport algorithms which can be used to encrypt session key used to encrypting assertion.
+Values order is from highest to lowest priority. Default value is ["http://www.w3.org/2001/04/xmlenc#rsa-oaep-mgf1p"]
+
+Valid values are:
+    - "http://www.w3.org/2001/04/xmlenc#rsa-1_5"
+    - "http://www.w3.org/2001/04/xmlenc#rsa-oaep-mgf1p"
+
+Example::
+
+    "encrypt_assertion_cert_key_algs": [
+        "http://www.w3.org/2001/04/xmlenc#rsa-oaep-mgf1p",
+        "http://www.w3.org/2001/04/xmlenc#rsa-1_5"
+    ]
 
 Specific directives
 -------------------

src/saml2/config.py

@@ -76,6 +76,8 @@
     "signing_algorithm",
     "digest_algorithm",
     "http_client_timeout",
+    "encrypt_assertion_session_key_algs",
+    "encrypt_assertion_cert_key_algs",
 ]
 
 SP_ARGS = [
@@ -229,6 +231,8 @@ def __init__(self, homedir="."):
         self.signing_algorithm = None
         self.digest_algorithm = None
         self.http_client_timeout = None
+        self.encrypt_assertion_session_key_algs = ["http://www.w3.org/2001/04/xmlenc#tripledes-cbc"]
+        self.encrypt_assertion_cert_key_algs = ["http://www.w3.org/2001/04/xmlenc#rsa-oaep-mgf1p"]
 
     def setattr(self, context, attr, val):
         if context == "":

src/saml2/entity.py

@@ -60,7 +60,7 @@
 from saml2.samlp import SessionIndex
 from saml2.samlp import artifact_resolve_from_string
 from saml2.samlp import response_from_string
-from saml2.sigver import SignatureError
+from saml2.sigver import SignatureError, XMLSEC_SESSION_KEY_URI_TO_ALG
 from saml2.sigver import SigverError
 from saml2.sigver import get_pem_wrapped_unwrapped
 from saml2.sigver import make_temp
@@ -78,7 +78,6 @@
 from saml2.xmldsig import SIG_ALLOWED_ALG
 from saml2.xmldsig import DefaultSignature
 
-
 logger = logging.getLogger(__name__)
 
 __author__ = "rolandh"
@@ -181,6 +180,9 @@ def __init__(self, entity_type, config=None, config_file="", virtual_organizatio
 
         self.sec = security_context(self.config)
 
+        self.encrypt_assertion_session_key_algs = self.config.encrypt_assertion_session_key_algs
+        self.encrypt_assertion_cert_key_algs = self.config.encrypt_assertion_cert_key_algs
+
         if virtual_organization:
             if isinstance(virtual_organization, str):
                 self.vorg = self.config.vorg[virtual_organization]
@@ -644,34 +646,97 @@ def has_encrypt_cert_in_metadata(self, sp_entity_id):
                 return True
         return False
 
-    def _encrypt_assertion(self, encrypt_cert, sp_entity_id, response, node_xpath=None):
+    def _encrypt_assertion(
+        self,
+        encrypt_cert,
+        sp_entity_id,
+        response,
+        node_xpath=None,
+        encrypt_cert_session_key_alg=None,
+        encrypt_cert_cert_key_alg=None,
+    ):
         """Encryption of assertions.
 
         :param encrypt_cert: Certificate to be used for encryption.
         :param sp_entity_id: Entity ID for the calling service provider.
         :param response: A samlp.Response
+        :param encrypt_cert_cert_key_alg: algorithm used for encrypting session key
+        :param encrypt_cert_session_key_alg: algorithm used for encrypting assertion
+        :param encrypt_cert_cert_key_alg:
         :param node_xpath: Unquie path to the element to be encrypted.
         :return: A new samlp.Resonse with the designated assertion encrypted.
         """
         _certs = []
 
         if encrypt_cert:
-            _certs.append((None, encrypt_cert))
+            _certs.append((None, encrypt_cert, None, None))
         elif sp_entity_id is not None:
-            _certs = self.metadata.certs(sp_entity_id, "any", "encryption")
+            _certs = self.metadata.certs(sp_entity_id, "any", "encryption", get_with_usage_and_encryption_methods=True)
         exception = None
-        for _cert_name, _cert in _certs:
+
+        # take certs with encryption and encryption_methods first (priority 1)
+        sorted_certs = []
+        for _unpacked_cert in _certs:
+            _cert_name, _cert, _cert_use, _cert_encryption_methods = _unpacked_cert
+            if _cert_use == "encryption" and _cert_encryption_methods:
+                sorted_certs.append(_unpacked_cert)
+
+        # take certs with encryption or encryption_methods (priority 2)
+        for _unpacked_cert in _certs:
+            _cert_name, _cert, _cert_use, _cert_encryption_methods = _unpacked_cert
+            if _cert_use == "encryption" and _unpacked_cert not in sorted_certs:
+                sorted_certs.append(_unpacked_cert)
+
+        for _unpacked_cert in _certs:
+            if _unpacked_cert not in sorted_certs:
+                sorted_certs.append(_unpacked_cert)
+
+        for _cert_name, _cert, _cert_use, _cert_encryption_methods in sorted_certs:
             wrapped_cert, unwrapped_cert = get_pem_wrapped_unwrapped(_cert)
             try:
                 tmp = make_temp(
                     wrapped_cert.encode("ascii"),
                     decode=False,
                     delete_tmpfiles=self.config.delete_tmpfiles,
                 )
+
+                msg_enc = (
+                    encrypt_cert_session_key_alg
+                    if encrypt_cert_session_key_alg
+                    else self.encrypt_assertion_session_key_algs[0]
+                )
+                key_enc = (
+                    encrypt_cert_cert_key_alg if encrypt_cert_cert_key_alg else self.encrypt_assertion_cert_key_algs[0]
+                )
+
+                if encrypt_cert != _cert and _cert_encryption_methods:
+                    viable_session_key_algs = []
+                    for alg in self.encrypt_assertion_session_key_algs:
+                        for cert_method in _cert_encryption_methods:
+                            if cert_method.get("algorithm") == alg:
+                                viable_session_key_algs.append(alg)
+
+                    viable_cert_algs = []
+                    for alg in self.encrypt_assertion_cert_key_algs:
+                        for cert_method in _cert_encryption_methods:
+                            if cert_method.get("algorithm") == alg:
+                                viable_cert_algs.append(alg)
+
+                    if viable_session_key_algs:
+                        msg_enc = viable_session_key_algs[0]
+
+                    if viable_cert_algs:
+                        key_enc = viable_cert_algs[0]
+
+                key_type = XMLSEC_SESSION_KEY_URI_TO_ALG.get(msg_enc)
+
                 response = self.sec.encrypt_assertion(
                     response,
                     tmp.name,
-                    pre_encryption_part(key_name=_cert_name, encrypt_cert=unwrapped_cert),
+                    pre_encryption_part(
+                        key_name=_cert_name, encrypt_cert=unwrapped_cert, msg_enc=msg_enc, key_enc=key_enc
+                    ),
+                    key_type=key_type,
                     node_xpath=node_xpath,
                 )
                 return response
@@ -697,7 +762,11 @@ def _response(
         encrypt_assertion_self_contained=False,
         encrypted_advice_attributes=False,
         encrypt_cert_advice=None,
+        encrypt_cert_advice_cert_key_alg=None,
+        encrypt_cert_advice_session_key_alg=None,
         encrypt_cert_assertion=None,
+        encrypt_cert_assertion_cert_key_alg=None,
+        encrypt_cert_assertion_session_key_alg=None,
         sign_assertion=None,
         pefim=False,
         sign_alg=None,
@@ -731,8 +800,16 @@ def _response(
         element should be encrypted.
         :param encrypt_cert_advice: Certificate to be used for encryption of
         assertions in the advice element.
+        :param encrypt_cert_advice_cert_key_alg: algorithm used for encrypting session key
+        by encrypt_cert_advice
+        :param encrypt_cert_advice_session_key_alg: algorithm used for encrypting assertion
+        when using encrypt_cert_advice
         :param encrypt_cert_assertion: Certificate to be used for encryption
         of assertions.
+        :param encrypt_cert_assertion_cert_key_alg: algorithm used for encrypting session key
+        by encrypt_cert_assertion
+        :param encrypt_cert_assertion_session_key_alg: algorithm used for encrypting assertion when
+        using encrypt_cert_assertion
         :param sign_assertion: True if assertions should be signed.
         :param pefim: True if a response according to the PEFIM profile
         should be created.
@@ -856,6 +933,8 @@ def _response(
                             sp_entity_id,
                             response,
                             node_xpath=node_xpath,
+                            encrypt_cert_session_key_alg=encrypt_cert_advice_session_key_alg,
+                            encrypt_cert_cert_key_alg=encrypt_cert_advice_cert_key_alg,
                         )
                         response = response_from_string(response)
 
@@ -900,7 +979,13 @@ def _response(
                     response = signed_instance_factory(response, self.sec, to_sign_assertion)
 
                 # XXX encrypt assertion
-                response = self._encrypt_assertion(encrypt_cert_assertion, sp_entity_id, response)
+                response = self._encrypt_assertion(
+                    encrypt_cert_assertion,
+                    sp_entity_id,
+                    response,
+                    encrypt_cert_session_key_alg=encrypt_cert_assertion_session_key_alg,
+                    encrypt_cert_cert_key_alg=encrypt_cert_assertion_cert_key_alg,
+                )
             else:
                 # XXX sign other parts! (defiend by to_sign)
                 if to_sign:
@@ -1357,7 +1442,6 @@ def create_manage_name_id_response(
         digest_alg=None,
         **kwargs,
     ):
-
         rinfo = self.response_args(request, bindings)
 
         response = self._status_response(

src/saml2/mdstore.py

@@ -476,7 +476,7 @@ def __eq__(self, other):
 
         return True
 
-    def certs(self, entity_id, descriptor, use="signing"):
+    def certs(self, entity_id, descriptor, use="signing", get_with_usage_and_encryption_methods=False):
         """
         Returns certificates for the given Entity
         """
@@ -494,7 +494,10 @@ def extract_certs(srvs):
                         for dat in key_info["x509_data"]:
                             cert = repack_cert(dat["x509_certificate"]["text"])
                             if cert not in res:
-                                res.append((key_name_txt, cert))
+                                if get_with_usage_and_encryption_methods:
+                                    res.append((key_name_txt, cert, key_use, key.get("encryption_method")))
+                                else:
+                                    res.append((key_name_txt, cert))
 
             return res
 
@@ -1327,7 +1330,7 @@ def subject_id_requirement(self, entity_id):
                     "name_format": "urn:oasis:names:tc:SAML:2.0:attrname-format:uri",
                     "friendly_name": "subject-id",
                     "is_required": "true",
-                }
+                },
             ]
         elif subject_id_req == "pairwise-id":
             return [

src/saml2/server.py

@@ -457,7 +457,11 @@ def _authn_response(
         best_effort=False,
         encrypt_assertion=False,
         encrypt_cert_advice=None,
+        encrypt_cert_advice_cert_key_alg=None,
+        encrypt_cert_advice_session_key_alg=None,
         encrypt_cert_assertion=None,
+        encrypt_cert_assertion_cert_key_alg=None,
+        encrypt_cert_assertion_session_key_alg=None,
         authn_statement=None,
         encrypt_assertion_self_contained=False,
         encrypted_advice_attributes=False,
@@ -492,8 +496,16 @@ def _authn_response(
         element should be encrypted.
         :param encrypt_cert_advice: Certificate to be used for encryption of
         assertions in the advice element.
+        :param encrypt_cert_advice_cert_key_alg: algorithm used for encrypting session key
+        by encrypt_cert_advice
+        :param encrypt_cert_advice_session_key_alg: algorithm used for encrypting assertion
+        when using encrypt_cert_advice
         :param encrypt_cert_assertion: Certificate to be used for encryption
         of assertions.
+        :param encrypt_cert_assertion_cert_key_alg: algorithm used for encrypting session key
+        by encrypt_cert_assertion
+        :param encrypt_cert_assertion_session_key_alg: algorithm used for encrypting assertion when
+        using encrypt_cert_assertion
         :param authn_statement: Authentication statement.
         :param pefim: True if a response according to the PEFIM profile
         should be created.
@@ -598,7 +610,11 @@ def _authn_response(
             sp_entity_id=sp_entity_id,
             encrypt_assertion=encrypt_assertion,
             encrypt_cert_advice=encrypt_cert_advice,
+            encrypt_cert_advice_cert_key_alg=encrypt_cert_advice_cert_key_alg,
+            encrypt_cert_advice_session_key_alg=encrypt_cert_advice_session_key_alg,
             encrypt_cert_assertion=encrypt_cert_assertion,
+            encrypt_cert_assertion_cert_key_alg=encrypt_cert_assertion_cert_key_alg,
+            encrypt_cert_assertion_session_key_alg=encrypt_cert_assertion_session_key_alg,
             encrypt_assertion_self_contained=encrypt_assertion_self_contained,
             encrypted_advice_attributes=encrypted_advice_attributes,
             sign_assertion=sign_assertion,
@@ -724,7 +740,6 @@ def gather_authn_response_args(self, sp_entity_id, name_id_policy, userid, **kwa
             ("encrypted_advice_attributes", "verify_encrypt_cert_advice", "encrypt_cert_advice", kwargs["pefim"]),
             ("encrypt_assertion", "verify_encrypt_cert_assertion", "encrypt_cert_assertion", False),
         ]:
-
             if args[arg] or pefim:
                 _enc_cert = self.config.getattr(attr, "idp")
 
@@ -789,7 +804,11 @@ def create_authn_response(
         sign_response=None,
         sign_assertion=None,
         encrypt_cert_advice=None,
+        encrypt_cert_advice_cert_key_alg=None,
+        encrypt_cert_advice_session_key_alg=None,
         encrypt_cert_assertion=None,
+        encrypt_cert_assertion_cert_key_alg=None,
+        encrypt_cert_assertion_session_key_alg=None,
         encrypt_assertion=None,
         encrypt_assertion_self_contained=True,
         encrypted_advice_attributes=False,
@@ -822,8 +841,16 @@ def create_authn_response(
         element should be encrypted.
         :param encrypt_cert_advice: Certificate to be used for encryption of
         assertions in the advice element.
+        :param encrypt_cert_advice_cert_key_alg: algorithm used for encrypting session key
+        by encrypt_cert_advice
+        :param encrypt_cert_advice_session_key_alg: algorithm used for encrypting assertion
+        when using encrypt_cert_advice
         :param encrypt_cert_assertion: Certificate to be used for encryption
         of assertions.
+        :param encrypt_cert_assertion_cert_key_alg: algorithm used for encrypting session key
+        by encrypt_cert_assertion
+        :param encrypt_cert_assertion_session_key_alg: algorithm used for encrypting assertion when
+        using encrypt_cert_assertion
         :param pefim: True if a response according to the PEFIM profile
         should be created.
         :return: A response instance
@@ -869,6 +896,10 @@ def create_authn_response(
                 sign_alg=sign_alg,
                 digest_alg=digest_alg,
                 session_not_on_or_after=session_not_on_or_after,
+                encrypt_cert_advice_cert_key_alg=encrypt_cert_advice_cert_key_alg,
+                encrypt_cert_advice_session_key_alg=encrypt_cert_advice_session_key_alg,
+                encrypt_cert_assertion_cert_key_alg=encrypt_cert_assertion_cert_key_alg,
+                encrypt_cert_assertion_session_key_alg=encrypt_cert_assertion_session_key_alg,
                 **args,
             )
         except MissingValue as exc:
@@ -1054,7 +1085,6 @@ def create_ecp_authn_request_response(
         digest_alg=None,
         **kwargs,
     ):
-
         # ----------------------------------------
         # <ecp:Response
         # ----------------------------------------