Merge pull request #222 from HaToHo/master

XML Encryption

Author: Roland Hedberg

example/idp2/idp.py

@@ -335,9 +335,7 @@ def do(self, query, binding_in, relay_state="", encrypt_cert=None,
 
                 _resp = IDP.create_authn_response(
                     identity, userid=self.user,
-                    encrypt_cert=encrypt_cert,
-                    encrypt_assertion_self_contained=True,
-                    encrypted_advice_attributes=True,
+                    encrypt_cert_assertion=encrypt_cert,
                     **resp_args)
             except Exception as excp:
                 logging.error(exception_trace(excp))
@@ -518,7 +516,7 @@ def do_authentication(environ, start_response, authn_context, key,
 
 PASSWD = {
     "daev0001": "qwerty",
-    "haho0032": "qwerty",
+    "testuser": "qwerty",
     "roland": "dianakra",
     "babs": "howes",
     "upper": "crust"}

example/idp2/idp_user.py

@@ -35,22 +35,22 @@
 #USERS = LDAPDict(**ldap_settings)
 
 USERS = {
-    "haho0032": {
-        "sn": "Hoerberg",
-        "givenName": "Hasse",
+    "testuser": {
+        "sn": "Testsson",
+        "givenName": "Test",
         "eduPersonAffiliation": "student",
         "eduPersonScopedAffiliation": "[email protected]",
-        "eduPersonPrincipalName": "haho@example.com",
-        "uid": "haho0032",
+        "eduPersonPrincipalName": "test@example.com",
+        "uid": "testuser",
         "eduPersonTargetedID": "one!for!all",
         "c": "SE",
         "o": "Example Co.",
         "ou": "IT",
         "initials": "P",
         "schacHomeOrganization": "example.com",
-        "email": "hans@example.com",
-        "displayName": "Hans Hoerberg",
-        "labeledURL": "http://www.example.com/haho My homepage",
+        "email": "test@example.com",
+        "displayName": "Test Testsson",
+        "labeledURL": "http://www.example.com/test My homepage",
         "norEduPersonNIN": "SE199012315555"
     },
     "roland": {

example/sp-wsgi/sp.py

@@ -35,7 +35,7 @@
 from saml2.s_utils import sid
 from saml2.s_utils import rndstr
 #from srtest import exception_trace
-from saml2.md import Extensions
+from saml2.samlp import Extensions
 import xmldsig as ds
 
 logger = logging.getLogger("")

src/saml2/__init__.py

@@ -591,13 +591,14 @@ def get_prefix_map(self, elements):
 
     def get_xml_string_with_self_contained_assertion_within_advice_encrypted_assertion(self, assertion_tag, advice_tag):
         for tmp_encrypted_assertion in self.assertion.advice.encrypted_assertion:
-            prefix_map = self.get_prefix_map([tmp_encrypted_assertion._to_element_tree().
-                                                  find(assertion_tag)])
-
-            tree = self._to_element_tree()
-
-            self.set_prefixes(tree.find(assertion_tag).find(advice_tag).find(tmp_encrypted_assertion._to_element_tree()
-                                                                             .tag).find(assertion_tag), prefix_map)
+            if tmp_encrypted_assertion.encrypted_data is None:
+                prefix_map = self.get_prefix_map([tmp_encrypted_assertion._to_element_tree().find(assertion_tag)])
+                tree = self._to_element_tree()
+                encs = tree.find(assertion_tag).find(advice_tag).findall(tmp_encrypted_assertion._to_element_tree().tag)
+                for enc in encs:
+                    assertion = enc.find(assertion_tag)
+                    if assertion is not None:
+                        self.set_prefixes(assertion, prefix_map)
 
         return ElementTree.tostring(tree, encoding="UTF-8")
 

src/saml2/client_base.py

@@ -542,7 +542,7 @@ def create_name_id_mapping_request(self, name_id_policy,
     # ======== response handling ===========
 
     def parse_authn_request_response(self, xmlstr, binding, outstanding=None,
-                                     outstanding_certs=None, decrypt=True):
+                                     outstanding_certs=None):
         """ Deal with an AuthnResponse
 
         :param xmlstr: The reply as a xml string
@@ -573,7 +573,6 @@ def parse_authn_request_response(self, xmlstr, binding, outstanding=None,
                 "attribute_converters": self.config.attribute_converters,
                 "allow_unknown_attributes":
                 self.config.allow_unknown_attributes,
-                "decrypt": decrypt
             }
             try:
                 resp = self._parse_response(xmlstr, AuthnResponse,

src/saml2/config.py

@@ -68,7 +68,8 @@
     "cert_handler_extra_class",
     "generate_cert_func",
     "generate_cert_info",
-    "verify_encrypt_cert",
+    "verify_encrypt_cert_advice",
+    "verify_encrypt_cert_assertion",
     "tmp_cert_file",
     "tmp_key_file",
     "validate_certificate",
@@ -98,6 +99,8 @@
     "sign_assertion",
     "sign_response",
     "encrypt_assertion",
+    "encrypted_advice_attributes",
+    "encrypt_assertion_self_contained",
     "want_authn_requests_signed",
     "want_authn_requests_only_with_valid_cert",
     "provided_attributes",
@@ -220,7 +223,8 @@ def __init__(self, homedir="."):
         self.allow_unsolicited = False
         self.extension_schema = {}
         self.cert_handler_extra_class = None
-        self.verify_encrypt_cert = None
+        self.verify_encrypt_cert_advice = None
+        self.verify_encrypt_cert_assertion = None
         self.generate_cert_func = None
         self.generate_cert_info = None
         self.tmp_cert_file = None

src/saml2/entity.py

@@ -1,5 +1,6 @@
 import base64
 #from binascii import hexlify
+import copy
 import logging
 from hashlib import sha1
 from Crypto.PublicKey import RSA
@@ -37,7 +38,7 @@
 from saml2.s_utils import success_status_factory
 from saml2.s_utils import decode_base64_and_inflate
 from saml2.s_utils import UnsupportedBinding
-from saml2.samlp import AuthnRequest, SessionIndex
+from saml2.samlp import AuthnRequest, SessionIndex, response_from_string
 from saml2.samlp import AuthzDecisionQuery
 from saml2.samlp import AuthnQuery
 from saml2.samlp import AssertionIDRequest
@@ -502,10 +503,46 @@ def _add_info(self, msg, **kwargs):
             else:
                 msg.extension_elements = extensions
 
+    def has_encrypt_cert_in_metadata(self, sp_entity_id):
+        if sp_entity_id is not None:
+            _certs = self.metadata.certs(sp_entity_id, "any", "encryption")
+            if len(_certs) > 0:
+                return True
+        return False
+
+
+    def _encrypt_assertion(self, encrypt_cert, sp_entity_id, response, node_xpath=None):
+        _certs = []
+        cbxs = CryptoBackendXmlSec1(self.config.xmlsec_binary)
+        if encrypt_cert:
+            _certs = []
+            _certs.append(encrypt_cert)
+        elif sp_entity_id is not None:
+            _certs = self.metadata.certs(sp_entity_id, "any", "encryption")
+        exception = None
+        for _cert in _certs:
+            try:
+                begin_cert = "-----BEGIN CERTIFICATE-----\n"
+                end_cert = "\n-----END CERTIFICATE-----\n"
+                if begin_cert not in _cert:
+                    _cert = "%s%s" % (begin_cert, _cert)
+                if end_cert not in _cert:
+                    _cert = "%s%s" % (_cert, end_cert)
+                _, cert_file = make_temp(_cert, decode=False)
+                response = cbxs.encrypt_assertion(response, cert_file,
+                                                  pre_encryption_part(), node_xpath=node_xpath)
+                return response
+            except Exception as ex:
+                exception = ex
+                pass
+        if exception:
+            raise exception
+        return response
+
     def _response(self, in_response_to, consumer_url=None, status=None,
-                  issuer=None, sign=False, to_sign=None,
+                  issuer=None, sign=False, to_sign=None, sp_entity_id=None,
                   encrypt_assertion=False, encrypt_assertion_self_contained=False, encrypted_advice_attributes=False,
-                  encrypt_cert=None, **kwargs):
+                  encrypt_cert_advice=None, encrypt_cert_assertion=None,sign_assertion=None, pefim=False, **kwargs):
         """ Create a Response.
             Encryption:
                 encrypt_assertion must be true for encryption to be performed. If encrypted_advice_attributes also is
@@ -542,43 +579,79 @@ def _response(self, in_response_to, consumer_url=None, status=None,
         if not sign and to_sign and not encrypt_assertion:
             return signed_instance_factory(response, self.sec, to_sign)
 
-        if encrypt_assertion:
-            node_xpath = None
+        has_encrypt_cert = self.has_encrypt_cert_in_metadata(sp_entity_id)
+        if not has_encrypt_cert and encrypt_cert_advice is None:
+            encrypted_advice_attributes = False
+        if not has_encrypt_cert and encrypt_cert_assertion is None:
+            encrypt_assertion = False
+
+        if encrypt_assertion or (encrypted_advice_attributes and response.assertion.advice is not None and
+                                         len(response.assertion.advice.assertion) == 1):
             if sign:
                 response.signature = pre_signature_part(response.id,
                                                         self.sec.my_cert, 1)
                 sign_class = [(class_name(response), response.id)]
             cbxs = CryptoBackendXmlSec1(self.config.xmlsec_binary)
+            encrypt_advice = False
             if encrypted_advice_attributes and response.assertion.advice is not None \
-                    and len(response.assertion.advice.assertion) == 1:
-                tmp_assertion = response.assertion.advice.assertion[0]
-                response.assertion.advice.encrypted_assertion = []
-                response.assertion.advice.encrypted_assertion.append(EncryptedAssertion())
-                if isinstance(tmp_assertion, list):
-                    response.assertion.advice.encrypted_assertion[0].add_extension_elements(tmp_assertion)
-                else:
-                    response.assertion.advice.encrypted_assertion[0].add_extension_element(tmp_assertion)
-                response.assertion.advice.assertion = []
+                    and len(response.assertion.advice.assertion) > 0:
+                _assertions = response.assertion
+                if not isinstance(_assertions, list):
+                    _assertions = [_assertions]
+                for _assertion in _assertions:
+                    _assertion.advice.encrypted_assertion = []
+                    _assertion.advice.encrypted_assertion.append(EncryptedAssertion())
+                    _advice_assertions = copy.deepcopy(_assertion.advice.assertion)
+                    _assertion.advice.assertion = []
+                    if not isinstance(_advice_assertions, list):
+                        _advice_assertions = [_advice_assertions]
+                    for tmp_assertion in _advice_assertions:
+                        to_sign_advice = []
+                        if sign_assertion and not pefim:
+                            tmp_assertion.signature = pre_signature_part(tmp_assertion.id, self.sec.my_cert, 1)
+                            to_sign_advice.append((class_name(tmp_assertion), tmp_assertion.id))
+                        #tmp_assertion = response.assertion.advice.assertion[0]
+                        _assertion.advice.encrypted_assertion[0].add_extension_element(tmp_assertion)
+
+                        if encrypt_assertion_self_contained:
+                            advice_tag = response.assertion.advice._to_element_tree().tag
+                            assertion_tag = tmp_assertion._to_element_tree().tag
+                            response = \
+                                response.get_xml_string_with_self_contained_assertion_within_advice_encrypted_assertion(
+                                    assertion_tag, advice_tag)
+                        node_xpath = ''.join(["/*[local-name()=\"%s\"]" % v for v in
+                                              ["Response", "Assertion", "Advice", "EncryptedAssertion", "Assertion"]])
+
+                        if to_sign_advice:
+                            response = signed_instance_factory(response, self.sec, to_sign_advice)
+                        response = self._encrypt_assertion(encrypt_cert_advice, sp_entity_id, response, node_xpath=node_xpath)
+                        response = response_from_string(response)
+
+            if encrypt_assertion:
+                to_sign_assertion = []
+                if sign_assertion is not None and sign_assertion:
+                    _assertions = response.assertion
+                    if not isinstance(_assertions, list):
+                        _assertions = [_assertions]
+                    for _assertion in _assertions:
+                        _assertion.signature = pre_signature_part(_assertion.id, self.sec.my_cert, 1)
+                        to_sign_assertion.append((class_name(_assertion), _assertion.id))
                 if encrypt_assertion_self_contained:
-                    advice_tag = response.assertion.advice._to_element_tree().tag
-                    assertion_tag = tmp_assertion._to_element_tree().tag
-                    response = response.get_xml_string_with_self_contained_assertion_within_advice_encrypted_assertion(
-                        assertion_tag, advice_tag)
-                node_xpath = ''.join(["/*[local-name()=\"%s\"]" % v for v in
-                                      ["Response", "Assertion", "Advice", "EncryptedAssertion", "Assertion"]])
-            elif encrypt_assertion_self_contained:
-                assertion_tag = response.assertion._to_element_tree().tag
-                response = pre_encrypt_assertion(response)
-                response = response.get_xml_string_with_self_contained_assertion_within_encrypted_assertion(
-                    assertion_tag)
+                    try:
+                        assertion_tag = response.assertion._to_element_tree().tag
+                    except:
+                        assertion_tag = response.assertion[0]._to_element_tree().tag
+                    response = pre_encrypt_assertion(response)
+                    response = response.get_xml_string_with_self_contained_assertion_within_encrypted_assertion(
+                        assertion_tag)
+                else:
+                    response = pre_encrypt_assertion(response)
+                if to_sign_assertion:
+                    response = signed_instance_factory(response, self.sec, to_sign_assertion)
+                response = self._encrypt_assertion(encrypt_cert_assertion, sp_entity_id, response)
             else:
-                response = pre_encrypt_assertion(response)
-            if to_sign:
-                response = signed_instance_factory(response, self.sec, to_sign)
-            _, cert_file = make_temp("%s" % encrypt_cert, decode=False)
-            response = cbxs.encrypt_assertion(response, cert_file,
-                                              pre_encryption_part(), node_xpath=node_xpath)
-                                              # template(response.assertion.id))
+                if to_sign:
+                    response = signed_instance_factory(response, self.sec, to_sign)
             if sign:
                 return signed_instance_factory(response, self.sec, sign_class)
             else:
@@ -968,23 +1041,23 @@ def _parse_response(self, xmlstr, response_cls, service, binding,
             logger.debug("XMLSTR: %s" % xmlstr)
 
             if response:
+                keys = None
                 if outstanding_certs:
                     try:
                         cert = outstanding_certs[response.in_response_to]
                     except KeyError:
-                        key_file = ""
+                        keys = None
                     else:
-                        _, key_file = make_temp("%s" % cert["key"],
-                                                decode=False)
-                else:
-                    key_file = ""
+                        if not isinstance(cert, list):
+                            cert = [cert]
+                        keys = []
+                        for _cert in cert:
+                            keys.append(_cert["key"])
                 only_identity_in_encrypted_assertion = False
                 if "only_identity_in_encrypted_assertion" in kwargs:
                     only_identity_in_encrypted_assertion = kwargs["only_identity_in_encrypted_assertion"]
-                decrypt = True
-                if "decrypt" in kwargs:
-                    decrypt = kwargs["decrypt"]
-                response = response.verify(key_file, decrypt=decrypt)
+
+                response = response.verify(keys)
 
             if not response:
                 return None

src/saml2/extension/pefim.py

@@ -2,7 +2,7 @@
 
 import saml2
 from saml2 import SamlBase
-from saml2.xmldsig import X509Data
+from saml2.xmldsig import KeyInfo
 
 NAMESPACE = 'urn:net:eustix:names:tc:PEFIM:0.0:assertion'
 
@@ -16,19 +16,31 @@ class SPCertEncType_(SamlBase):
     c_attributes = SamlBase.c_attributes.copy()
     c_child_order = SamlBase.c_child_order[:]
     c_cardinality = SamlBase.c_cardinality.copy()
-    c_children['{http://www.w3.org/2000/09/xmldsig#}X509Data'] = ('x509_data',
-                                                                  [X509Data])
+    c_children['{http://www.w3.org/2000/09/xmldsig#}KeyInfo'] = ('key_info',
+                                                                 [KeyInfo])
+    c_cardinality['key_info'] = {"min": 1}
+    c_attributes['VerifyDepth'] = ('verify_depth', 'unsignedByte', False)
+    c_child_order.extend(['key_info'])
 
     def __init__(self,
+                 key_info=None,
                  x509_data=None,
+                 verify_depth='1',
                  text=None,
                  extension_elements=None,
                  extension_attributes=None):
         SamlBase.__init__(self,
                           text=text,
                           extension_elements=extension_elements,
                           extension_attributes=extension_attributes)
-        self.x509_data = x509_data
+        if key_info:
+            self.key_info = key_info
+        elif x509_data:
+            self.key_info = KeyInfo(x509_data=x509_data)
+        else:
+            self.key_info = []
+        self.verify_depth = verify_depth
+        #self.x509_data = x509_data
 
 
 def spcertenc_type__from_string(xml_string):
@@ -62,5 +74,4 @@ def spcertenc_from_string(xml_string):
 
 
 def factory(tag, **kwargs):
-    return ELEMENT_BY_TAG[tag](**kwargs)
-
+    return ELEMENT_BY_TAG[tag](**kwargs)