Fixes for EncryptedAssertion and signing.

Author: Hans Hörberg

example/idp2/idp.py

@@ -301,7 +301,7 @@ def do(self, query, binding_in, relay_state="", encrypt_cert=None):
             try:
                 _resp = IDP.create_authn_response(
                     identity, userid=self.user,
-                    authn=AUTHN_BROKER[self.environ["idp.authn_ref"]], sign_response=False, encrypt_cert=encrypt_cert,
+                    authn=AUTHN_BROKER[self.environ["idp.authn_ref"]], encrypt_cert=encrypt_cert,
                     **resp_args)
             except Exception, excp:
                 logging.error(exception_trace(excp))

example/sp-repoze/sp_conf.py

@@ -78,6 +78,7 @@ def generate_cert():
             #Information needed for generated cert (NO CERT) solution.
             "authn_requests_signed": "true", #Will sign the request!
             "want_assertions_signed": "false", #Demands that the assertion is signed.
+            "want_response_signed": "true",
             "allow_unsolicited": "true", #Allows the message not to be ment for this sp.
             #############################################################
             "name": "LocalTestSPHans",

src/saml2/__init__.py

@@ -558,6 +558,8 @@ def to_string(self, nspair=None):
                 except AttributeError:
                     # Backwards compatibility with ET < 1.3
                     ElementTree._namespace_map[uri] = prefix
+                except ValueError:
+                    pass
 
         return ElementTree.tostring(self._to_element_tree(), encoding="UTF-8")
 

src/saml2/client_base.py

@@ -122,8 +122,9 @@ def __init__(self, config=None, identity_cache=None, state_cache=None,
         self.allow_unsolicited = False
         self.authn_requests_signed = False
         self.want_assertions_signed = False
+        self.want_response_signed = False
         for foo in ["allow_unsolicited", "authn_requests_signed",
-                    "logout_requests_signed", "want_assertions_signed"]:
+                    "logout_requests_signed", "want_assertions_signed", "want_response_signed"]:
             v = self.config.getattr(foo, "sp")
             if v is True or v == 'true':
                 setattr(self, foo, True)
@@ -530,6 +531,7 @@ def parse_authn_request_response(self, xmlstr, binding, outstanding=None, outsta
                 "outstanding_certs": outstanding_certs,
                 "allow_unsolicited": self.allow_unsolicited,
                 "want_assertions_signed": self.want_assertions_signed,
+                "want_response_signed": self.want_response_signed,
                 "return_addrs": self.service_urls(),
                 "entity_id": self.config.entityid,
                 "attribute_converters": self.config.attribute_converters,

src/saml2/config.py

@@ -80,6 +80,7 @@
     "idp",
     "aa",
     "subject_data",
+    "want_response_signed",
     "want_assertions_signed",
     "authn_requests_signed",
     "name_form",

src/saml2/entity.py

@@ -460,10 +460,16 @@ def _response(self, in_response_to, consumer_url=None, status=None,
             return signed_instance_factory(response, self.sec, to_sign)
 
         if encrypt_assertion:
+            sign_class = [(class_name(response), response.id)]
+            if sign:
+                response.signature = pre_signature_part(response.id, self.sec.my_cert, 1)
             cbxs = CryptoBackendXmlSec1(self.config.xmlsec_binary)
             _, cert_file = make_temp("%s" % encrypt_cert, decode=False)
-            return cbxs.encrypt_assertion(response, cert_file, pre_encryption_part())#template(response.assertion.id))
-            #response = response_from_string(response_str)
+            response = cbxs.encrypt_assertion(response, cert_file, pre_encryption_part())#template(response.assertion.id))
+            if sign:
+                return signed_instance_factory(response, self.sec, sign_class)
+            else:
+                return response
 
         if sign:
             return self.sign(response, to_sign=to_sign)
@@ -811,16 +817,18 @@ def _parse_response(self, xmlstr, response_cls, service, binding, outstanding_ce
                 raise
 
             xmlstr = self.unravel(xmlstr, binding, response_cls.msgtype)
+            origxml = xmlstr
             if outstanding_certs is not None:
                 _response = samlp.any_response_from_string(xmlstr)
-                _, cert_file = make_temp("%s" % outstanding_certs[_response.in_response_to]["key"], decode=False)
-                cbxs = CryptoBackendXmlSec1(self.config.xmlsec_binary)
-                xmlstr = cbxs.decrypt(xmlstr, cert_file)
+                if len(_response.encrypted_assertion) > 0:
+                    _, cert_file = make_temp("%s" % outstanding_certs[_response.in_response_to]["key"], decode=False)
+                    cbxs = CryptoBackendXmlSec1(self.config.xmlsec_binary)
+                    xmlstr = cbxs.decrypt(xmlstr, cert_file)
             if not xmlstr:  # Not a valid reponse
                 return None
 
             try:
-                response = response.loads(xmlstr, False)
+                response = response.loads(xmlstr, False, origxml=origxml)
             except SigverError, err:
                 logger.error("Signature Error: %s" % err)
                 return None
@@ -831,6 +839,13 @@ def _parse_response(self, xmlstr, response_cls, service, binding, outstanding_ce
 
             logger.debug("XMLSTR: %s" % xmlstr)
 
+            for encrypted_assertion in response.response.encrypted_assertion:
+                if encrypted_assertion.extension_elements is not None:
+                    assertion_list = extension_elements_to_elements(encrypted_assertion.extension_elements, [saml])
+                    for assertion in assertion_list:
+                        _assertion = saml.assertion_from_string(str(assertion))
+                        response.response.assertion.append(_assertion)
+
             if response:
                 response = response.verify()
 

src/saml2/response.py

@@ -268,6 +268,7 @@ def __init__(self, sec_context, return_addrs=None, timeslack=0,
         self.in_response_to = None
         self.signature_check = self.sec.correctly_signed_response
         self.require_signature = False
+        self.require_response_signature = False
         self.not_signed = False
         self.asynchop = asynchop
 
@@ -318,7 +319,9 @@ def _loads(self, xmldata, decode=True, origxml=None):
         logger.debug("xmlstr: %s" % (self.xmlstr,))
 
         try:
-            self.response = self.signature_check(xmldata, origdoc=origxml, must=self.require_signature)
+            self.response = self.signature_check(xmldata, origdoc=origxml, must=self.require_signature,
+                                                 require_response_signature=self.require_response_signature)
+
         except TypeError:
             raise
         except SignatureError:
@@ -452,7 +455,7 @@ def __init__(self, sec_context, attribute_converters, entity_id,
                  return_addrs=None, outstanding_queries=None,
                  timeslack=0, asynchop=True, allow_unsolicited=False,
                  test=False, allow_unknown_attributes=False,
-                 want_assertions_signed=False, **kwargs):
+                 want_assertions_signed=False, want_response_signed=False, **kwargs):
 
         StatusResponse.__init__(self, sec_context, return_addrs, timeslack,
                                 asynchop=asynchop)
@@ -469,6 +472,7 @@ def __init__(self, sec_context, attribute_converters, entity_id,
         self.session_not_on_or_after = 0
         self.allow_unsolicited = allow_unsolicited
         self.require_signature = want_assertions_signed
+        self.require_response_signature = want_response_signed
         self.test = test
         self.allow_unknown_attributes = allow_unknown_attributes
         #

src/saml2/server.py

@@ -427,7 +427,8 @@ def create_attribute_response(self, identity, in_response_to, destination,
     def create_authn_response(self, identity, in_response_to, destination,
                               sp_entity_id, name_id_policy=None, userid=None,
                               name_id=None, authn=None, issuer=None,
-                              sign_response=False, sign_assertion=None, encrypt_cert=None, **kwargs):
+                              sign_response=None, sign_assertion=None, encrypt_cert=None, encrypt_assertion=None,
+                              **kwargs):
         """ Constructs an AuthenticationResponse
 
         :param identity: Information about an user
@@ -465,7 +466,11 @@ def create_authn_response(self, identity, in_response_to, destination,
         if sign_response is None:
             sign_response = False
 
-        encrypt_assertion = self.config.getattr("encrypt_assertion", "idp")
+        if encrypt_assertion is None:
+            encrypt_assertion = self.config.getattr("encrypt_assertion", "idp")
+        if encrypt_assertion is None:
+            encrypt_assertion = False
+
         if encrypt_assertion:
             if encrypt_cert is not None:
                 verify_encrypt_cert = self.config.getattr("verify_encrypt_cert", "idp")

src/saml2/sigver.py

@@ -1474,7 +1474,7 @@ def correctly_signed_assertion_id_response(self, decoded_xml, must=False,
         return self.correctly_signed_message(decoded_xml, "assertion", must,
                                              origdoc, only_valid_cert)
 
-    def correctly_signed_response(self, decoded_xml, must=False, origdoc=None):
+    def correctly_signed_response(self, decoded_xml, must=False, origdoc=None, require_response_signature=False):
         """ Check if a instance is correctly signed, if we have metadata for
         the IdP that sent the info use that, if not use the key that are in
         the message if any.
@@ -1492,26 +1492,18 @@ def correctly_signed_response(self, decoded_xml, must=False, origdoc=None):
         if response.signature:
             self._check_signature(decoded_xml, response, class_name(response),
                                   origdoc)
+        elif require_response_signature:
+            raise SignatureError("Signature missing for response")
 
         if isinstance(response, Response) and (response.assertion or
                                                    response.encrypted_assertion):
             # Try to find the signing cert in the assertion
             for assertion in (
                 response.assertion or response.encrypted_assertion):
-                if response.encrypted_assertion:
-                    assertion_list = extension_elements_to_elements(assertion.extension_elements, [saml])
-                    if len(assertion_list) > 0:
-                        assertion = saml.assertion_from_string(str(assertion_list[0]))
-                        response.assertion.append(assertion)
-                    else:
-                        decoded_xml = self.decrypt(assertion.encrypted_data.to_string())
-                        assertion = saml.assertion_from_string(decoded_xml)
-                        response.assertion.append(assertion)
-
-                if not assertion.signature:
+                if not hasattr(assertion, 'signature') or not assertion.signature:
                     logger.debug("unsigned")
                     if must:
-                        raise SignatureError("Signature missing")
+                        raise SignatureError("Signature missing for assertion")
                     continue
                 else:
                     logger.debug("signed")