diff --git a/build-deps.txt b/build-deps.txt
new file mode 100644
index 00000000..532a294d
--- /dev/null
+++ b/build-deps.txt
@@ -0,0 +1,3 @@
+libxmlsec1-dev
+libxml2-dev
+
diff --git a/deps.txt b/deps.txt
new file mode 100644
index 00000000..084fb198
--- /dev/null
+++ b/deps.txt
@@ -0,0 +1,6 @@
+python-defusedxml (>= 0.4.1)
+python-defusedxml (<< 0.5)
+python-isodate (>= 0.5.0)
+python-isodate (<< 0.6)
+python-xmlsec-binding (>= 1.3.2)
+python-xmlsec-binding (<< 1.4)
\ No newline at end of file
diff --git a/src/onelogin/saml2/auth.py b/src/onelogin/saml2/auth.py
index bb564e4d..5eba6816 100644
--- a/src/onelogin/saml2/auth.py
+++ b/src/onelogin/saml2/auth.py
@@ -35,7 +35,7 @@ class OneLogin_Saml2_Auth(object):
SAML Response, a Logout Request or a Logout Response).
"""
- def __init__(self, request_data, old_settings=None, custom_base_path=None):
+ def __init__(self, request_data, old_settings=None, custom_base_path=None, settings=None):
"""
Initializes the SP SAML instance.
@@ -49,7 +49,10 @@ def __init__(self, request_data, old_settings=None, custom_base_path=None):
:type custom_base_path: string
"""
self.__request_data = request_data
- self.__settings = OneLogin_Saml2_Settings(old_settings, custom_base_path)
+
+ # TS: New hotness: create OneLogin_Saml2_Settings once; add it using the settings argument -- never specify old_settings or custom_base_path.
+ self.__settings = settings if settings else OneLogin_Saml2_Settings(old_settings, custom_base_path)
+
self.__attributes = []
self.__nameid = None
self.__nameid_format = None
@@ -66,6 +69,7 @@ def __init__(self, request_data, old_settings=None, custom_base_path=None):
self.__last_response = None
def get_settings(self):
+ # TS: UNUSED!
"""
Returns the settings info
:return: Setting info
diff --git a/src/onelogin/saml2/indirect_for_mocking.py b/src/onelogin/saml2/indirect_for_mocking.py
new file mode 100644
index 00000000..df1aafa1
--- /dev/null
+++ b/src/onelogin/saml2/indirect_for_mocking.py
@@ -0,0 +1,64 @@
+from contextlib import contextmanager
+from datetime import datetime as _datetime
+from hashlib import sha1
+from time import time as _time
+from uuid import uuid4
+
+# TS: Hackish way to enable random-id and time-mocking (to test SAMLResponse's).
+#
+# Set `mocked_time_fn to a truthy-value, a function returning UNIX-Time (time.time()'s format).
+#
+# Set `mocked_random_id_fn' to a truthy-value, a function returning a `random-id-string' when called.
+
+mocked_time_fn = None
+mocked_random_id_fn = None
+
+
+@contextmanager
+def mocked_time(time_fn):
+ global mocked_time_fn
+
+ prev_time_fn = mocked_time_fn
+ try:
+ mocked_time_fn = time_fn
+ yield
+ finally:
+ mocked_time_fn = prev_time_fn
+
+@contextmanager
+def mocked_generate_unique_id(random_id_fn):
+ global mocked_random_id_fn
+
+ prev_random_id_fn = mocked_random_id_fn
+ try:
+ mocked_random_id_fn = random_id_fn
+ yield
+ finally:
+ mocked_random_id_fn = prev_random_id_fn
+
+def time():
+ if mocked_time_fn:
+ return mocked_time_fn()
+ return _time()
+
+def _mocked_datetime_utcnow():
+ return _datetime.utcfromtimestamp(mocked_time_fn())
+
+class datetime(object):
+ def __getattr__(self, attr):
+ if mocked_time_fn:
+ return self._mocked_getattr(attr)
+ return getattr(_datetime, attr)
+
+ def _mocked_getattr(self, attr):
+ if attr == 'utcnow':
+ return _mocked_datetime_utcnow
+ return getattr(_datetime, attr)
+
+
+datetime = datetime()
+
+def generate_unique_id():
+ if mocked_random_id_fn:
+ return mocked_random_id_fn()
+ return 'ONELOGIN_%s' % sha1(uuid4().hex).hexdigest() # Weird, but the original implementation.
diff --git a/src/onelogin/saml2/metadata.py b/src/onelogin/saml2/metadata.py
index c431be48..6b47608d 100644
--- a/src/onelogin/saml2/metadata.py
+++ b/src/onelogin/saml2/metadata.py
@@ -9,9 +9,11 @@
"""
-from time import gmtime, strftime, time
-from datetime import datetime
+from time import gmtime, strftime
+from datetime import datetime as datetimeClass
+from .indirect_for_mocking import time
from defusedxml.minidom import parseString
+from xml.sax.saxutils import escape as escapeXml
from onelogin.saml2.constants import OneLogin_Saml2_Constants
from onelogin.saml2.utils import OneLogin_Saml2_Utils
@@ -56,7 +58,7 @@ def builder(sp, authnsign=False, wsign=False, valid_until=None, cache_duration=N
if valid_until is None:
valid_until = int(time()) + OneLogin_Saml2_Metadata.TIME_VALID
if not isinstance(valid_until, basestring):
- if isinstance(valid_until, datetime):
+ if isinstance(valid_until, datetimeClass):
valid_until_time = valid_until.timetuple()
else:
valid_until_time = gmtime(valid_until)
@@ -161,13 +163,19 @@ def builder(sp, authnsign=False, wsign=False, valid_until=None, cache_duration=N
if len(contacts) > 0:
contacts_info = []
for (ctype, info) in contacts.items():
+ surName = info.get('surName')
+ surNameXml = ''
+ if surName:
+ surNameXml = '\n %s' % escapeXml(surName)
+
contact = """
- %(name)s
+ %(name)s%(surNameXml)s
%(email)s
""" % \
{
'type': ctype,
'name': info['givenName'],
+ 'surNameXml': surNameXml,
'email': info['emailAddress'],
}
contacts_info.append(contact)
diff --git a/src/onelogin/saml2/settings.py b/src/onelogin/saml2/settings.py
index d5db3a05..f92615b0 100644
--- a/src/onelogin/saml2/settings.py
+++ b/src/onelogin/saml2/settings.py
@@ -11,7 +11,7 @@
import json
import re
-from time import time
+from .indirect_for_mocking import time
from os.path import dirname, exists, join, sep, abspath
from xml.dom.minidom import Document
@@ -128,11 +128,12 @@ def __load_paths(self, base_path=None):
self.__paths = {
'base': base_path,
'cert': base_path + 'certs' + sep,
- 'lib': base_path + 'lib' + sep,
- 'extlib': base_path + 'extlib' + sep,
+ 'lib': base_path + 'lib' + sep, # TS: Not used (ever)!
+ 'extlib': base_path + 'extlib' + sep, # TS: Not used (ever)!
}
def __update_paths(self, settings):
+ # TS: Broken for a "custom_base_path"; use __init__'s custom_base_path argument instead!
"""
Set custom paths if necessary
"""
@@ -154,6 +155,7 @@ def get_base_path(self):
return self.__paths['base']
def get_cert_path(self):
+ # TS: accessor NOT USED! (--> self.__paths['cert'] is :-s)
"""
Returns cert path
@@ -163,6 +165,7 @@ def get_cert_path(self):
return self.__paths['cert']
def get_lib_path(self):
+ # TS: NOT USED!
"""
Returns lib path
@@ -172,6 +175,7 @@ def get_lib_path(self):
return self.__paths['lib']
def get_ext_lib_path(self):
+ # TS: NOT USED!
"""
Returns external lib path
@@ -181,6 +185,7 @@ def get_ext_lib_path(self):
return self.__paths['extlib']
def get_schemas_path(self):
+ # TS: accessor NOT USED! (--> is manually re-computed in OneLogin_Saml2_Utils.validate_xml :-s)
"""
Returns schema path
@@ -517,6 +522,7 @@ def get_sp_key(self):
if not key and exists(key_file_name):
with open(key_file_name) as f:
key = f.read()
+ self.__sp['privateKey'] = key # TS: use self.__sp settings as a cache.
return key or None
@@ -533,6 +539,7 @@ def get_sp_cert(self):
if not cert and exists(cert_file_name):
with open(cert_file_name) as f:
cert = f.read()
+ self.__sp['x509cert'] = cert # TS: use self.__sp settings as a cache.
return cert or None
diff --git a/src/onelogin/saml2/utils.py b/src/onelogin/saml2/utils.py
index dafeaa73..1e461082 100644
--- a/src/onelogin/saml2/utils.py
+++ b/src/onelogin/saml2/utils.py
@@ -12,14 +12,15 @@
import base64
from copy import deepcopy
from datetime import datetime
-import calendar
+from .indirect_for_mocking import datetime, generate_unique_id
+from calendar import timegm
from hashlib import sha1, sha256, sha384, sha512
from isodate import parse_duration as duration_parser
from lxml import etree
from defusedxml.lxml import tostring, fromstring
from os.path import basename, dirname, join
import re
-from sys import stderr
+import sys
from tempfile import NamedTemporaryFile
from textwrap import wrap
from urllib import quote_plus
@@ -114,6 +115,7 @@ def deflate_and_base64_encode(value):
@staticmethod
def validate_xml(xml, schema, debug=False):
+ # TS: added caching of schema's
"""
Validates a xml against a schema
:param xml: The xml that will be validated
@@ -139,19 +141,16 @@ def validate_xml(xml, schema, debug=False):
except Exception:
return 'unloaded_xml'
- schema_file = join(dirname(__file__), 'schemas', schema)
- f_schema = open(schema_file, 'r')
- schema_doc = etree.parse(f_schema)
- f_schema.close()
- xmlschema = etree.XMLSchema(schema_doc)
+ xmlschema = loadXmlSchemaByFilename(schema)
if not xmlschema.validate(dom):
if debug:
- stderr.write('Errors validating the metadata')
- stderr.write(':\n\n')
+ sys.stderr.write('Errors validating the metadata')
+ sys.stderr.write(':\n\n')
for error in xmlschema.error_log:
- stderr.write('%s\n' % error.message)
+ sys.stderr.write('%s\n' % error.message)
+ sys.stderr.flush()
return 'invalid_xml'
return parseString(tostring(dom, encoding='unicode').encode('utf-8'))
@@ -424,7 +423,7 @@ def generate_unique_id():
:return: A unique string
:rtype: string
"""
- return 'ONELOGIN_%s' % sha1(uuid4().hex).hexdigest()
+ return generate_unique_id()
@staticmethod
def parse_time_to_SAML(time):
@@ -457,7 +456,7 @@ def parse_SAML_to_time(timestr):
data = datetime.strptime(timestr, '%Y-%m-%dT%H:%M:%SZ')
except ValueError:
data = datetime.strptime(timestr, '%Y-%m-%dT%H:%M:%S.%fZ')
- return calendar.timegm(data.utctimetuple())
+ return timegm(data.utctimetuple())
@staticmethod
def now():
@@ -465,7 +464,7 @@ def now():
:return: unix timestamp of actual time.
:rtype: int
"""
- return calendar.timegm(datetime.utcnow().utctimetuple())
+ return timegm(datetime.utcnow().utctimetuple())
@staticmethod
def parse_duration(duration, timestamp=None):
@@ -490,7 +489,7 @@ def parse_duration(duration, timestamp=None):
data = datetime.utcnow() + timedelta
else:
data = datetime.utcfromtimestamp(timestamp) + timedelta
- return calendar.timegm(data.utctimetuple())
+ return timegm(data.utctimetuple())
@staticmethod
def get_expire_time(cache_duration=None, valid_until=None):
@@ -671,11 +670,8 @@ def generate_name_id(value, sp_nq, sp_format=None, cert=None, debug=False, nq=No
# Load the public cert
mngr = xmlsec.KeysMngr()
- file_cert = OneLogin_Saml2_Utils.write_temp_file(cert)
- key_data = xmlsec.Key.load(file_cert.name, xmlsec.KeyDataFormatCertPem, None)
- key_data.name = basename(file_cert.name)
+ key_data = xmlsec.Key.loadMemory(cert, xmlsec.KeyDataFormatCertPem, None)
mngr.addKey(key_data)
- file_cert.close()
# Prepare for encryption
enc_data = EncData(xmlsec.TransformAes128Cbc, type=xmlsec.TypeEncElement)
@@ -814,6 +810,9 @@ def write_temp_file(content):
@staticmethod
def add_sign(xml, key, cert, debug=False, sign_algorithm=OneLogin_Saml2_Constants.RSA_SHA1, digest_algorithm=OneLogin_Saml2_Constants.SHA1):
+ # TS: **Only** called from get_sp_metadata -to-> sign_metadata -to-> add_sign (here)
+ # So: caching (key, cert) -> xmlsec Key abstraction is OK; because:
+ # - There is only one (key, cert) combination per configuration.
"""
Adds signature key and senders certificate to an element (Message or
Assertion).
@@ -908,13 +907,8 @@ def add_sign(xml, key, cert, debug=False, sign_algorithm=OneLogin_Saml2_Constant
key_info.addX509Data()
dsig_ctx = xmlsec.DSigCtx()
- sign_key = xmlsec.Key.loadMemory(key, xmlsec.KeyDataFormatPem, None)
-
- file_cert = OneLogin_Saml2_Utils.write_temp_file(cert)
- sign_key.loadCert(file_cert.name, xmlsec.KeyDataFormatCertPem)
- file_cert.close()
- dsig_ctx.signKey = sign_key
+ dsig_ctx.signKey = add_sign_signkey(key=key, cert=cert)
dsig_ctx.sign(signature)
newdoc = parseString(tostring(elem, encoding='unicode').encode('utf-8'))
@@ -1153,17 +1147,13 @@ def validate_node_sign(signature_node, elem, cert=None, fingerprint=None, finger
OneLogin_Saml2_Error.CERT_NOT_FOUND
)
- file_cert = OneLogin_Saml2_Utils.write_temp_file(cert)
-
if validatecert:
mngr = xmlsec.KeysMngr()
- mngr.loadCert(file_cert.name, xmlsec.KeyDataFormatCertPem, xmlsec.KeyDataTypeTrusted)
+ mngr.loadCertMemory(cert, xmlsec.KeyDataFormatCertPem, xmlsec.KeyDataTypeTrusted)
dsig_ctx = xmlsec.DSigCtx(mngr)
else:
dsig_ctx = xmlsec.DSigCtx()
- dsig_ctx.signKey = xmlsec.Key.load(file_cert.name, xmlsec.KeyDataFormatCertPem, None)
-
- file_cert.close()
+ dsig_ctx.signKey = xmlsec.Key.loadMemory(cert, xmlsec.KeyDataFormatCertPem, None)
dsig_ctx.setEnabledKeyData([xmlsec.KeyDataX509])
@@ -1209,10 +1199,7 @@ def validate_binary_sign(signed_query, signature, cert=None, algorithm=OneLogin_
xmlsec.set_error_callback(error_callback_method)
dsig_ctx = xmlsec.DSigCtx()
-
- file_cert = OneLogin_Saml2_Utils.write_temp_file(cert)
- dsig_ctx.signKey = xmlsec.Key.load(file_cert.name, xmlsec.KeyDataFormatCertPem, None)
- file_cert.close()
+ dsig_ctx.signKey = xmlsec.Key.loadMemory(cert, xmlsec.KeyDataFormatCertPem, None)
# Sign the metadata with our private key.
sign_algorithm_transform_map = {
@@ -1254,3 +1241,36 @@ def extract_raw_query_parameter(query_string, parameter, default=''):
def case_sensitive_urlencode(to_encode, lowercase=False):
encoded = quote_plus(to_encode)
return re.sub(r"%[A-F0-9]{2}", lambda m: m.group(0).lower(), encoded) if lowercase else encoded
+
+
+def loadXmlSchemaByFilename(schemaFilename):
+ schemaFilepath = join(dirname(__file__), 'schemas', schemaFilename)
+ xmlschema = XML_SCHEMA_CACHE.get(schemaFilepath)
+ if not xmlschema:
+ with open(schemaFilepath, 'r') as f:
+ schema_doc = etree.parse(f)
+
+ xmlschema = etree.XMLSchema(schema_doc)
+ XML_SCHEMA_CACHE[schemaFilepath] = xmlschema
+
+ return xmlschema
+
+def add_sign_signkey(key, cert):
+ cacheKey = (key, cert)
+ sign_key = ADD_SIGN_SIGNKEY_CACHE.get(cacheKey)
+ if not sign_key:
+ sign_key = xmlsec.Key.loadMemory(key, xmlsec.KeyDataFormatPem, None)
+
+ file_cert = OneLogin_Saml2_Utils.write_temp_file(cert)
+ sign_key.loadCert(file_cert.name, xmlsec.KeyDataFormatCertPem)
+ file_cert.close()
+
+ ADD_SIGN_SIGNKEY_CACHE[cacheKey] = sign_key
+
+ return sign_key
+
+
+
+
+ADD_SIGN_SIGNKEY_CACHE = {} # sign_key (constructed from key and cert); by (key, cert) "data"
+XML_SCHEMA_CACHE = {} # lxml parsed xml-schema files (by file(base)name).
diff --git a/tests/src/OneLogin/saml2_tests/auth_test.py b/tests/src/OneLogin/saml2_tests/auth_test.py
index c50fc6e8..35e57a36 100644
--- a/tests/src/OneLogin/saml2_tests/auth_test.py
+++ b/tests/src/OneLogin/saml2_tests/auth_test.py
@@ -3,13 +3,16 @@
# Copyright (c) 2014, OneLogin, Inc.
# All rights reserved.
+from StringIO import StringIO
from base64 import b64decode, b64encode
-import json
+from lxml.etree import parse
from os.path import dirname, join, exists
-import unittest
from teamcity import is_running_under_teamcity
from teamcity.unittestpy import TeamcityTestRunner
-from urlparse import urlparse, parse_qs
+from urlparse import urlparse, urlsplit, parse_qs
+from zlib import decompress, MAX_WBITS
+import json
+import unittest
from onelogin.saml2.auth import OneLogin_Saml2_Auth
from onelogin.saml2.constants import OneLogin_Saml2_Constants
@@ -18,6 +21,8 @@
from onelogin.saml2.logout_request import OneLogin_Saml2_Logout_Request
from onelogin.saml2.errors import OneLogin_Saml2_Error
+from onelogin.saml2.indirect_for_mocking import mocked_generate_unique_id, mocked_time
+
class OneLogin_Saml2_Auth_Test(unittest.TestCase):
data_path = join(dirname(dirname(dirname(dirname(__file__)))), 'data')
@@ -70,6 +75,57 @@ def testGetSSOurl(self):
sso_url = settings_info['idp']['singleSignOnService']['url']
self.assertEqual(auth.get_sso_url(), sso_url)
+ def testLoginHasMockedId(self):
+ settings_info = self.loadSettingsJSON()
+ auth = OneLogin_Saml2_Auth(self.get_request(), old_settings=settings_info)
+
+ def reqIdFromSSOLoginRedirectUrl(url):
+ _, _, _, query, _ = urlsplit(url)
+ samlRequestXmlStr = decompress(b64decode(parse_qs(query)['SAMLRequest'][0]), -MAX_WBITS)
+ lxmlNode = parse(StringIO(samlRequestXmlStr))
+ reqId = lxmlNode.xpath('/samlp:AuthnRequest/@ID', namespaces=SAML_NAMESPACES)[0]
+ return reqId
+
+ _id = [0]
+ def id_fn():
+ _id[0] += 1
+ return 'MOCK_ID_{0}'.format(_id[0])
+
+ self.assertTrue(reqIdFromSSOLoginRedirectUrl(auth.login()).startswith('ONELOGIN_'))
+
+ with mocked_generate_unique_id(id_fn):
+ self.assertEquals('MOCK_ID_1', reqIdFromSSOLoginRedirectUrl(auth.login()))
+ self.assertEquals('MOCK_ID_2', reqIdFromSSOLoginRedirectUrl(auth.login()))
+
+ def testLoginHasMockedTime(self):
+ settings_info = self.loadSettingsJSON()
+ auth = OneLogin_Saml2_Auth(self.get_request(), old_settings=settings_info)
+
+ def instantFromSSOLoginRedirectUrl(url):
+ _, _, _, query, _ = urlsplit(url)
+ samlRequestXmlStr = decompress(b64decode(parse_qs(query)['SAMLRequest'][0]), -MAX_WBITS)
+ lxmlNode = parse(StringIO(samlRequestXmlStr))
+ issueInstant = lxmlNode.xpath('/samlp:AuthnRequest/@IssueInstant', namespaces=SAML_NAMESPACES)[0]
+ return issueInstant
+
+ _time = [0.0]
+ def time_fn():
+ _time[0] += 1.0
+ return _time[0]
+
+ self.assertTrue(instantFromSSOLoginRedirectUrl(auth.login()).startswith('20'))
+
+ with mocked_time(time_fn):
+ self.assertEquals('1970-01-01T00:00:01Z', instantFromSSOLoginRedirectUrl(auth.login()))
+ self.assertEquals('1970-01-01T00:00:02Z', instantFromSSOLoginRedirectUrl(auth.login()))
+
+ def testAuthInitiatedWithSettings(self):
+ # TS: a.k.a. does-not-crash-so-must-be-working.
+ settings = OneLogin_Saml2_Settings(custom_base_path=self.settings_path)
+ auth = OneLogin_Saml2_Auth(self.get_request(), settings=settings)
+ sso_url = settings.get_idp_data()['singleSignOnService']['url']
+ self.assertEqual(auth.get_sso_url(), sso_url)
+
def testGetSLOurl(self):
"""
Tests the get_slo_url method of the OneLogin_Saml2_Auth class
@@ -977,7 +1033,7 @@ def testBuildRequestSignature(self):
valid_signature = 'Pb1EXAX5TyipSJ1SndEKZstLQTsT+1D00IZAhEepBM+OkAZQSToivu3njgJu47HZiZAqgXZFgloBuuWE/+GdcSsRYEMkEkiSDWTpUr25zKYLJDSg6GNo6iAHsKSuFt46Z54Xe/keYxYP03Hdy97EwuuSjBzzgRc5tmpV+KC7+a0='
self.assertEqual(signature, valid_signature)
- settings['sp']['privatekey'] = ''
+ settings['sp']['privateKey'] = ''
settings['custom_base_path'] = u'invalid/path/'
auth2 = OneLogin_Saml2_Auth(self.get_request(), old_settings=settings)
with self.assertRaisesRegexp(OneLogin_Saml2_Error, "Trying to sign the SAMLRequest but can't load the SP private key"):
@@ -996,7 +1052,7 @@ def testBuildResponseSignature(self):
valid_signature = 'IcyWLRX6Dz3wHBfpcUaNLVDMGM3uo6z2Z11Gjq0/APPJaHboKGljffsgMVAGBml497yckq+eYKmmz+jpURV9yTj2sF9qfD6CwX2dEzSzMdRzB40X7pWyHgEJGIhs6BhaOt5oXEk4T+h3AczERqpVYFpL00yo7FNtyQkhZFpHFhM='
self.assertEqual(signature, valid_signature)
- settings['sp']['privatekey'] = ''
+ settings['sp']['privateKey'] = ''
settings['custom_base_path'] = u'invalid/path/'
auth2 = OneLogin_Saml2_Auth(self.get_request(), old_settings=settings)
with self.assertRaisesRegexp(OneLogin_Saml2_Error, "Trying to sign the SAMLResponse but can't load the SP private key"):
@@ -1165,6 +1221,11 @@ def testGetIdFromLogoutResponse(self):
auth.process_slo()
self.assertIn(auth.get_last_message_id(), '_f9ee61bd9dbf63606faa9ae3b10548d5b3656fb859')
+SAML_NAMESPACES = {
+ 'samlp': 'urn:oasis:names:tc:SAML:2.0:protocol',
+ 'saml': 'urn:oasis:names:tc:SAML:2.0:assertion',
+}
+
if __name__ == '__main__':
if is_running_under_teamcity():
runner = TeamcityTestRunner()
diff --git a/tests/src/OneLogin/saml2_tests/authn_request_test.py b/tests/src/OneLogin/saml2_tests/authn_request_test.py
index 6aee012d..8175c99e 100644
--- a/tests/src/OneLogin/saml2_tests/authn_request_test.py
+++ b/tests/src/OneLogin/saml2_tests/authn_request_test.py
@@ -17,6 +17,8 @@
from onelogin.saml2.settings import OneLogin_Saml2_Settings
from onelogin.saml2.utils import OneLogin_Saml2_Utils
+from onelogin.saml2.indirect_for_mocking import mocked_generate_unique_id
+
class OneLogin_Saml2_Authn_Request_Test(unittest.TestCase):
def loadSettingsJSON(self, filename='settings1.json'):
@@ -91,6 +93,17 @@ def testGetXML(self):
inflated = authn_request.get_xml()
self.assertRegexpMatches(inflated, '^Surname', metadata)
+
def testBuilder(self):
"""
Tests the builder method of the OneLogin_Saml2_Metadata
diff --git a/tests/src/OneLogin/saml2_tests/settings_test.py b/tests/src/OneLogin/saml2_tests/settings_test.py
index bcdd7e1f..b27c253e 100644
--- a/tests/src/OneLogin/saml2_tests/settings_test.py
+++ b/tests/src/OneLogin/saml2_tests/settings_test.py
@@ -5,6 +5,7 @@
import json
from os.path import dirname, join, exists, sep
+from time import time
import unittest
from teamcity import is_running_under_teamcity
from teamcity.unittestpy import TeamcityTestRunner
@@ -13,6 +14,8 @@
from onelogin.saml2.settings import OneLogin_Saml2_Settings
from onelogin.saml2.utils import OneLogin_Saml2_Utils
+from onelogin.saml2.indirect_for_mocking import mocked_time
+
class OneLogin_Saml2_Settings_Test(unittest.TestCase):
data_path = join(dirname(dirname(dirname(dirname(__file__)))), 'data')
@@ -113,6 +116,57 @@ def testLoadSettingsFromFile(self):
settings_3 = OneLogin_Saml2_Settings(custom_base_path=custom_base_path)
self.assertEqual(len(settings_3.get_errors()), 0)
+ def testMocked_time_call_in_validate_xml(self):
+ custom_base_path = join(dirname(dirname(dirname(dirname(__file__)))), 'settings')
+ settings = OneLogin_Saml2_Settings(custom_base_path=custom_base_path)
+ self.assertEqual(len(settings.get_errors()), 0)
+
+ metadata = settings.get_sp_metadata()
+
+ # Original fn
+ errors = settings.validate_metadata(metadata)
+ self.assertFalse(errors, errors)
+
+ delta = 3600 * 24 * 10**6
+ _time = [float(time() + delta)]
+ _log = []
+ def time_fn():
+ _time[0] += 1.0
+ _log.append(True)
+ return _time[0]
+
+ # Mocked
+ with mocked_time(time_fn):
+ errors = settings.validate_metadata(metadata)
+ self.assertEquals(['expired_xml'], errors)
+ self.assertEquals(2, len(_log))
+
+ # Reverted fn
+ errors = settings.validate_metadata(metadata)
+ self.assertFalse(errors, errors)
+
+ def testMocked_time_call_in_metadata_builder(self):
+ # called via get_sp_metadata()
+ custom_base_path = join(dirname(dirname(dirname(dirname(__file__)))), 'settings')
+ settings = OneLogin_Saml2_Settings(custom_base_path=custom_base_path)
+ self.assertEqual(len(settings.get_errors()), 0)
+
+ delta = 3600 * 24 * 10**6
+ _time = [float(time() + delta)]
+ _log = []
+ def time_fn():
+ _time[0] += 1.0
+ _log.append(True)
+ return _time[0]
+
+ # Mocked
+ with mocked_time(time_fn):
+ metadata = settings.get_sp_metadata()
+ self.assertEquals(1, len(_log))
+ errors = settings.validate_metadata(metadata)
+ self.assertFalse(errors)
+ self.assertEquals(3, len(_log))
+
def testGetCertPath(self):
"""
Tests getCertPath method of the OneLogin_Saml2_Settings
@@ -428,6 +482,8 @@ def testGetSPMetadataSigned(self):
# Now try again with SP keys set directly from files that no exists:
settings_info['custom_base_path'] = '../path/not/exists/'
+ del settings_info['sp']['privateKey']
+ del settings_info['sp']['x509cert']
with self.assertRaises(OneLogin_Saml2_Error):
OneLogin_Saml2_Settings(settings_info).get_sp_metadata()
diff --git a/tests/src/OneLogin/saml2_tests/utils_test.py b/tests/src/OneLogin/saml2_tests/utils_test.py
index 0b9772f2..0e121d77 100644
--- a/tests/src/OneLogin/saml2_tests/utils_test.py
+++ b/tests/src/OneLogin/saml2_tests/utils_test.py
@@ -8,6 +8,7 @@
from defusedxml.lxml import fromstring
from lxml import etree
from os.path import dirname, join, exists
+from time import sleep
import unittest
from teamcity import is_running_under_teamcity
from teamcity.unittestpy import TeamcityTestRunner
@@ -18,6 +19,8 @@
from onelogin.saml2.utils import OneLogin_Saml2_Utils
from onelogin.saml2.errors import OneLogin_Saml2_Error, OneLogin_Saml2_ValidationError
+from onelogin.saml2.indirect_for_mocking import mocked_time, mocked_generate_unique_id
+
class OneLogin_Saml2_Utils_Test(unittest.TestCase):
data_path = join(dirname(dirname(dirname(dirname(__file__)))), 'data')
@@ -502,6 +505,48 @@ def testGetExpireTime(self):
self.assertNotEqual('3311642371', OneLogin_Saml2_Utils.get_expire_time('PT360000S', '2074-12-10T04:39:31Z'))
self.assertNotEqual('3311642371', OneLogin_Saml2_Utils.get_expire_time('PT360000S', 1418186371))
+ def testMocking_generate_unique_id(self):
+ _id = [0]
+ def id_fn():
+ _id[0] += 1
+ return 'MOCKID_{0}'.format(_id[0])
+
+ # Original fn
+ anId1 = OneLogin_Saml2_Utils.generate_unique_id()
+ self.assertTrue(anId1.startswith('ONELOGIN_'), anId1)
+
+ # Mocked
+ with mocked_generate_unique_id(id_fn):
+ self.assertEquals('MOCKID_1', OneLogin_Saml2_Utils.generate_unique_id())
+ self.assertEquals('MOCKID_2', OneLogin_Saml2_Utils.generate_unique_id())
+
+ # Reverted fn
+ anId2 = OneLogin_Saml2_Utils.generate_unique_id()
+ self.assertTrue(anId2.startswith('ONELOGIN_'), anId2)
+ self.assertNotEqual(anId1, anId2)
+
+ def testMocking_datetime_utcnow(self):
+ _time = [0.0]
+ def time_fn():
+ _time[0] += 1.0
+ return _time[0]
+
+ # Original fn
+ now1 = OneLogin_Saml2_Utils.now()
+ self.assertEquals(int, type(now1))
+ self.assertTrue(1400000000 < now1) # Seems OK
+
+ # Mocked
+ with mocked_time(time_fn):
+ self.assertEquals(1, OneLogin_Saml2_Utils.now())
+ self.assertEquals(2, OneLogin_Saml2_Utils.now())
+
+ # Reverted fn
+ sleep(1.01) # Make sure at least one second has lapsed.
+ now2 = OneLogin_Saml2_Utils.now()
+ self.assertTrue(now1 < now2, repr((now1, now2)))
+ self.assertTrue(0 < (now2 - now1) < 3)
+
def testQuery(self):
"""
Tests the query method of the OneLogin_Saml2_Utils