feat: configurable encryption algorithm types

Author: xpavlic

docs/howto/config.rst

@@ -656,6 +656,60 @@ 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"
+    - "http://www.w3.org/2009/xmlenc11#rsa-oaep" (xmlsec1 version>=1.3.0 is required)
+
+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"
+    ]
+
+default_rsa_oaep_mgf_alg
+^^^^^^^^^^^^^^^^^^^^^^^^
+If encryption key from metadata has no encryption method specified or does not have one matching configuration and
+"http://www.w3.org/2009/xmlenc11#rsa-oaep" is selected, it will be used with mask generation function specified by
+this configuration option. Default value is None
+
+Valid values are:
+    - "http://www.w3.org/2009/xmlenc11#mgf1sha1"
+    - "http://www.w3.org/2009/xmlenc11#mgf1sha224"
+    - "http://www.w3.org/2009/xmlenc11#mgf1sha256"
+    - "http://www.w3.org/2009/xmlenc11#mgf1sha384"
+    - "http://www.w3.org/2009/xmlenc11#mgf1sha512"
+
+Example::
+
+    "default_rsa_oaep_mgf_alg": "http://www.w3.org/2009/xmlenc11#mgf1sha1"
 
 Specific directives
 -------------------

src/saml2/config.py

@@ -76,6 +76,9 @@
     "signing_algorithm",
     "digest_algorithm",
     "http_client_timeout",
+    "encrypt_assertion_session_key_algs",
+    "encrypt_assertion_cert_key_algs",
+    "default_rsa_oaep_mgf_alg",
 ]
 
 SP_ARGS = [
@@ -229,6 +232,9 @@ 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"]
+        self.default_rsa_oaep_mgf_alg = None
 
     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, RSA_OAEP
 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,10 @@ 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
+        self.default_rsa_oaep_mgf_alg = self.config.default_rsa_oaep_mgf_alg
+
         if virtual_organization:
             if isinstance(virtual_organization, str):
                 self.vorg = self.config.vorg[virtual_organization]
@@ -194,7 +197,6 @@ def __init__(self, entity_type, config=None, config_file="", virtual_organizatio
             self.sourceid = self.metadata.construct_source_id()
         else:
             self.sourceid = {}
-
         self.msg_cb = msg_cb
 
     def reload_metadata(self, metadata_conf):
@@ -644,34 +646,105 @@ 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 _get_first_matching_alg(self, priority_list, metadata_list):
+        for alg in priority_list:
+            for cert_method in metadata_list:
+                if cert_method.get("algorithm") == alg:
+                    return cert_method
+        return 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]
+                )
+
+                rsa_oaep_mgf_alg = self.default_rsa_oaep_mgf_alg if key_enc == RSA_OAEP else None
+                if encrypt_cert != _cert and _cert_encryption_methods:
+                    viable_session_key_alg = self._get_first_matching_alg(
+                        self.encrypt_assertion_session_key_algs, _cert_encryption_methods
+                    )
+                    if viable_session_key_alg:
+                        msg_enc = viable_session_key_alg.get("algorithm")
+
+                    viable_cert_alg = self._get_first_matching_alg(
+                        self.encrypt_assertion_cert_key_algs, _cert_encryption_methods
+                    )
+                    if viable_cert_alg:
+                        key_enc = viable_cert_alg.get("algorithm")
+                        mgf = viable_cert_alg.get("mgf")
+                        rsa_oaep_mgf_alg = mgf.get("algorithm") if mgf else None
+
+                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,
+                        rsa_oaep_mgf_alg=rsa_oaep_mgf_alg,
+                    ),
+                    key_type=key_type,
                     node_xpath=node_xpath,
                 )
                 return response
@@ -697,7 +770,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 +808,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 +941,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 +987,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 +1450,6 @@ def create_manage_name_id_response(
         digest_alg=None,
         **kwargs,
     ):
-
         rinfo = self.response_args(request, bindings)
 
         response = self._status_response(

src/saml2/md.py

@@ -10,7 +10,6 @@
 from saml2 import xmldsig as ds
 from saml2 import xmlenc as xenc
 
-
 NAMESPACE = "urn:oasis:names:tc:SAML:2.0:metadata"
 
 
@@ -803,13 +802,15 @@ def __init__(
         text=None,
         extension_elements=None,
         extension_attributes=None,
+        mgf=None,
     ):
         SamlBase.__init__(
             self, text=text, extension_elements=extension_elements, extension_attributes=extension_attributes
         )
         self.key_info = key_info
         self.encryption_method = encryption_method or []
         self.use = use
+        self.mgf = mgf
 
 
 def key_descriptor_type__from_string(xml_string):

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 [