"""tests for passlib.context this file is a clone of the 1.5 test_context.py, containing the tests using the legacy CryptPolicy api. it's being preserved here to ensure the old api doesn't break (until Passlib 1.8, when this and the legacy api will be removed). """ #========================================================= #imports #========================================================= from __future__ import with_statement #core import hashlib from logging import getLogger import os import time import warnings import sys #site try: from pkg_resources import resource_filename except ImportError: resource_filename = None #pkg from passlib import hash from passlib.context import CryptContext, CryptPolicy, LazyCryptContext from passlib.exc import PasslibConfigWarning from passlib.utils import tick, to_bytes, to_unicode from passlib.utils.compat import irange, u import passlib.utils.handlers as uh from passlib.tests.utils import TestCase, mktemp, catch_warnings, \ gae_env, set_file from passlib.registry import register_crypt_handler_path, has_crypt_handler, \ _unload_handler_name as unload_handler_name #module log = getLogger(__name__) #========================================================= # #========================================================= class CryptPolicyTest(TestCase): "test CryptPolicy object" #TODO: need to test user categories w/in all this descriptionPrefix = "CryptPolicy" #========================================================= #sample crypt policies used for testing #========================================================= #----------------------------------------------------- #sample 1 - average config file #----------------------------------------------------- #NOTE: copy of this is stored in file passlib/tests/sample_config_1s.cfg sample_config_1s = """\ [passlib] schemes = des_crypt, md5_crypt, bsdi_crypt, sha512_crypt default = md5_crypt all.vary_rounds = 10%% bsdi_crypt.max_rounds = 30000 bsdi_crypt.default_rounds = 25000 sha512_crypt.max_rounds = 50000 sha512_crypt.min_rounds = 40000 """ sample_config_1s_path = os.path.abspath(os.path.join( os.path.dirname(__file__), "sample_config_1s.cfg")) if not os.path.exists(sample_config_1s_path) and resource_filename: #in case we're zipped up in an egg. sample_config_1s_path = resource_filename("passlib.tests", "sample_config_1s.cfg") #make sure sample_config_1s uses \n linesep - tests rely on this assert sample_config_1s.startswith("[passlib]\nschemes") sample_config_1pd = dict( schemes = [ "des_crypt", "md5_crypt", "bsdi_crypt", "sha512_crypt"], default = "md5_crypt", # NOTE: not maintaining backwards compat for rendering to "10%" all__vary_rounds = 0.1, bsdi_crypt__max_rounds = 30000, bsdi_crypt__default_rounds = 25000, sha512_crypt__max_rounds = 50000, sha512_crypt__min_rounds = 40000, ) sample_config_1pid = { "schemes": "des_crypt, md5_crypt, bsdi_crypt, sha512_crypt", "default": "md5_crypt", # NOTE: not maintaining backwards compat for rendering to "10%" "all.vary_rounds": 0.1, "bsdi_crypt.max_rounds": 30000, "bsdi_crypt.default_rounds": 25000, "sha512_crypt.max_rounds": 50000, "sha512_crypt.min_rounds": 40000, } sample_config_1prd = dict( schemes = [ hash.des_crypt, hash.md5_crypt, hash.bsdi_crypt, hash.sha512_crypt], default = "md5_crypt", # NOTE: passlib <= 1.5 was handler obj. # NOTE: not maintaining backwards compat for rendering to "10%" all__vary_rounds = 0.1, bsdi_crypt__max_rounds = 30000, bsdi_crypt__default_rounds = 25000, sha512_crypt__max_rounds = 50000, sha512_crypt__min_rounds = 40000, ) #----------------------------------------------------- #sample 2 - partial policy & result of overlay on sample 1 #----------------------------------------------------- sample_config_2s = """\ [passlib] bsdi_crypt.min_rounds = 29000 bsdi_crypt.max_rounds = 35000 bsdi_crypt.default_rounds = 31000 sha512_crypt.min_rounds = 45000 """ sample_config_2pd = dict( #using this to test full replacement of existing options bsdi_crypt__min_rounds = 29000, bsdi_crypt__max_rounds = 35000, bsdi_crypt__default_rounds = 31000, #using this to test partial replacement of existing options sha512_crypt__min_rounds=45000, ) sample_config_12pd = dict( schemes = [ "des_crypt", "md5_crypt", "bsdi_crypt", "sha512_crypt"], default = "md5_crypt", # NOTE: not maintaining backwards compat for rendering to "10%" all__vary_rounds = 0.1, bsdi_crypt__min_rounds = 29000, bsdi_crypt__max_rounds = 35000, bsdi_crypt__default_rounds = 31000, sha512_crypt__max_rounds = 50000, sha512_crypt__min_rounds=45000, ) #----------------------------------------------------- #sample 3 - just changing default #----------------------------------------------------- sample_config_3pd = dict( default="sha512_crypt", ) sample_config_123pd = dict( schemes = [ "des_crypt", "md5_crypt", "bsdi_crypt", "sha512_crypt"], default = "sha512_crypt", # NOTE: not maintaining backwards compat for rendering to "10%" all__vary_rounds = 0.1, bsdi_crypt__min_rounds = 29000, bsdi_crypt__max_rounds = 35000, bsdi_crypt__default_rounds = 31000, sha512_crypt__max_rounds = 50000, sha512_crypt__min_rounds=45000, ) #----------------------------------------------------- #sample 4 - category specific #----------------------------------------------------- sample_config_4s = """ [passlib] schemes = sha512_crypt all.vary_rounds = 10%% default.sha512_crypt.max_rounds = 20000 admin.all.vary_rounds = 5%% admin.sha512_crypt.max_rounds = 40000 """ sample_config_4pd = dict( schemes = [ "sha512_crypt" ], # NOTE: not maintaining backwards compat for rendering to "10%" all__vary_rounds = 0.1, sha512_crypt__max_rounds = 20000, # NOTE: not maintaining backwards compat for rendering to "5%" admin__all__vary_rounds = 0.05, admin__sha512_crypt__max_rounds = 40000, ) #----------------------------------------------------- #sample 5 - to_string & deprecation testing #----------------------------------------------------- sample_config_5s = sample_config_1s + """\ deprecated = des_crypt admin__context__deprecated = des_crypt, bsdi_crypt """ sample_config_5pd = sample_config_1pd.copy() sample_config_5pd.update( deprecated = [ "des_crypt" ], admin__context__deprecated = [ "des_crypt", "bsdi_crypt" ], ) sample_config_5pid = sample_config_1pid.copy() sample_config_5pid.update({ "deprecated": "des_crypt", "admin.context.deprecated": "des_crypt, bsdi_crypt", }) sample_config_5prd = sample_config_1prd.copy() sample_config_5prd.update({ # XXX: should deprecated return the actual handlers in this case? # would have to modify how policy stores info, for one. "deprecated": ["des_crypt"], "admin__context__deprecated": ["des_crypt", "bsdi_crypt"], }) #========================================================= #constructors #========================================================= def setUp(self): TestCase.setUp(self) warnings.filterwarnings("ignore", r"The CryptPolicy class has been deprecated") def test_00_constructor(self): "test CryptPolicy() constructor" policy = CryptPolicy(**self.sample_config_1pd) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #check key with too many separators is rejected self.assertRaises(TypeError, CryptPolicy, schemes = [ "des_crypt", "md5_crypt", "bsdi_crypt", "sha512_crypt"], bad__key__bsdi_crypt__max_rounds = 30000, ) #check nameless handler rejected class nameless(uh.StaticHandler): name = None self.assertRaises(ValueError, CryptPolicy, schemes=[nameless]) # check scheme must be name or crypt handler self.assertRaises(TypeError, CryptPolicy, schemes=[uh.StaticHandler]) #check name conflicts are rejected class dummy_1(uh.StaticHandler): name = 'dummy_1' self.assertRaises(KeyError, CryptPolicy, schemes=[dummy_1, dummy_1]) #with unknown deprecated value self.assertRaises(KeyError, CryptPolicy, schemes=['des_crypt'], deprecated=['md5_crypt']) #with unknown default value self.assertRaises(KeyError, CryptPolicy, schemes=['des_crypt'], default='md5_crypt') def test_01_from_path_simple(self): "test CryptPolicy.from_path() constructor" #NOTE: this is separate so it can also run under GAE #test preset stored in existing file path = self.sample_config_1s_path policy = CryptPolicy.from_path(path) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #test if path missing self.assertRaises(EnvironmentError, CryptPolicy.from_path, path + 'xxx') def test_01_from_path(self): "test CryptPolicy.from_path() constructor with encodings" if gae_env: return self.skipTest("GAE doesn't offer read/write filesystem access") path = mktemp() #test "\n" linesep set_file(path, self.sample_config_1s) policy = CryptPolicy.from_path(path) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #test "\r\n" linesep set_file(path, self.sample_config_1s.replace("\n","\r\n")) policy = CryptPolicy.from_path(path) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #test with custom encoding uc2 = to_bytes(self.sample_config_1s, "utf-16", source_encoding="utf-8") set_file(path, uc2) policy = CryptPolicy.from_path(path, encoding="utf-16") self.assertEqual(policy.to_dict(), self.sample_config_1pd) def test_02_from_string(self): "test CryptPolicy.from_string() constructor" #test "\n" linesep policy = CryptPolicy.from_string(self.sample_config_1s) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #test "\r\n" linesep policy = CryptPolicy.from_string( self.sample_config_1s.replace("\n","\r\n")) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #test with unicode data = to_unicode(self.sample_config_1s) policy = CryptPolicy.from_string(data) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #test with non-ascii-compatible encoding uc2 = to_bytes(self.sample_config_1s, "utf-16", source_encoding="utf-8") policy = CryptPolicy.from_string(uc2, encoding="utf-16") self.assertEqual(policy.to_dict(), self.sample_config_1pd) #test category specific options policy = CryptPolicy.from_string(self.sample_config_4s) self.assertEqual(policy.to_dict(), self.sample_config_4pd) def test_03_from_source(self): "test CryptPolicy.from_source() constructor" #pass it a path policy = CryptPolicy.from_source(self.sample_config_1s_path) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #pass it a string policy = CryptPolicy.from_source(self.sample_config_1s) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #pass it a dict (NOTE: make a copy to detect in-place modifications) policy = CryptPolicy.from_source(self.sample_config_1pd.copy()) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #pass it existing policy p2 = CryptPolicy.from_source(policy) self.assertIs(policy, p2) #pass it something wrong self.assertRaises(TypeError, CryptPolicy.from_source, 1) self.assertRaises(TypeError, CryptPolicy.from_source, []) def test_04_from_sources(self): "test CryptPolicy.from_sources() constructor" #pass it empty list self.assertRaises(ValueError, CryptPolicy.from_sources, []) #pass it one-element list policy = CryptPolicy.from_sources([self.sample_config_1s]) self.assertEqual(policy.to_dict(), self.sample_config_1pd) #pass multiple sources policy = CryptPolicy.from_sources( [ self.sample_config_1s_path, self.sample_config_2s, self.sample_config_3pd, ]) self.assertEqual(policy.to_dict(), self.sample_config_123pd) def test_05_replace(self): "test CryptPolicy.replace() constructor" p1 = CryptPolicy(**self.sample_config_1pd) #check overlaying sample 2 p2 = p1.replace(**self.sample_config_2pd) self.assertEqual(p2.to_dict(), self.sample_config_12pd) #check repeating overlay makes no change p2b = p2.replace(**self.sample_config_2pd) self.assertEqual(p2b.to_dict(), self.sample_config_12pd) #check overlaying sample 3 p3 = p2.replace(self.sample_config_3pd) self.assertEqual(p3.to_dict(), self.sample_config_123pd) def test_06_forbidden(self): "test CryptPolicy() forbidden kwds" #salt not allowed to be set self.assertRaises(KeyError, CryptPolicy, schemes=["des_crypt"], des_crypt__salt="xx", ) self.assertRaises(KeyError, CryptPolicy, schemes=["des_crypt"], all__salt="xx", ) #schemes not allowed for category self.assertRaises(KeyError, CryptPolicy, schemes=["des_crypt"], user__context__schemes=["md5_crypt"], ) #========================================================= #reading #========================================================= def test_10_has_schemes(self): "test has_schemes() method" p1 = CryptPolicy(**self.sample_config_1pd) self.assertTrue(p1.has_schemes()) p3 = CryptPolicy(**self.sample_config_3pd) self.assertTrue(not p3.has_schemes()) def test_11_iter_handlers(self): "test iter_handlers() method" p1 = CryptPolicy(**self.sample_config_1pd) s = self.sample_config_1prd['schemes'] self.assertEqual(list(p1.iter_handlers()), s) p3 = CryptPolicy(**self.sample_config_3pd) self.assertEqual(list(p3.iter_handlers()), []) def test_12_get_handler(self): "test get_handler() method" p1 = CryptPolicy(**self.sample_config_1pd) #check by name self.assertIs(p1.get_handler("bsdi_crypt"), hash.bsdi_crypt) #check by missing name self.assertIs(p1.get_handler("sha256_crypt"), None) self.assertRaises(KeyError, p1.get_handler, "sha256_crypt", required=True) #check default self.assertIs(p1.get_handler(), hash.md5_crypt) def test_13_get_options(self): "test get_options() method" p12 = CryptPolicy(**self.sample_config_12pd) self.assertEqual(p12.get_options("bsdi_crypt"),dict( # NOTE: not maintaining backwards compat for rendering to "10%" vary_rounds = 0.1, min_rounds = 29000, max_rounds = 35000, default_rounds = 31000, )) self.assertEqual(p12.get_options("sha512_crypt"),dict( # NOTE: not maintaining backwards compat for rendering to "10%" vary_rounds = 0.1, min_rounds = 45000, max_rounds = 50000, )) p4 = CryptPolicy.from_string(self.sample_config_4s) self.assertEqual(p4.get_options("sha512_crypt"), dict( # NOTE: not maintaining backwards compat for rendering to "10%" vary_rounds=0.1, max_rounds=20000, )) self.assertEqual(p4.get_options("sha512_crypt", "user"), dict( # NOTE: not maintaining backwards compat for rendering to "10%" vary_rounds=0.1, max_rounds=20000, )) self.assertEqual(p4.get_options("sha512_crypt", "admin"), dict( # NOTE: not maintaining backwards compat for rendering to "5%" vary_rounds=0.05, max_rounds=40000, )) def test_14_handler_is_deprecated(self): "test handler_is_deprecated() method" pa = CryptPolicy(**self.sample_config_1pd) pb = CryptPolicy(**self.sample_config_5pd) self.assertFalse(pa.handler_is_deprecated("des_crypt")) self.assertFalse(pa.handler_is_deprecated(hash.bsdi_crypt)) self.assertFalse(pa.handler_is_deprecated("sha512_crypt")) self.assertTrue(pb.handler_is_deprecated("des_crypt")) self.assertFalse(pb.handler_is_deprecated(hash.bsdi_crypt)) self.assertFalse(pb.handler_is_deprecated("sha512_crypt")) #check categories as well self.assertTrue(pb.handler_is_deprecated("des_crypt", "user")) self.assertFalse(pb.handler_is_deprecated("bsdi_crypt", "user")) self.assertTrue(pb.handler_is_deprecated("des_crypt", "admin")) self.assertTrue(pb.handler_is_deprecated("bsdi_crypt", "admin")) # check deprecation is overridden per category pc = CryptPolicy( schemes=["md5_crypt", "des_crypt"], deprecated=["md5_crypt"], user__context__deprecated=["des_crypt"], ) self.assertTrue(pc.handler_is_deprecated("md5_crypt")) self.assertFalse(pc.handler_is_deprecated("des_crypt")) self.assertFalse(pc.handler_is_deprecated("md5_crypt", "user")) self.assertTrue(pc.handler_is_deprecated("des_crypt", "user")) def test_15_min_verify_time(self): "test get_min_verify_time() method" # silence deprecation warnings for min verify time warnings.filterwarnings("ignore", category=DeprecationWarning) pa = CryptPolicy() self.assertEqual(pa.get_min_verify_time(), 0) self.assertEqual(pa.get_min_verify_time('admin'), 0) pb = pa.replace(min_verify_time=.1) self.assertEqual(pb.get_min_verify_time(), .1) self.assertEqual(pb.get_min_verify_time('admin'), .1) pc = pa.replace(admin__context__min_verify_time=.2) self.assertEqual(pc.get_min_verify_time(), 0) self.assertEqual(pc.get_min_verify_time('admin'), .2) pd = pb.replace(admin__context__min_verify_time=.2) self.assertEqual(pd.get_min_verify_time(), .1) self.assertEqual(pd.get_min_verify_time('admin'), .2) #========================================================= #serialization #========================================================= def test_20_iter_config(self): "test iter_config() method" p5 = CryptPolicy(**self.sample_config_5pd) self.assertEqual(dict(p5.iter_config()), self.sample_config_5pd) self.assertEqual(dict(p5.iter_config(resolve=True)), self.sample_config_5prd) self.assertEqual(dict(p5.iter_config(ini=True)), self.sample_config_5pid) def test_21_to_dict(self): "test to_dict() method" p5 = CryptPolicy(**self.sample_config_5pd) self.assertEqual(p5.to_dict(), self.sample_config_5pd) self.assertEqual(p5.to_dict(resolve=True), self.sample_config_5prd) def test_22_to_string(self): "test to_string() method" pa = CryptPolicy(**self.sample_config_5pd) s = pa.to_string() #NOTE: can't compare string directly, ordering etc may not match pb = CryptPolicy.from_string(s) self.assertEqual(pb.to_dict(), self.sample_config_5pd) #========================================================= # #========================================================= #========================================================= #CryptContext #========================================================= class CryptContextTest(TestCase): "test CryptContext class" descriptionPrefix = "CryptContext" def setUp(self): TestCase.setUp(self) warnings.filterwarnings("ignore", r"CryptContext\(\)\.replace\(\) has been deprecated.*") warnings.filterwarnings("ignore", r"The CryptContext ``policy`` keyword has been deprecated.*") warnings.filterwarnings("ignore", ".*(CryptPolicy|context\.policy).*(has|have) been deprecated.*") #========================================================= #constructor #========================================================= def test_00_constructor(self): "test constructor" #create crypt context using handlers cc = CryptContext([hash.md5_crypt, hash.bsdi_crypt, hash.des_crypt]) c,b,a = cc.policy.iter_handlers() self.assertIs(a, hash.des_crypt) self.assertIs(b, hash.bsdi_crypt) self.assertIs(c, hash.md5_crypt) #create context using names cc = CryptContext(["md5_crypt", "bsdi_crypt", "des_crypt"]) c,b,a = cc.policy.iter_handlers() self.assertIs(a, hash.des_crypt) self.assertIs(b, hash.bsdi_crypt) self.assertIs(c, hash.md5_crypt) #TODO: test policy & other options def test_01_replace(self): "test replace()" cc = CryptContext(["md5_crypt", "bsdi_crypt", "des_crypt"]) self.assertIs(cc.policy.get_handler(), hash.md5_crypt) cc2 = cc.replace() self.assertIsNot(cc2, cc) # NOTE: was not able to maintain backward compatibility with this... ##self.assertIs(cc2.policy, cc.policy) cc3 = cc.replace(default="bsdi_crypt") self.assertIsNot(cc3, cc) # NOTE: was not able to maintain backward compatibility with this... ##self.assertIs(cc3.policy, cc.policy) self.assertIs(cc3.policy.get_handler(), hash.bsdi_crypt) def test_02_no_handlers(self): "test no handlers" #check constructor... cc = CryptContext() self.assertRaises(KeyError, cc.identify, 'hash', required=True) self.assertRaises(KeyError, cc.encrypt, 'secret') self.assertRaises(KeyError, cc.verify, 'secret', 'hash') #check updating policy after the fact... cc = CryptContext(['md5_crypt']) p = CryptPolicy(schemes=[]) cc.policy = p self.assertRaises(KeyError, cc.identify, 'hash', required=True) self.assertRaises(KeyError, cc.encrypt, 'secret') self.assertRaises(KeyError, cc.verify, 'secret', 'hash') #========================================================= #policy adaptation #========================================================= sample_policy_1 = dict( schemes = [ "des_crypt", "md5_crypt", "phpass", "bsdi_crypt", "sha256_crypt"], deprecated = [ "des_crypt", ], default = "sha256_crypt", bsdi_crypt__max_rounds = 30, bsdi_crypt__default_rounds = 25, bsdi_crypt__vary_rounds = 0, sha256_crypt__max_rounds = 3000, sha256_crypt__min_rounds = 2000, sha256_crypt__default_rounds = 3000, phpass__ident = "H", phpass__default_rounds = 7, ) def test_10_01_genconfig_settings(self): "test genconfig() settings" cc = CryptContext(policy=None, schemes=["md5_crypt", "phpass"], phpass__ident="H", phpass__default_rounds=7, ) # hash specific settings self.assertTrue(cc.genconfig().startswith("$1$")) self.assertEqual( cc.genconfig(scheme="phpass", salt='.'*8), '$H$5........', ) self.assertEqual( cc.genconfig(scheme="phpass", salt='.'*8, rounds=8, ident='P'), '$P$6........', ) # unsupported hash settings should be rejected self.assertRaises(KeyError, cc.replace, md5_crypt__ident="P") def test_10_02_genconfig_rounds_limits(self): "test genconfig() policy rounds limits" cc = CryptContext(policy=None, schemes=["sha256_crypt"], all__min_rounds=2000, all__max_rounds=3000, all__default_rounds=2500, ) # min rounds with catch_warnings(record=True) as wlog: # set below handler min c2 = cc.replace(all__min_rounds=500, all__max_rounds=None, all__default_rounds=500) self.consumeWarningList(wlog, [PasslibConfigWarning]*2) self.assertEqual(c2.genconfig(salt="nacl"), "$5$rounds=1000$nacl$") self.consumeWarningList(wlog) # below self.assertEqual( cc.genconfig(rounds=1999, salt="nacl"), '$5$rounds=2000$nacl$', ) self.consumeWarningList(wlog, PasslibConfigWarning) # equal self.assertEqual( cc.genconfig(rounds=2000, salt="nacl"), '$5$rounds=2000$nacl$', ) self.consumeWarningList(wlog) # above self.assertEqual( cc.genconfig(rounds=2001, salt="nacl"), '$5$rounds=2001$nacl$' ) self.consumeWarningList(wlog) # max rounds with catch_warnings(record=True) as wlog: # set above handler max c2 = cc.replace(all__max_rounds=int(1e9)+500, all__min_rounds=None, all__default_rounds=int(1e9)+500) self.consumeWarningList(wlog, [PasslibConfigWarning]*2) self.assertEqual(c2.genconfig(salt="nacl"), "$5$rounds=999999999$nacl$") self.consumeWarningList(wlog) # above self.assertEqual( cc.genconfig(rounds=3001, salt="nacl"), '$5$rounds=3000$nacl$' ) self.consumeWarningList(wlog, PasslibConfigWarning) # equal self.assertEqual( cc.genconfig(rounds=3000, salt="nacl"), '$5$rounds=3000$nacl$' ) self.consumeWarningList(wlog) # below self.assertEqual( cc.genconfig(rounds=2999, salt="nacl"), '$5$rounds=2999$nacl$', ) self.consumeWarningList(wlog) # explicit default rounds self.assertEqual(cc.genconfig(salt="nacl"), '$5$rounds=2500$nacl$') # fallback default rounds - use handler's default df = hash.sha256_crypt.default_rounds c2 = cc.copy(all__default_rounds=None, all__max_rounds=df<<1) self.assertEqual(c2.genconfig(salt="nacl"), '$5$rounds=%d$nacl$' % df) # fallback default rounds - use handler's, but clipped to max rounds c2 = cc.replace(all__default_rounds=None, all__max_rounds=3000) self.assertEqual(c2.genconfig(salt="nacl"), '$5$rounds=3000$nacl$') # TODO: test default falls back to mx / mn if handler has no default. #default rounds - out of bounds self.assertRaises(ValueError, cc.replace, all__default_rounds=1999) cc.policy.replace(all__default_rounds=2000) cc.policy.replace(all__default_rounds=3000) self.assertRaises(ValueError, cc.replace, all__default_rounds=3001) # invalid min/max bounds c2 = CryptContext(policy=None, schemes=["sha256_crypt"]) self.assertRaises(ValueError, c2.replace, all__min_rounds=-1) self.assertRaises(ValueError, c2.replace, all__max_rounds=-1) self.assertRaises(ValueError, c2.replace, all__min_rounds=2000, all__max_rounds=1999) def test_10_03_genconfig_linear_vary_rounds(self): "test genconfig() linear vary rounds" cc = CryptContext(policy=None, schemes=["sha256_crypt"], all__min_rounds=1995, all__max_rounds=2005, all__default_rounds=2000, ) # test negative self.assertRaises(ValueError, cc.replace, all__vary_rounds=-1) self.assertRaises(ValueError, cc.replace, all__vary_rounds="-1%") self.assertRaises(ValueError, cc.replace, all__vary_rounds="101%") # test static c2 = cc.replace(all__vary_rounds=0) self.assert_rounds_range(c2, "sha256_crypt", 2000, 2000) c2 = cc.replace(all__vary_rounds="0%") self.assert_rounds_range(c2, "sha256_crypt", 2000, 2000) # test absolute c2 = cc.replace(all__vary_rounds=1) self.assert_rounds_range(c2, "sha256_crypt", 1999, 2001) c2 = cc.replace(all__vary_rounds=100) self.assert_rounds_range(c2, "sha256_crypt", 1995, 2005) # test relative c2 = cc.replace(all__vary_rounds="0.1%") self.assert_rounds_range(c2, "sha256_crypt", 1998, 2002) c2 = cc.replace(all__vary_rounds="100%") self.assert_rounds_range(c2, "sha256_crypt", 1995, 2005) def test_10_03_genconfig_log2_vary_rounds(self): "test genconfig() log2 vary rounds" cc = CryptContext(policy=None, schemes=["bcrypt"], all__min_rounds=15, all__max_rounds=25, all__default_rounds=20, ) # test negative self.assertRaises(ValueError, cc.replace, all__vary_rounds=-1) self.assertRaises(ValueError, cc.replace, all__vary_rounds="-1%") self.assertRaises(ValueError, cc.replace, all__vary_rounds="101%") # test static c2 = cc.replace(all__vary_rounds=0) self.assert_rounds_range(c2, "bcrypt", 20, 20) c2 = cc.replace(all__vary_rounds="0%") self.assert_rounds_range(c2, "bcrypt", 20, 20) # test absolute c2 = cc.replace(all__vary_rounds=1) self.assert_rounds_range(c2, "bcrypt", 19, 21) c2 = cc.replace(all__vary_rounds=100) self.assert_rounds_range(c2, "bcrypt", 15, 25) # test relative - should shift over at 50% mark c2 = cc.replace(all__vary_rounds="1%") self.assert_rounds_range(c2, "bcrypt", 20, 20) c2 = cc.replace(all__vary_rounds="49%") self.assert_rounds_range(c2, "bcrypt", 20, 20) c2 = cc.replace(all__vary_rounds="50%") self.assert_rounds_range(c2, "bcrypt", 19, 20) c2 = cc.replace(all__vary_rounds="100%") self.assert_rounds_range(c2, "bcrypt", 15, 21) def assert_rounds_range(self, context, scheme, lower, upper): "helper to check vary_rounds covers specified range" # NOTE: this runs enough times the min and max *should* be hit, # though there's a faint chance it will randomly fail. handler = context.policy.get_handler(scheme) salt = handler.default_salt_chars[0:1] * handler.max_salt_size seen = set() for i in irange(300): h = context.genconfig(scheme, salt=salt) r = handler.from_string(h).rounds seen.add(r) self.assertEqual(min(seen), lower, "vary_rounds had wrong lower limit:") self.assertEqual(max(seen), upper, "vary_rounds had wrong upper limit:") def test_11_encrypt_settings(self): "test encrypt() honors policy settings" cc = CryptContext(**self.sample_policy_1) # hash specific settings self.assertEqual( cc.encrypt("password", scheme="phpass", salt='.'*8), '$H$5........De04R5Egz0aq8Tf.1eVhY/', ) self.assertEqual( cc.encrypt("password", scheme="phpass", salt='.'*8, ident="P"), '$P$5........De04R5Egz0aq8Tf.1eVhY/', ) # NOTE: more thorough job of rounds limits done in genconfig() test, # which is much cheaper, and shares the same codebase. # min rounds with catch_warnings(record=True) as wlog: self.assertEqual( cc.encrypt("password", rounds=1999, salt="nacl"), '$5$rounds=2000$nacl$9/lTZ5nrfPuz8vphznnmHuDGFuvjSNvOEDsGmGfsS97', ) self.consumeWarningList(wlog, PasslibConfigWarning) self.assertEqual( cc.encrypt("password", rounds=2001, salt="nacl"), '$5$rounds=2001$nacl$8PdeoPL4aXQnJ0woHhqgIw/efyfCKC2WHneOpnvF.31' ) self.consumeWarningList(wlog) # max rounds, etc tested in genconfig() # make default > max throws error if attempted self.assertRaises(ValueError, cc.replace, sha256_crypt__default_rounds=4000) def test_12_hash_needs_update(self): "test hash_needs_update() method" cc = CryptContext(**self.sample_policy_1) #check deprecated scheme self.assertTrue(cc.hash_needs_update('9XXD4trGYeGJA')) self.assertFalse(cc.hash_needs_update('$1$J8HC2RCr$HcmM.7NxB2weSvlw2FgzU0')) #check min rounds self.assertTrue(cc.hash_needs_update('$5$rounds=1999$jD81UCoo.zI.UETs$Y7qSTQ6mTiU9qZB4fRr43wRgQq4V.5AAf7F97Pzxey/')) self.assertFalse(cc.hash_needs_update('$5$rounds=2000$228SSRje04cnNCaQ$YGV4RYu.5sNiBvorQDlO0WWQjyJVGKBcJXz3OtyQ2u8')) #check max rounds self.assertFalse(cc.hash_needs_update('$5$rounds=3000$fS9iazEwTKi7QPW4$VasgBC8FqlOvD7x2HhABaMXCTh9jwHclPA9j5YQdns.')) self.assertTrue(cc.hash_needs_update('$5$rounds=3001$QlFHHifXvpFX4PLs$/0ekt7lSs/lOikSerQ0M/1porEHxYq7W/2hdFpxA3fA')) #========================================================= #identify #========================================================= def test_20_basic(self): "test basic encrypt/identify/verify functionality" handlers = [hash.md5_crypt, hash.des_crypt, hash.bsdi_crypt] cc = CryptContext(handlers, policy=None) #run through handlers for crypt in handlers: h = cc.encrypt("test", scheme=crypt.name) self.assertEqual(cc.identify(h), crypt.name) self.assertEqual(cc.identify(h, resolve=True), crypt) self.assertTrue(cc.verify('test', h)) self.assertTrue(not cc.verify('notest', h)) #test default h = cc.encrypt("test") self.assertEqual(cc.identify(h), "md5_crypt") #test genhash h = cc.genhash('secret', cc.genconfig()) self.assertEqual(cc.identify(h), 'md5_crypt') h = cc.genhash('secret', cc.genconfig(), scheme='md5_crypt') self.assertEqual(cc.identify(h), 'md5_crypt') self.assertRaises(ValueError, cc.genhash, 'secret', cc.genconfig(), scheme="des_crypt") def test_21_identify(self): "test identify() border cases" handlers = ["md5_crypt", "des_crypt", "bsdi_crypt"] cc = CryptContext(handlers, policy=None) #check unknown hash self.assertEqual(cc.identify('$9$232323123$1287319827'), None) self.assertRaises(ValueError, cc.identify, '$9$232323123$1287319827', required=True) def test_22_verify(self): "test verify() scheme kwd" handlers = ["md5_crypt", "des_crypt", "bsdi_crypt"] cc = CryptContext(handlers, policy=None) h = hash.md5_crypt.encrypt("test") #check base verify self.assertTrue(cc.verify("test", h)) self.assertTrue(not cc.verify("notest", h)) #check verify using right alg self.assertTrue(cc.verify('test', h, scheme='md5_crypt')) self.assertTrue(not cc.verify('notest', h, scheme='md5_crypt')) #check verify using wrong alg self.assertRaises(ValueError, cc.verify, 'test', h, scheme='bsdi_crypt') def test_24_min_verify_time(self): "test verify() honors min_verify_time" #NOTE: this whole test assumes time.sleep() and tick() # have better than 100ms accuracy - set via delta. delta = .05 min_delay = 2*delta min_verify_time = 5*delta max_delay = 8*delta class TimedHash(uh.StaticHandler): "psuedo hash that takes specified amount of time" name = "timed_hash" delay = 0 @classmethod def identify(cls, hash): return True def _calc_checksum(self, secret): time.sleep(self.delay) return to_unicode(secret + 'x') # silence deprecation warnings for min verify time with catch_warnings(record=True) as wlog: cc = CryptContext([TimedHash], min_verify_time=min_verify_time) self.consumeWarningList(wlog, DeprecationWarning) def timecall(func, *args, **kwds): start = tick() result = func(*args, **kwds) end = tick() return end-start, result #verify genhash delay works TimedHash.delay = min_delay elapsed, result = timecall(TimedHash.genhash, 'stub', None) self.assertEqual(result, 'stubx') self.assertAlmostEqual(elapsed, min_delay, delta=delta) #ensure min verify time is honored elapsed, result = timecall(cc.verify, "stub", "stubx") self.assertTrue(result) self.assertAlmostEqual(elapsed, min_delay, delta=delta) elapsed, result = timecall(cc.verify, "blob", "stubx") self.assertFalse(result) self.assertAlmostEqual(elapsed, min_verify_time, delta=delta) #ensure taking longer emits a warning. TimedHash.delay = max_delay with catch_warnings(record=True) as wlog: elapsed, result = timecall(cc.verify, "blob", "stubx") self.assertFalse(result) self.assertAlmostEqual(elapsed, max_delay, delta=delta) self.consumeWarningList(wlog, ".*verify exceeded min_verify_time") def test_25_verify_and_update(self): "test verify_and_update()" cc = CryptContext(**self.sample_policy_1) #create some hashes h1 = cc.encrypt("password", scheme="des_crypt") h2 = cc.encrypt("password", scheme="sha256_crypt") #check bad password, deprecated hash ok, new_hash = cc.verify_and_update("wrongpass", h1) self.assertFalse(ok) self.assertIs(new_hash, None) #check bad password, good hash ok, new_hash = cc.verify_and_update("wrongpass", h2) self.assertFalse(ok) self.assertIs(new_hash, None) #check right password, deprecated hash ok, new_hash = cc.verify_and_update("password", h1) self.assertTrue(ok) self.assertTrue(cc.identify(new_hash), "sha256_crypt") #check right password, good hash ok, new_hash = cc.verify_and_update("password", h2) self.assertTrue(ok) self.assertIs(new_hash, None) #========================================================= # border cases #========================================================= def test_30_nonstring_hash(self): "test non-string hash values cause error" # # test hash=None or some other non-string causes TypeError # and that explicit-scheme code path behaves the same. # cc = CryptContext(["des_crypt"]) for hash, kwds in [ (None, {}), (None, {"scheme": "des_crypt"}), (1, {}), ((), {}), ]: self.assertRaises(TypeError, cc.identify, hash, **kwds) self.assertRaises(TypeError, cc.genhash, 'stub', hash, **kwds) self.assertRaises(TypeError, cc.verify, 'stub', hash, **kwds) self.assertRaises(TypeError, cc.verify_and_update, 'stub', hash, **kwds) self.assertRaises(TypeError, cc.hash_needs_update, hash, **kwds) # # but genhash *should* accept None if default scheme lacks config string. # cc2 = CryptContext(["mysql323"]) self.assertRaises(TypeError, cc2.identify, None) self.assertIsInstance(cc2.genhash("stub", None), str) self.assertRaises(TypeError, cc2.verify, 'stub', None) self.assertRaises(TypeError, cc2.verify_and_update, 'stub', None) self.assertRaises(TypeError, cc2.hash_needs_update, None) def test_31_nonstring_secret(self): "test non-string password values cause error" cc = CryptContext(["des_crypt"]) hash = cc.encrypt("stub") # # test secret=None, or some other non-string causes TypeError # for secret, kwds in [ (None, {}), (None, {"scheme": "des_crypt"}), (1, {}), ((), {}), ]: self.assertRaises(TypeError, cc.encrypt, secret, **kwds) self.assertRaises(TypeError, cc.genhash, secret, hash, **kwds) self.assertRaises(TypeError, cc.verify, secret, hash, **kwds) self.assertRaises(TypeError, cc.verify_and_update, secret, hash, **kwds) #========================================================= # other #========================================================= def test_90_bcrypt_normhash(self): "teset verify_and_update / hash_needs_update corrects bcrypt padding" # see issue 25. bcrypt = hash.bcrypt PASS1 = "loppux" BAD1 = "$2a$12$oaQbBqq8JnSM1NHRPQGXORm4GCUMqp7meTnkft4zgSnrbhoKdDV0C" GOOD1 = "$2a$12$oaQbBqq8JnSM1NHRPQGXOOm4GCUMqp7meTnkft4zgSnrbhoKdDV0C" ctx = CryptContext(["bcrypt"]) with catch_warnings(record=True) as wlog: self.assertTrue(ctx.hash_needs_update(BAD1)) self.assertFalse(ctx.hash_needs_update(GOOD1)) if bcrypt.has_backend(): self.assertEqual(ctx.verify_and_update(PASS1,GOOD1), (True,None)) self.assertEqual(ctx.verify_and_update("x",BAD1), (False,None)) res = ctx.verify_and_update(PASS1, BAD1) self.assertTrue(res[0] and res[1] and res[1] != BAD1) #========================================================= #eoc #========================================================= #========================================================= #LazyCryptContext #========================================================= class dummy_2(uh.StaticHandler): name = "dummy_2" class LazyCryptContextTest(TestCase): descriptionPrefix = "LazyCryptContext" def setUp(self): # make sure this isn't registered before OR after unload_handler_name("dummy_2") self.addCleanup(unload_handler_name, "dummy_2") # silence some warnings warnings.filterwarnings("ignore", r"CryptContext\(\)\.replace\(\) has been deprecated") warnings.filterwarnings("ignore", ".*(CryptPolicy|context\.policy).*(has|have) been deprecated.*") def test_kwd_constructor(self): "test plain kwds" self.assertFalse(has_crypt_handler("dummy_2")) register_crypt_handler_path("dummy_2", "passlib.tests.test_context") cc = LazyCryptContext(iter(["dummy_2", "des_crypt"]), deprecated=["des_crypt"]) self.assertFalse(has_crypt_handler("dummy_2", True)) self.assertTrue(cc.policy.handler_is_deprecated("des_crypt")) self.assertEqual(cc.policy.schemes(), ["dummy_2", "des_crypt"]) self.assertTrue(has_crypt_handler("dummy_2", True)) def test_callable_constructor(self): "test create_policy() hook, returning CryptPolicy" self.assertFalse(has_crypt_handler("dummy_2")) register_crypt_handler_path("dummy_2", "passlib.tests.test_context") def create_policy(flag=False): self.assertTrue(flag) return CryptPolicy(schemes=iter(["dummy_2", "des_crypt"]), deprecated=["des_crypt"]) cc = LazyCryptContext(create_policy=create_policy, flag=True) self.assertFalse(has_crypt_handler("dummy_2", True)) self.assertTrue(cc.policy.handler_is_deprecated("des_crypt")) self.assertEqual(cc.policy.schemes(), ["dummy_2", "des_crypt"]) self.assertTrue(has_crypt_handler("dummy_2", True)) #========================================================= #EOF #=========================================================