Fixed handling of signed and then encrypted response assertions. At the same time added support for dealing with any combination of encrypted/non-encrypted assertions.

Author: Roland Hedberg

src/saml2/assertion.py

@@ -543,107 +543,109 @@ class EntityCategories(object):
     pass
 
 
-class Assertion(dict):
-    """ Handles assertions about subjects """
-    
-    def __init__(self, dic=None):
-        dict.__init__(self, dic)
-        self.acs = []
-
-    @staticmethod
-    def _authn_context_decl(decl, authn_auth=None):
-        """
-        Construct the authn context with a authn context declaration
-        :param decl: The authn context declaration
-        :param authn_auth: Authenticating Authority
-        :return: An AuthnContext instance
-        """
+def _authn_context_class_ref(authn_class, authn_auth=None):
+    """
+    Construct the authn context with a authn context class reference
+    :param authn_class: The authn context class reference
+    :param authn_auth: Authenticating Authority
+    :return: An AuthnContext instance
+    """
+    cntx_class = factory(saml.AuthnContextClassRef, text=authn_class)
+    if authn_auth:
         return factory(saml.AuthnContext,
-                       authn_context_decl=decl,
+                       authn_context_class_ref=cntx_class,
                        authenticating_authority=factory(
                            saml.AuthenticatingAuthority, text=authn_auth))
-
-    def _authn_context_decl_ref(self, decl_ref, authn_auth=None):
-        """
-        Construct the authn context with a authn context declaration reference
-        :param decl_ref: The authn context declaration reference
-        :param authn_auth: Authenticating Authority
-        :return: An AuthnContext instance
-        """
+    else:
         return factory(saml.AuthnContext,
-                       authn_context_decl_ref=decl_ref,
-                       authenticating_authority=factory(
-                           saml.AuthenticatingAuthority, text=authn_auth))
+                       authn_context_class_ref=cntx_class)
 
-    @staticmethod
-    def _authn_context_class_ref(authn_class, authn_auth=None):
-        """
-        Construct the authn context with a authn context class reference
-        :param authn_class: The authn context class reference
-        :param authn_auth: Authenticating Authority
-        :return: An AuthnContext instance
-        """
-        cntx_class = factory(saml.AuthnContextClassRef, text=authn_class)
-        if authn_auth:
-            return factory(saml.AuthnContext, 
-                           authn_context_class_ref=cntx_class,
-                           authenticating_authority=factory(
-                               saml.AuthenticatingAuthority, text=authn_auth))
-        else:
-            return factory(saml.AuthnContext,
-                           authn_context_class_ref=cntx_class)
-        
-    def _authn_statement(self, authn_class=None, authn_auth=None,
-                         authn_decl=None, authn_decl_ref=None, authn_instant="",
-                         subject_locality=""):
-        """
-        Construct the AuthnStatement
-        :param authn_class: Authentication Context Class reference
-        :param authn_auth: Authenticating Authority
-        :param authn_decl: Authentication Context Declaration
-        :param authn_decl_ref: Authentication Context Declaration reference
-        :param authn_instant: When the Authentication was performed.
-            Assumed to be seconds since the Epoch.
-        :param subject_locality: Specifies the DNS domain name and IP address
-            for the system from which the assertion subject was apparently
-            authenticated.
-        :return: An AuthnContext instance
-        """
-        if authn_instant:
-            _instant = instant(time_stamp=authn_instant)
-        else:
-            _instant = instant()
-
-        if authn_class:
-            res = factory(
-                saml.AuthnStatement,
-                authn_instant=_instant,
-                session_index=sid(),
-                authn_context=self._authn_context_class_ref(
-                    authn_class, authn_auth))
-        elif authn_decl:
-            res = factory(
-                saml.AuthnStatement,
-                authn_instant=_instant,
-                session_index=sid(),
-                authn_context=self._authn_context_decl(authn_decl, authn_auth))
-        elif authn_decl_ref:
-            res = factory(
-                saml.AuthnStatement,
-                authn_instant=_instant,
-                session_index=sid(),
-                authn_context=self._authn_context_decl_ref(authn_decl_ref,
-                                                           authn_auth))
-        else:
-            res = factory(
-                saml.AuthnStatement,
-                authn_instant=_instant,
-                session_index=sid())
 
-        if subject_locality:
-            res.subject_locality = saml.SubjectLocality(text=subject_locality)
+def _authn_context_decl(decl, authn_auth=None):
+    """
+    Construct the authn context with a authn context declaration
+    :param decl: The authn context declaration
+    :param authn_auth: Authenticating Authority
+    :return: An AuthnContext instance
+    """
+    return factory(saml.AuthnContext,
+                   authn_context_decl=decl,
+                   authenticating_authority=factory(
+                       saml.AuthenticatingAuthority, text=authn_auth))
 
-        return res
+
+def _authn_context_decl_ref(decl_ref, authn_auth=None):
+    """
+    Construct the authn context with a authn context declaration reference
+    :param decl_ref: The authn context declaration reference
+    :param authn_auth: Authenticating Authority
+    :return: An AuthnContext instance
+    """
+    return factory(saml.AuthnContext,
+                   authn_context_decl_ref=decl_ref,
+                   authenticating_authority=factory(
+                       saml.AuthenticatingAuthority, text=authn_auth))
+
+
+def authn_statement(authn_class=None, authn_auth=None,
+                    authn_decl=None, authn_decl_ref=None, authn_instant="",
+                    subject_locality=""):
+    """
+    Construct the AuthnStatement
+    :param authn_class: Authentication Context Class reference
+    :param authn_auth: Authenticating Authority
+    :param authn_decl: Authentication Context Declaration
+    :param authn_decl_ref: Authentication Context Declaration reference
+    :param authn_instant: When the Authentication was performed.
+        Assumed to be seconds since the Epoch.
+    :param subject_locality: Specifies the DNS domain name and IP address
+        for the system from which the assertion subject was apparently
+        authenticated.
+    :return: An AuthnContext instance
+    """
+    if authn_instant:
+        _instant = instant(time_stamp=authn_instant)
+    else:
+        _instant = instant()
+
+    if authn_class:
+        res = factory(
+            saml.AuthnStatement,
+            authn_instant=_instant,
+            session_index=sid(),
+            authn_context=_authn_context_class_ref(
+                authn_class, authn_auth))
+    elif authn_decl:
+        res = factory(
+            saml.AuthnStatement,
+            authn_instant=_instant,
+            session_index=sid(),
+            authn_context=_authn_context_decl(authn_decl, authn_auth))
+    elif authn_decl_ref:
+        res = factory(
+            saml.AuthnStatement,
+            authn_instant=_instant,
+            session_index=sid(),
+            authn_context=_authn_context_decl_ref(authn_decl_ref,
+                                                       authn_auth))
+    else:
+        res = factory(
+            saml.AuthnStatement,
+            authn_instant=_instant,
+            session_index=sid())
+
+    if subject_locality:
+        res.subject_locality = saml.SubjectLocality(text=subject_locality)
+
+    return res
+
+
+class Assertion(dict):
+    """ Handles assertions about subjects """
+    
+    def __init__(self, dic=None):
+        dict.__init__(self, dic)
+        self.acs = []
 
     def construct(self, sp_entity_id, in_response_to, consumer_url,
                   name_id, attrconvs, policy, issuer, authn_class=None,
@@ -695,10 +697,10 @@ def construct(self, sp_entity_id, in_response_to, consumer_url,
         conds = policy.conditions(sp_entity_id)
 
         if authn_auth or authn_class or authn_decl or authn_decl_ref:
-            _authn_statement = self._authn_statement(authn_class, authn_auth,
-                                                     authn_decl, authn_decl_ref,
-                                                     authn_instant,
-                                                     subject_locality)
+            _authn_statement = authn_statement(authn_class, authn_auth,
+                                               authn_decl, authn_decl_ref,
+                                               authn_instant,
+                                               subject_locality)
         else:
             _authn_statement = None
 

src/saml2/entity.py

@@ -852,14 +852,6 @@ def _parse_response(self, xmlstr, response_cls, service, binding,
 
             xmlstr = self.unravel(xmlstr, binding, response_cls.msgtype)
             origxml = xmlstr
-            if outstanding_certs is not None:
-                _response = samlp.any_response_from_string(xmlstr)
-                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
 
@@ -878,18 +870,14 @@ def _parse_response(self, xmlstr, response_cls, service, binding,
 
             logger.debug("XMLSTR: %s" % xmlstr)
 
-            if hasattr(response.response, 'encrypted_assertion'):
-                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()
+                if outstanding_certs is not None:
+                    _, key_file = make_temp(
+                        "%s" % outstanding_certs[
+                            response.in_response_to]["key"], decode=False)
+                else:
+                    key_file = ""
+                response = response.verify(key_file)
 
             if not response:
                 return None