Refactor logout_request, now is_valid is not a static method (following the php-saml style). Added get_error methods at Logout Request and Logout Response. Refactor tests

Author: pitbulk

src/onelogin/saml2/auth.py

@@ -40,8 +40,14 @@ def __init__(self, request_data, old_settings=None, custom_base_path=None):
         """
         Initializes the SP SAML instance.
 
-        Arguments are:
-            * (dict)   old_settings. Setting data
+        :param request_data: Request Data
+        :type request_data: dict
+
+        :param settings: Optional. SAML Toolkit Settings
+        :type settings: dict|object
+
+        :param custom_base_path: Optional. Path where are stored the settings file and the cert folder
+        :type custom_base_path: string
         """
         self.__request_data = request_data
         self.__settings = OneLogin_Saml2_Settings(old_settings, custom_base_path)
@@ -124,14 +130,14 @@ def process_slo(self, keep_local_session=False, request_id=None, delete_session_
                 OneLogin_Saml2_Utils.delete_local_session(delete_session_cb)
 
         elif 'get_data' in self.__request_data and 'SAMLRequest' in self.__request_data['get_data']:
-            request = OneLogin_Saml2_Utils.decode_base64_and_inflate(self.__request_data['get_data']['SAMLRequest'])
-            if not OneLogin_Saml2_Logout_Request.is_valid(self.__settings, request, self.__request_data):
+            logout_request = OneLogin_Saml2_Logout_Request(self.__settings, self.__request_data['get_data']['SAMLRequest'])
+            if not logout_request.is_valid(self.__request_data):
                 self.__errors.append('invalid_logout_request')
             else:
                 if not keep_local_session:
                     OneLogin_Saml2_Utils.delete_local_session(delete_session_cb)
 
-                in_response_to = OneLogin_Saml2_Logout_Request.get_id(request)
+                in_response_to = OneLogin_Saml2_Logout_Request.get_id(OneLogin_Saml2_Utils.decode_base64_and_inflate(self.__request_data['get_data']['SAMLRequest']))
                 response_builder = OneLogin_Saml2_Logout_Response(self.__settings)
                 response_builder.build(in_response_to)
                 logout_response = response_builder.get_response()

src/onelogin/saml2/logout_request.py

@@ -9,11 +9,12 @@
 
 """
 
+from zlib import decompress
 from base64 import b64decode
+from lxml import etree
 from defusedxml.lxml import fromstring
 from urllib import quote_plus
 from xml.dom.minidom import Document
-from defusedxml.minidom import parseString
 
 from onelogin.saml2.constants import OneLogin_Saml2_Constants
 from onelogin.saml2.utils import OneLogin_Saml2_Utils
@@ -28,51 +29,61 @@ class OneLogin_Saml2_Logout_Request(object):
 
     """
 
-    def __init__(self, settings):
+    def __init__(self, settings, request=None):
         """
         Constructs the Logout Request object.
 
         Arguments are:
             * (OneLogin_Saml2_Settings)   settings. Setting data
         """
         self.__settings = settings
+        self.__error = None
 
-        sp_data = self.__settings.get_sp_data()
-        idp_data = self.__settings.get_idp_data()
-        security = self.__settings.get_security_data()
-
-        uid = OneLogin_Saml2_Utils.generate_unique_id()
-        name_id_value = OneLogin_Saml2_Utils.generate_unique_id()
-        issue_instant = OneLogin_Saml2_Utils.parse_time_to_SAML(OneLogin_Saml2_Utils.now())
-
-        cert = None
-        if 'nameIdEncrypted' in security and security['nameIdEncrypted']:
-            cert = idp_data['x509cert']
-
-        name_id = OneLogin_Saml2_Utils.generate_name_id(
-            name_id_value,
-            sp_data['entityId'],
-            sp_data['NameIDFormat'],
-            cert
-        )
-
-        logout_request = """<samlp:LogoutRequest
-    xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol"
-    xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion"
-    ID="%(id)s"
-    Version="2.0"
-    IssueInstant="%(issue_instant)s"
-    Destination="%(single_logout_url)s">
-    <saml:Issuer>%(entity_id)s</saml:Issuer>
-    %(name_id)s
-</samlp:LogoutRequest>""" % \
-            {
-                'id': uid,
-                'issue_instant': issue_instant,
-                'single_logout_url': idp_data['singleLogoutService']['url'],
-                'entity_id': sp_data['entityId'],
-                'name_id': name_id,
-            }
+        if request is None:
+            sp_data = self.__settings.get_sp_data()
+            idp_data = self.__settings.get_idp_data()
+            security = self.__settings.get_security_data()
+
+            uid = OneLogin_Saml2_Utils.generate_unique_id()
+            name_id_value = OneLogin_Saml2_Utils.generate_unique_id()
+            issue_instant = OneLogin_Saml2_Utils.parse_time_to_SAML(OneLogin_Saml2_Utils.now())
+
+            cert = None
+            if 'nameIdEncrypted' in security and security['nameIdEncrypted']:
+                cert = idp_data['x509cert']
+
+            name_id = OneLogin_Saml2_Utils.generate_name_id(
+                name_id_value,
+                sp_data['entityId'],
+                sp_data['NameIDFormat'],
+                cert
+            )
+
+            logout_request = """<samlp:LogoutRequest
+        xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol"
+        xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion"
+        ID="%(id)s"
+        Version="2.0"
+        IssueInstant="%(issue_instant)s"
+        Destination="%(single_logout_url)s">
+        <saml:Issuer>%(entity_id)s</saml:Issuer>
+        %(name_id)s
+    </samlp:LogoutRequest>""" % \
+                {
+                    'id': uid,
+                    'issue_instant': issue_instant,
+                    'single_logout_url': idp_data['singleLogoutService']['url'],
+                    'entity_id': sp_data['entityId'],
+                    'name_id': name_id,
+                }
+        else:
+            decoded = b64decode(request)
+            # We try to inflate
+            try:
+                inflated = decompress(decoded, -15)
+                logout_request = inflated
+            except Exception:
+                logout_request = decoded
 
         self.__logout_request = logout_request
 
@@ -93,11 +104,13 @@ def get_id(request):
         :return: string ID
         :rtype: str object
         """
-        if isinstance(request, Document):
-            dom = request
+        if isinstance(request, etree._Element):
+            elem = request
         else:
-            dom = parseString(request)
-        return dom.documentElement.getAttribute('ID')
+            if isinstance(request, Document):
+                request = request.toxml()
+            elem = fromstring(request)
+        return elem.get('ID', None)
 
     @staticmethod
     def get_nameid_data(request, key=None):
@@ -110,23 +123,26 @@ def get_nameid_data(request, key=None):
         :return: Name ID Data (Value, Format, NameQualifier, SPNameQualifier)
         :rtype: dict
         """
-        if isinstance(request, Document):
-            request = request.toxml()
-        doc = fromstring(request)
+        if isinstance(request, etree._Element):
+            elem = request
+        else:
+            if isinstance(request, Document):
+                request = request.toxml()
+            elem = fromstring(request)
 
         name_id = None
-        encrypted_entries = OneLogin_Saml2_Utils.query(doc, '/samlp:LogoutRequest/saml:EncryptedID')
+        encrypted_entries = OneLogin_Saml2_Utils.query(elem, '/samlp:LogoutRequest/saml:EncryptedID')
 
         if len(encrypted_entries) == 1:
             if key is None:
                 raise Exception('Key is required in order to decrypt the NameID')
 
-            encrypted_data_nodes = OneLogin_Saml2_Utils.query(doc, '/samlp:LogoutRequest/saml:EncryptedID/xenc:EncryptedData')
+            encrypted_data_nodes = OneLogin_Saml2_Utils.query(elem, '/samlp:LogoutRequest/saml:EncryptedID/xenc:EncryptedData')
             if len(encrypted_data_nodes) == 1:
                 encrypted_data = encrypted_data_nodes[0]
                 name_id = OneLogin_Saml2_Utils.decrypt_element(encrypted_data, key)
         else:
-            entries = OneLogin_Saml2_Utils.query(doc, '/samlp:LogoutRequest/saml:NameID')
+            entries = OneLogin_Saml2_Utils.query(elem, '/samlp:LogoutRequest/saml:NameID')
             if len(entries) == 1:
                 name_id = entries[0]
 
@@ -165,12 +181,15 @@ def get_issuer(request):
         :return: The Issuer
         :rtype: string
         """
-        if isinstance(request, Document):
-            request = request.toxml()
-        dom = fromstring(request)
+        if isinstance(request, etree._Element):
+            elem = request
+        else:
+            if isinstance(request, Document):
+                request = request.toxml()
+            elem = fromstring(request)
 
         issuer = None
-        issuer_nodes = OneLogin_Saml2_Utils.query(dom, '/samlp:LogoutRequest/saml:Issuer')
+        issuer_nodes = OneLogin_Saml2_Utils.query(elem, '/samlp:LogoutRequest/saml:Issuer')
         if len(issuer_nodes) == 1:
             issuer = issuer_nodes[0].text
         return issuer
@@ -184,54 +203,58 @@ def get_session_indexes(request):
         :return: The SessionIndex value
         :rtype: list
         """
-        if isinstance(request, Document):
-            request = request.toxml()
-        dom = fromstring(request)
+        if isinstance(request, etree._Element):
+            elem = request
+        else:
+            if isinstance(request, Document):
+                request = request.toxml()
+            elem = fromstring(request)
 
         session_indexes = []
-        session_index_nodes = OneLogin_Saml2_Utils.query(dom, '/samlp:LogoutRequest/samlp:SessionIndex')
+        session_index_nodes = OneLogin_Saml2_Utils.query(elem, '/samlp:LogoutRequest/samlp:SessionIndex')
         for session_index_node in session_index_nodes:
             session_indexes.append(session_index_node.text)
         return session_indexes
 
-    @staticmethod
-    def is_valid(settings, request, get_data, debug=False):
+    def is_valid(self, request_data):
         """
         Checks if the Logout Request recieved is valid
-        :param settings: Settings
-        :type settings: OneLogin_Saml2_Settings
-        :param request: Logout Request Message
-        :type request: string|DOMDocument
+        :param request_data: Request Data
+        :type request_data: dict
+
         :return: If the Logout Request is or not valid
         :rtype: boolean
         """
+        self.__error = None
         try:
-            if isinstance(request, Document):
-                dom = request
-            else:
-                dom = parseString(request)
+            dom = fromstring(self.__logout_request)
 
-            idp_data = settings.get_idp_data()
+            idp_data = self.__settings.get_idp_data()
             idp_entity_id = idp_data['entityId']
 
-            if settings.is_strict():
-                res = OneLogin_Saml2_Utils.validate_xml(dom, 'saml-schema-protocol-2.0.xsd', debug)
+            if 'get_data' in request_data.keys():
+                get_data = request_data['get_data']
+            else:
+                get_data = {}
+
+            if self.__settings.is_strict():
+                res = OneLogin_Saml2_Utils.validate_xml(dom, 'saml-schema-protocol-2.0.xsd', self.__settings.is_debug_active())
                 if not isinstance(res, Document):
                     raise Exception('Invalid SAML Logout Request. Not match the saml-schema-protocol-2.0.xsd')
 
-                security = settings.get_security_data()
+                security = self.__settings.get_security_data()
 
-                current_url = OneLogin_Saml2_Utils.get_self_url_no_query(get_data)
+                current_url = OneLogin_Saml2_Utils.get_self_url_no_query(request_data)
 
                 # Check NotOnOrAfter
-                if dom.documentElement.hasAttribute('NotOnOrAfter'):
-                    na = OneLogin_Saml2_Utils.parse_SAML_to_time(dom.documentElement.getAttribute('NotOnOrAfter'))
+                if dom.get('NotOnOrAfter', None):
+                    na = OneLogin_Saml2_Utils.parse_SAML_to_time(dom.get('NotOnOrAfter'))
                     if na <= OneLogin_Saml2_Utils.now():
                         raise Exception('Timing issues (please check your clock settings)')
 
                 # Check destination
-                if dom.documentElement.hasAttribute('Destination'):
-                    destination = dom.documentElement.getAttribute('Destination')
+                if dom.get('Destination', None):
+                    destination = dom.get('Destination')
                     if destination != '':
                         if current_url not in destination:
                             raise Exception('The LogoutRequest was received at $currentURL instead of $destination')
@@ -268,7 +291,14 @@ def is_valid(settings, request, get_data, debug=False):
 
             return True
         except Exception as err:
-            debug = settings.is_debug_active()
+            self.__error = err.__str__()
+            debug = self.__settings.is_debug_active()
             if debug:
-                print err
+                print err.__str__()
             return False
+
+    def get_error(self):
+        """
+        After execute a validation process, if fails this method returns the cause
+        """
+        return self.__error

src/onelogin/saml2/logout_response.py

@@ -39,6 +39,8 @@ def __init__(self, settings, response=None):
                                                     response from the IdP.
         """
         self.__settings = settings
+        self.__error = None
+
         if response is not None:
             self.__logout_response = OneLogin_Saml2_Utils.decode_base64_and_inflate(response)
             self.document = parseString(self.__logout_response)
@@ -75,6 +77,7 @@ def is_valid(self, request_data, request_id=None):
         :return: Returns if the SAML LogoutResponse is or not valid
         :rtype: boolean
         """
+        self.__error = None
         try:
             idp_data = self.__settings.get_idp_data()
             idp_entity_id = idp_data['entityId']
@@ -134,9 +137,10 @@ def is_valid(self, request_data, request_id=None):
 
             return True
         except Exception as err:
+            self.__error = err.__str__()
             debug = self.__settings.is_debug_active()
             if debug:
-                print err
+                print err.__str__()
             return False
 
     def __query(self, query):
@@ -193,3 +197,9 @@ def get_response(self):
         :rtype: string
         """
         return OneLogin_Saml2_Utils.deflate_and_base64_encode(self.__logout_response)
+
+    def get_error(self):
+        """
+        After execute a validation process, if fails this method returns the cause
+        """
+        return self.__error

src/onelogin/saml2/response.py

@@ -53,7 +53,10 @@ def __init__(self, settings, response):
 
     def is_valid(self, request_data, request_id=None):
         """
-        Constructs the response object.
+        Validates the response object.
+
+        :param request_data: Request Data
+        :type request_data: dict
 
         :param request_id: Optional argument. The ID of the AuthNRequest sent by this SP to the IdP
         :type request_id: string

src/onelogin/saml2/utils.py

@@ -15,7 +15,7 @@
 from hashlib import sha1
 from isodate import parse_duration as duration_parser
 from lxml import etree
-from defusedxml.lxml import fromstring
+from defusedxml.lxml import tostring, fromstring
 from os.path import basename, dirname, join
 import re
 from sys import stderr
@@ -97,11 +97,13 @@ def validate_xml(xml, schema, debug=False):
         :returns: Error code or the DomDocument of the xml
         :rtype: string
         """
-        assert isinstance(xml, basestring) or isinstance(xml, Document)
+        assert isinstance(xml, basestring) or isinstance(xml, Document) or isinstance(xml, etree._Element)
         assert isinstance(schema, basestring)
 
         if isinstance(xml, Document):
             xml = xml.toxml()
+        elif isinstance(xml, etree._Element):
+            xml = tostring(xml)
 
         # Switch to lxml for schema validation
         try: