From 8b5969dcff375a224b0d3a38136bd9a2b641554f Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 18:51:40 +0000 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?venv/Lib/site-packages/passlib/tests=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../passlib/tests/test_context.py | 1786 +++++++++++++++++ .../passlib/tests/test_context_deprecated.py | 743 +++++++ .../passlib/tests/test_crypto_builtin_md4.py | 160 ++ .../passlib/tests/test_crypto_des.py | 194 ++ .../passlib/tests/test_crypto_digest.py | 544 +++++ 5 files changed, 3427 insertions(+) create mode 100644 venv/Lib/site-packages/passlib/tests/test_context.py create mode 100644 venv/Lib/site-packages/passlib/tests/test_context_deprecated.py create mode 100644 venv/Lib/site-packages/passlib/tests/test_crypto_builtin_md4.py create mode 100644 venv/Lib/site-packages/passlib/tests/test_crypto_des.py create mode 100644 venv/Lib/site-packages/passlib/tests/test_crypto_digest.py diff --git a/venv/Lib/site-packages/passlib/tests/test_context.py b/venv/Lib/site-packages/passlib/tests/test_context.py new file mode 100644 index 0000000..09b52c0 --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_context.py @@ -0,0 +1,1786 @@ +"""tests for passlib.context""" +#============================================================================= +# imports +#============================================================================= +# core +from __future__ import with_statement +from passlib.utils.compat import PY3 +if PY3: + from configparser import NoSectionError +else: + from ConfigParser import NoSectionError +import datetime +from functools import partial +import logging; log = logging.getLogger(__name__) +import os +import warnings +# site +# pkg +from passlib import hash +from passlib.context import CryptContext, LazyCryptContext +from passlib.exc import PasslibConfigWarning, PasslibHashWarning +from passlib.utils import tick, to_unicode +from passlib.utils.compat import irange, u, unicode, str_to_uascii, PY2, PY26 +import passlib.utils.handlers as uh +from passlib.tests.utils import (TestCase, set_file, TICK_RESOLUTION, + quicksleep, time_call, handler_derived_from) +from passlib.registry import (register_crypt_handler_path, + _has_crypt_handler as has_crypt_handler, + _unload_handler_name as unload_handler_name, + get_crypt_handler, + ) +# local +#============================================================================= +# support +#============================================================================= +here = os.path.abspath(os.path.dirname(__file__)) + +def merge_dicts(first, *args, **kwds): + target = first.copy() + for arg in args: + target.update(arg) + if kwds: + target.update(kwds) + return target + +#============================================================================= +# +#============================================================================= +class CryptContextTest(TestCase): + descriptionPrefix = "CryptContext" + + # TODO: these unittests could really use a good cleanup + # and reorganizing, to ensure they're getting everything. + + #=================================================================== + # sample configurations used in tests + #=================================================================== + + #--------------------------------------------------------------- + # sample 1 - typical configuration + #--------------------------------------------------------------- + sample_1_schemes = ["des_crypt", "md5_crypt", "bsdi_crypt", "sha512_crypt"] + sample_1_handlers = [get_crypt_handler(name) for name in sample_1_schemes] + + sample_1_dict = dict( + schemes = sample_1_schemes, + default = "md5_crypt", + all__vary_rounds = 0.1, + bsdi_crypt__max_rounds = 30001, + bsdi_crypt__default_rounds = 25001, + sha512_crypt__max_rounds = 50000, + sha512_crypt__min_rounds = 40000, + ) + + sample_1_resolved_dict = merge_dicts(sample_1_dict, + schemes = sample_1_handlers) + + sample_1_unnormalized = u("""\ +[passlib] +schemes = des_crypt, md5_crypt, bsdi_crypt, sha512_crypt +default = md5_crypt +; this is using %... +all__vary_rounds = 10%% +bsdi_crypt__default_rounds = 25001 +bsdi_crypt__max_rounds = 30001 +sha512_crypt__max_rounds = 50000 +sha512_crypt__min_rounds = 40000 +""") + + sample_1_unicode = u("""\ +[passlib] +schemes = des_crypt, md5_crypt, bsdi_crypt, sha512_crypt +default = md5_crypt +all__vary_rounds = 0.1 +bsdi_crypt__default_rounds = 25001 +bsdi_crypt__max_rounds = 30001 +sha512_crypt__max_rounds = 50000 +sha512_crypt__min_rounds = 40000 + +""") + + #--------------------------------------------------------------- + # sample 1 external files + #--------------------------------------------------------------- + + # sample 1 string with '\n' linesep + sample_1_path = os.path.join(here, "sample1.cfg") + + # sample 1 with '\r\n' linesep + sample_1b_unicode = sample_1_unicode.replace(u("\n"), u("\r\n")) + sample_1b_path = os.path.join(here, "sample1b.cfg") + + # sample 1 using UTF-16 and alt section + sample_1c_bytes = sample_1_unicode.replace(u("[passlib]"), + u("[mypolicy]")).encode("utf-16") + sample_1c_path = os.path.join(here, "sample1c.cfg") + + # enable to regenerate sample files + if False: + set_file(sample_1_path, sample_1_unicode) + set_file(sample_1b_path, sample_1b_unicode) + set_file(sample_1c_path, sample_1c_bytes) + + #--------------------------------------------------------------- + # sample 2 & 12 - options patch + #--------------------------------------------------------------- + sample_2_dict = dict( + # using this to test full replacement of existing options + bsdi_crypt__min_rounds = 29001, + bsdi_crypt__max_rounds = 35001, + bsdi_crypt__default_rounds = 31001, + # using this to test partial replacement of existing options + sha512_crypt__min_rounds=45000, + ) + + sample_2_unicode = """\ +[passlib] +bsdi_crypt__min_rounds = 29001 +bsdi_crypt__max_rounds = 35001 +bsdi_crypt__default_rounds = 31001 +sha512_crypt__min_rounds = 45000 +""" + + # sample 2 overlayed on top of sample 1 + sample_12_dict = merge_dicts(sample_1_dict, sample_2_dict) + + #--------------------------------------------------------------- + # sample 3 & 123 - just changing default from sample 1 + #--------------------------------------------------------------- + sample_3_dict = dict( + default="sha512_crypt", + ) + + # sample 3 overlayed on 2 overlayed on 1 + sample_123_dict = merge_dicts(sample_12_dict, sample_3_dict) + + #--------------------------------------------------------------- + # sample 4 - used by api tests + #--------------------------------------------------------------- + sample_4_dict = dict( + schemes = [ "des_crypt", "md5_crypt", "phpass", "bsdi_crypt", + "sha256_crypt"], + deprecated = [ "des_crypt", ], + default = "sha256_crypt", + bsdi_crypt__max_rounds = 31, + 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, + ) + + #=================================================================== + # setup + #=================================================================== + def setUp(self): + super(CryptContextTest, self).setUp() + warnings.filterwarnings("ignore", "The 'all' scheme is deprecated.*") + warnings.filterwarnings("ignore", ".*'scheme' keyword is deprecated as of Passlib 1.7.*") + + #=================================================================== + # constructors + #=================================================================== + def test_01_constructor(self): + """test class constructor""" + + # test blank constructor works correctly + ctx = CryptContext() + self.assertEqual(ctx.to_dict(), {}) + + # test sample 1 with scheme=names + ctx = CryptContext(**self.sample_1_dict) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test sample 1 with scheme=handlers + ctx = CryptContext(**self.sample_1_resolved_dict) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test sample 2: options w/o schemes + ctx = CryptContext(**self.sample_2_dict) + self.assertEqual(ctx.to_dict(), self.sample_2_dict) + + # test sample 3: default only + ctx = CryptContext(**self.sample_3_dict) + self.assertEqual(ctx.to_dict(), self.sample_3_dict) + + # test unicode scheme names (issue 54) + ctx = CryptContext(schemes=[u("sha256_crypt")]) + self.assertEqual(ctx.schemes(), ("sha256_crypt",)) + + def test_02_from_string(self): + """test from_string() constructor""" + # test sample 1 unicode + ctx = CryptContext.from_string(self.sample_1_unicode) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test sample 1 with unnormalized inputs + ctx = CryptContext.from_string(self.sample_1_unnormalized) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test sample 1 utf-8 + ctx = CryptContext.from_string(self.sample_1_unicode.encode("utf-8")) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test sample 1 w/ '\r\n' linesep + ctx = CryptContext.from_string(self.sample_1b_unicode) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test sample 1 using UTF-16 and alt section + ctx = CryptContext.from_string(self.sample_1c_bytes, section="mypolicy", + encoding="utf-16") + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test wrong type + self.assertRaises(TypeError, CryptContext.from_string, None) + + # test missing section + self.assertRaises(NoSectionError, CryptContext.from_string, + self.sample_1_unicode, section="fakesection") + + def test_03_from_path(self): + """test from_path() constructor""" + # make sure sample files exist + if not os.path.exists(self.sample_1_path): + raise RuntimeError("can't find data file: %r" % self.sample_1_path) + + # test sample 1 + ctx = CryptContext.from_path(self.sample_1_path) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test sample 1 w/ '\r\n' linesep + ctx = CryptContext.from_path(self.sample_1b_path) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test sample 1 encoding using UTF-16 and alt section + ctx = CryptContext.from_path(self.sample_1c_path, section="mypolicy", + encoding="utf-16") + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test missing file + self.assertRaises(EnvironmentError, CryptContext.from_path, + os.path.join(here, "sample1xxx.cfg")) + + # test missing section + self.assertRaises(NoSectionError, CryptContext.from_path, + self.sample_1_path, section="fakesection") + + def test_04_copy(self): + """test copy() method""" + cc1 = CryptContext(**self.sample_1_dict) + + # overlay sample 2 onto copy + cc2 = cc1.copy(**self.sample_2_dict) + self.assertEqual(cc1.to_dict(), self.sample_1_dict) + self.assertEqual(cc2.to_dict(), self.sample_12_dict) + + # check that repeating overlay makes no change + cc2b = cc2.copy(**self.sample_2_dict) + self.assertEqual(cc1.to_dict(), self.sample_1_dict) + self.assertEqual(cc2b.to_dict(), self.sample_12_dict) + + # overlay sample 3 on copy + cc3 = cc2.copy(**self.sample_3_dict) + self.assertEqual(cc3.to_dict(), self.sample_123_dict) + + # test empty copy creates separate copy + cc4 = cc1.copy() + self.assertIsNot(cc4, cc1) + self.assertEqual(cc1.to_dict(), self.sample_1_dict) + self.assertEqual(cc4.to_dict(), self.sample_1_dict) + + # ... and that modifying copy doesn't affect original + cc4.update(**self.sample_2_dict) + self.assertEqual(cc1.to_dict(), self.sample_1_dict) + self.assertEqual(cc4.to_dict(), self.sample_12_dict) + + def test_09_repr(self): + """test repr()""" + cc1 = CryptContext(**self.sample_1_dict) + # NOTE: "0x-1234" format used by Pyston 0.5.1 (support deprecated 2019-11) + self.assertRegex(repr(cc1), "^$") + + #=================================================================== + # modifiers + #=================================================================== + def test_10_load(self): + """test load() / load_path() method""" + # NOTE: load() is the workhorse that handles all policy parsing, + # compilation, and validation. most of its features are tested + # elsewhere, since all the constructors and modifiers are just + # wrappers for it. + + # source_type 'auto' + ctx = CryptContext() + + # detect dict + ctx.load(self.sample_1_dict) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # detect unicode string + ctx.load(self.sample_1_unicode) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # detect bytes string + ctx.load(self.sample_1_unicode.encode("utf-8")) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # anything else - TypeError + self.assertRaises(TypeError, ctx.load, None) + + # NOTE: load_path() tested by from_path() + # NOTE: additional string tests done by from_string() + + # update flag - tested by update() method tests + # encoding keyword - tested by from_string() & from_path() + # section keyword - tested by from_string() & from_path() + + # test load empty + ctx = CryptContext(**self.sample_1_dict) + ctx.load({}, update=True) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # multiple loads should clear the state + ctx = CryptContext() + ctx.load(self.sample_1_dict) + ctx.load(self.sample_2_dict) + self.assertEqual(ctx.to_dict(), self.sample_2_dict) + + def test_11_load_rollback(self): + """test load() errors restore old state""" + # create initial context + cc = CryptContext(["des_crypt", "sha256_crypt"], + sha256_crypt__default_rounds=5000, + all__vary_rounds=0.1, + ) + result = cc.to_string() + + # do an update operation that should fail during parsing + # XXX: not sure what the right error type is here. + self.assertRaises(TypeError, cc.update, too__many__key__parts=True) + self.assertEqual(cc.to_string(), result) + + # do an update operation that should fail during extraction + # FIXME: this isn't failing even in broken case, need to figure out + # way to ensure some keys come after this one. + self.assertRaises(KeyError, cc.update, fake_context_option=True) + self.assertEqual(cc.to_string(), result) + + # do an update operation that should fail during compilation + self.assertRaises(ValueError, cc.update, sha256_crypt__min_rounds=10000) + self.assertEqual(cc.to_string(), result) + + def test_12_update(self): + """test update() method""" + + # empty overlay + ctx = CryptContext(**self.sample_1_dict) + ctx.update() + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + + # test basic overlay + ctx = CryptContext(**self.sample_1_dict) + ctx.update(**self.sample_2_dict) + self.assertEqual(ctx.to_dict(), self.sample_12_dict) + + # ... and again + ctx.update(**self.sample_3_dict) + self.assertEqual(ctx.to_dict(), self.sample_123_dict) + + # overlay w/ dict arg + ctx = CryptContext(**self.sample_1_dict) + ctx.update(self.sample_2_dict) + self.assertEqual(ctx.to_dict(), self.sample_12_dict) + + # overlay w/ string + ctx = CryptContext(**self.sample_1_dict) + ctx.update(self.sample_2_unicode) + self.assertEqual(ctx.to_dict(), self.sample_12_dict) + + # too many args + self.assertRaises(TypeError, ctx.update, {}, {}) + self.assertRaises(TypeError, ctx.update, {}, schemes=['des_crypt']) + + # wrong arg type + self.assertRaises(TypeError, ctx.update, None) + + #=================================================================== + # option parsing + #=================================================================== + def test_20_options(self): + """test basic option parsing""" + def parse(**kwds): + return CryptContext(**kwds).to_dict() + + # + # common option parsing tests + # + + # test keys with blank fields are rejected + # blank option + self.assertRaises(TypeError, CryptContext, __=0.1) + self.assertRaises(TypeError, CryptContext, default__scheme__='x') + + # blank scheme + self.assertRaises(TypeError, CryptContext, __option='x') + self.assertRaises(TypeError, CryptContext, default____option='x') + + # blank category + self.assertRaises(TypeError, CryptContext, __scheme__option='x') + + # test keys with too many field are rejected + self.assertRaises(TypeError, CryptContext, + category__scheme__option__invalid = 30000) + + # keys with mixed separators should be handled correctly. + # (testing actual data, not to_dict(), since re-render hid original bug) + self.assertRaises(KeyError, parse, + **{"admin.context__schemes":"md5_crypt"}) + ctx = CryptContext(**{"schemes":"md5_crypt,des_crypt", + "admin.context__default":"des_crypt"}) + self.assertEqual(ctx.default_scheme("admin"), "des_crypt") + + # + # context option -specific tests + # + + # test context option key parsing + result = dict(default="md5_crypt") + self.assertEqual(parse(default="md5_crypt"), result) + self.assertEqual(parse(context__default="md5_crypt"), result) + self.assertEqual(parse(default__context__default="md5_crypt"), result) + self.assertEqual(parse(**{"context.default":"md5_crypt"}), result) + self.assertEqual(parse(**{"default.context.default":"md5_crypt"}), result) + + # test context option key parsing w/ category + result = dict(admin__context__default="md5_crypt") + self.assertEqual(parse(admin__context__default="md5_crypt"), result) + self.assertEqual(parse(**{"admin.context.default":"md5_crypt"}), result) + + # + # hash option -specific tests + # + + # test hash option key parsing + result = dict(all__vary_rounds=0.1) + self.assertEqual(parse(all__vary_rounds=0.1), result) + self.assertEqual(parse(default__all__vary_rounds=0.1), result) + self.assertEqual(parse(**{"all.vary_rounds":0.1}), result) + self.assertEqual(parse(**{"default.all.vary_rounds":0.1}), result) + + # test hash option key parsing w/ category + result = dict(admin__all__vary_rounds=0.1) + self.assertEqual(parse(admin__all__vary_rounds=0.1), result) + self.assertEqual(parse(**{"admin.all.vary_rounds":0.1}), result) + + # settings not allowed if not in hash.setting_kwds + ctx = CryptContext(["phpass", "md5_crypt"], phpass__ident="P") + self.assertRaises(KeyError, ctx.copy, md5_crypt__ident="P") + + # hash options 'salt' and 'rounds' not allowed + self.assertRaises(KeyError, CryptContext, schemes=["des_crypt"], + des_crypt__salt="xx") + self.assertRaises(KeyError, CryptContext, schemes=["des_crypt"], + all__salt="xx") + + def test_21_schemes(self): + """test 'schemes' context option parsing""" + + # schemes can be empty + cc = CryptContext(schemes=None) + self.assertEqual(cc.schemes(), ()) + + # schemes can be list of names + cc = CryptContext(schemes=["des_crypt", "md5_crypt"]) + self.assertEqual(cc.schemes(), ("des_crypt", "md5_crypt")) + + # schemes can be comma-sep string + cc = CryptContext(schemes=" des_crypt, md5_crypt, ") + self.assertEqual(cc.schemes(), ("des_crypt", "md5_crypt")) + + # schemes can be list of handlers + cc = CryptContext(schemes=[hash.des_crypt, hash.md5_crypt]) + self.assertEqual(cc.schemes(), ("des_crypt", "md5_crypt")) + + # scheme must be name or handler + self.assertRaises(TypeError, CryptContext, schemes=[uh.StaticHandler]) + + # handlers must have a name + class nameless(uh.StaticHandler): + name = None + self.assertRaises(ValueError, CryptContext, schemes=[nameless]) + + # names must be unique + class dummy_1(uh.StaticHandler): + name = 'dummy_1' + self.assertRaises(KeyError, CryptContext, schemes=[dummy_1, dummy_1]) + + # schemes not allowed per-category + self.assertRaises(KeyError, CryptContext, + admin__context__schemes=["md5_crypt"]) + + def test_22_deprecated(self): + """test 'deprecated' context option parsing""" + def getdep(ctx, category=None): + return [name for name in ctx.schemes() + if ctx.handler(name, category).deprecated] + + # no schemes - all deprecated values allowed + cc = CryptContext(deprecated=["md5_crypt"]) + cc.update(schemes=["md5_crypt", "des_crypt"]) + self.assertEqual(getdep(cc),["md5_crypt"]) + + # deprecated values allowed if subset of schemes + cc = CryptContext(deprecated=["md5_crypt"], schemes=["md5_crypt", "des_crypt"]) + self.assertEqual(getdep(cc), ["md5_crypt"]) + + # can be handler + # XXX: allow handlers in deprecated list? not for now. + self.assertRaises(TypeError, CryptContext, deprecated=[hash.md5_crypt], + schemes=["md5_crypt", "des_crypt"]) +## cc = CryptContext(deprecated=[hash.md5_crypt], schemes=["md5_crypt", "des_crypt"]) +## self.assertEqual(getdep(cc), ["md5_crypt"]) + + # comma sep list + cc = CryptContext(deprecated="md5_crypt,des_crypt", schemes=["md5_crypt", "des_crypt", "sha256_crypt"]) + self.assertEqual(getdep(cc), ["md5_crypt", "des_crypt"]) + + # values outside of schemes not allowed + self.assertRaises(KeyError, CryptContext, schemes=['des_crypt'], + deprecated=['md5_crypt']) + + # deprecating ALL schemes should cause ValueError + self.assertRaises(ValueError, CryptContext, + schemes=['des_crypt'], + deprecated=['des_crypt']) + self.assertRaises(ValueError, CryptContext, + schemes=['des_crypt', 'md5_crypt'], + admin__context__deprecated=['des_crypt', 'md5_crypt']) + + # deprecating explicit default scheme should cause ValueError + + # ... default listed as deprecated + self.assertRaises(ValueError, CryptContext, + schemes=['des_crypt', 'md5_crypt'], + default="md5_crypt", + deprecated="md5_crypt") + + # ... global default deprecated per-category + self.assertRaises(ValueError, CryptContext, + schemes=['des_crypt', 'md5_crypt'], + default="md5_crypt", + admin__context__deprecated="md5_crypt") + + # ... category default deprecated globally + self.assertRaises(ValueError, CryptContext, + schemes=['des_crypt', 'md5_crypt'], + admin__context__default="md5_crypt", + deprecated="md5_crypt") + + # ... category default deprecated in category + self.assertRaises(ValueError, CryptContext, + schemes=['des_crypt', 'md5_crypt'], + admin__context__default="md5_crypt", + admin__context__deprecated="md5_crypt") + + # category deplist should shadow default deplist + CryptContext( + schemes=['des_crypt', 'md5_crypt'], + deprecated="md5_crypt", + admin__context__default="md5_crypt", + admin__context__deprecated=[]) + + # wrong type + self.assertRaises(TypeError, CryptContext, deprecated=123) + + # deprecated per-category + cc = CryptContext(deprecated=["md5_crypt"], + schemes=["md5_crypt", "des_crypt"], + admin__context__deprecated=["des_crypt"], + ) + self.assertEqual(getdep(cc), ["md5_crypt"]) + self.assertEqual(getdep(cc, "user"), ["md5_crypt"]) + self.assertEqual(getdep(cc, "admin"), ["des_crypt"]) + + # blank per-category deprecated list, shadowing default list + cc = CryptContext(deprecated=["md5_crypt"], + schemes=["md5_crypt", "des_crypt"], + admin__context__deprecated=[], + ) + self.assertEqual(getdep(cc), ["md5_crypt"]) + self.assertEqual(getdep(cc, "user"), ["md5_crypt"]) + self.assertEqual(getdep(cc, "admin"), []) + + def test_23_default(self): + """test 'default' context option parsing""" + + # anything allowed if no schemes + self.assertEqual(CryptContext(default="md5_crypt").to_dict(), + dict(default="md5_crypt")) + + # default allowed if in scheme list + ctx = CryptContext(default="md5_crypt", schemes=["des_crypt", "md5_crypt"]) + self.assertEqual(ctx.default_scheme(), "md5_crypt") + + # default can be handler + # XXX: sure we want to allow this ? maybe deprecate in future. + ctx = CryptContext(default=hash.md5_crypt, schemes=["des_crypt", "md5_crypt"]) + self.assertEqual(ctx.default_scheme(), "md5_crypt") + + # implicit default should be first non-deprecated scheme + ctx = CryptContext(schemes=["des_crypt", "md5_crypt"]) + self.assertEqual(ctx.default_scheme(), "des_crypt") + ctx.update(deprecated="des_crypt") + self.assertEqual(ctx.default_scheme(), "md5_crypt") + + # error if not in scheme list + self.assertRaises(KeyError, CryptContext, schemes=['des_crypt'], + default='md5_crypt') + + # wrong type + self.assertRaises(TypeError, CryptContext, default=1) + + # per-category + ctx = CryptContext(default="des_crypt", + schemes=["des_crypt", "md5_crypt"], + admin__context__default="md5_crypt") + self.assertEqual(ctx.default_scheme(), "des_crypt") + self.assertEqual(ctx.default_scheme("user"), "des_crypt") + self.assertEqual(ctx.default_scheme("admin"), "md5_crypt") + + def test_24_vary_rounds(self): + """test 'vary_rounds' hash option parsing""" + def parse(v): + return CryptContext(all__vary_rounds=v).to_dict()['all__vary_rounds'] + + # floats should be preserved + self.assertEqual(parse(0.1), 0.1) + self.assertEqual(parse('0.1'), 0.1) + + # 'xx%' should be converted to float + self.assertEqual(parse('10%'), 0.1) + + # ints should be preserved + self.assertEqual(parse(1000), 1000) + self.assertEqual(parse('1000'), 1000) + + #=================================================================== + # inspection & serialization + #=================================================================== + + def assertHandlerDerivedFrom(self, handler, base, msg=None): + self.assertTrue(handler_derived_from(handler, base), msg=msg) + + def test_30_schemes(self): + """test schemes() method""" + # NOTE: also checked under test_21 + + # test empty + ctx = CryptContext() + self.assertEqual(ctx.schemes(), ()) + self.assertEqual(ctx.schemes(resolve=True), ()) + + # test sample 1 + ctx = CryptContext(**self.sample_1_dict) + self.assertEqual(ctx.schemes(), tuple(self.sample_1_schemes)) + self.assertEqual(ctx.schemes(resolve=True, unconfigured=True), tuple(self.sample_1_handlers)) + for result, correct in zip(ctx.schemes(resolve=True), self.sample_1_handlers): + self.assertTrue(handler_derived_from(result, correct)) + + # test sample 2 + ctx = CryptContext(**self.sample_2_dict) + self.assertEqual(ctx.schemes(), ()) + + def test_31_default_scheme(self): + """test default_scheme() method""" + # NOTE: also checked under test_23 + + # test empty + ctx = CryptContext() + self.assertRaises(KeyError, ctx.default_scheme) + + # test sample 1 + ctx = CryptContext(**self.sample_1_dict) + self.assertEqual(ctx.default_scheme(), "md5_crypt") + self.assertEqual(ctx.default_scheme(resolve=True, unconfigured=True), hash.md5_crypt) + self.assertHandlerDerivedFrom(ctx.default_scheme(resolve=True), hash.md5_crypt) + + # test sample 2 + ctx = CryptContext(**self.sample_2_dict) + self.assertRaises(KeyError, ctx.default_scheme) + + # test defaults to first in scheme + ctx = CryptContext(schemes=self.sample_1_schemes) + self.assertEqual(ctx.default_scheme(), "des_crypt") + + # categories tested under test_23 + + def test_32_handler(self): + """test handler() method""" + + # default for empty + ctx = CryptContext() + self.assertRaises(KeyError, ctx.handler) + self.assertRaises(KeyError, ctx.handler, "md5_crypt") + + # default for sample 1 + ctx = CryptContext(**self.sample_1_dict) + self.assertEqual(ctx.handler(unconfigured=True), hash.md5_crypt) + self.assertHandlerDerivedFrom(ctx.handler(), hash.md5_crypt) + + # by name + self.assertEqual(ctx.handler("des_crypt", unconfigured=True), hash.des_crypt) + self.assertHandlerDerivedFrom(ctx.handler("des_crypt"), hash.des_crypt) + + # name not in schemes + self.assertRaises(KeyError, ctx.handler, "mysql323") + + # check handler() honors category default + ctx = CryptContext("sha256_crypt,md5_crypt", admin__context__default="md5_crypt") + self.assertEqual(ctx.handler(unconfigured=True), hash.sha256_crypt) + self.assertHandlerDerivedFrom(ctx.handler(), hash.sha256_crypt) + + self.assertEqual(ctx.handler(category="staff", unconfigured=True), hash.sha256_crypt) + self.assertHandlerDerivedFrom(ctx.handler(category="staff"), hash.sha256_crypt) + + self.assertEqual(ctx.handler(category="admin", unconfigured=True), hash.md5_crypt) + self.assertHandlerDerivedFrom(ctx.handler(category="staff"), hash.sha256_crypt) + + # test unicode category strings are accepted under py2 + if PY2: + self.assertEqual(ctx.handler(category=u("staff"), unconfigured=True), hash.sha256_crypt) + self.assertEqual(ctx.handler(category=u("admin"), unconfigured=True), hash.md5_crypt) + + def test_33_options(self): + """test internal _get_record_options() method""" + + def options(ctx, scheme, category=None): + return ctx._config._get_record_options_with_flag(scheme, category)[0] + + # this checks that (3 schemes, 3 categories) inherit options correctly. + # the 'user' category is not present in the options. + cc4 = CryptContext( + truncate_error=True, + schemes = [ "sha512_crypt", "des_crypt", "bsdi_crypt"], + deprecated = ["sha512_crypt", "des_crypt"], + all__vary_rounds = 0.1, + bsdi_crypt__vary_rounds=0.2, + sha512_crypt__max_rounds = 20000, + admin__context__deprecated = [ "des_crypt", "bsdi_crypt" ], + admin__all__vary_rounds = 0.05, + admin__bsdi_crypt__vary_rounds=0.3, + admin__sha512_crypt__max_rounds = 40000, + ) + self.assertEqual(cc4._config.categories, ("admin",)) + + # + # sha512_crypt + # NOTE: 'truncate_error' shouldn't be passed along... + # + self.assertEqual(options(cc4, "sha512_crypt"), dict( + deprecated=True, + vary_rounds=0.1, # inherited from all__ + max_rounds=20000, + )) + + self.assertEqual(options(cc4, "sha512_crypt", "user"), dict( + deprecated=True, # unconfigured category inherits from default + vary_rounds=0.1, + max_rounds=20000, + )) + + self.assertEqual(options(cc4, "sha512_crypt", "admin"), dict( + # NOT deprecated - context option overridden per-category + vary_rounds=0.05, # global overridden per-cateogry + max_rounds=40000, # overridden per-category + )) + + # + # des_crypt + # NOTE: vary_rounds shouldn't be passed along... + # + self.assertEqual(options(cc4, "des_crypt"), dict( + deprecated=True, + truncate_error=True, + )) + + self.assertEqual(options(cc4, "des_crypt", "user"), dict( + deprecated=True, # unconfigured category inherits from default + truncate_error=True, + )) + + self.assertEqual(options(cc4, "des_crypt", "admin"), dict( + deprecated=True, # unchanged though overidden + truncate_error=True, + )) + + # + # bsdi_crypt + # + self.assertEqual(options(cc4, "bsdi_crypt"), dict( + vary_rounds=0.2, # overridden from all__vary_rounds + )) + + self.assertEqual(options(cc4, "bsdi_crypt", "user"), dict( + vary_rounds=0.2, # unconfigured category inherits from default + )) + + self.assertEqual(options(cc4, "bsdi_crypt", "admin"), dict( + vary_rounds=0.3, + deprecated=True, # deprecation set per-category + )) + + def test_34_to_dict(self): + """test to_dict() method""" + # NOTE: this is tested all throughout this test case. + ctx = CryptContext(**self.sample_1_dict) + self.assertEqual(ctx.to_dict(), self.sample_1_dict) + self.assertEqual(ctx.to_dict(resolve=True), self.sample_1_resolved_dict) + + def test_35_to_string(self): + """test to_string() method""" + + # create ctx and serialize + ctx = CryptContext(**self.sample_1_dict) + dump = ctx.to_string() + + # check ctx->string returns canonical format. + # NOTE: ConfigParser for PY26 doesn't use OrderedDict, + # making to_string()'s ordering unpredictable... + # so we skip this test under PY26. + if not PY26: + self.assertEqual(dump, self.sample_1_unicode) + + # check ctx->string->ctx->dict returns original + ctx2 = CryptContext.from_string(dump) + self.assertEqual(ctx2.to_dict(), self.sample_1_dict) + + # test section kwd is honored + other = ctx.to_string(section="password-security") + self.assertEqual(other, dump.replace("[passlib]","[password-security]")) + + # test unmanaged handler warning + from passlib.tests.test_utils_handlers import UnsaltedHash + ctx3 = CryptContext([UnsaltedHash, "md5_crypt"]) + dump = ctx3.to_string() + self.assertRegex(dump, r"# NOTE: the 'unsalted_test_hash' handler\(s\)" + r" are not registered with Passlib") + + #=================================================================== + # password hash api + #=================================================================== + nonstring_vectors = [ + (None, {}), + (None, {"scheme": "des_crypt"}), + (1, {}), + ((), {}), + ] + + def test_40_basic(self): + """test basic hash/identify/verify functionality""" + handlers = [hash.md5_crypt, hash.des_crypt, hash.bsdi_crypt] + cc = CryptContext(handlers, bsdi_crypt__default_rounds=5) + + # run through handlers + for crypt in handlers: + h = cc.hash("test", scheme=crypt.name) + self.assertEqual(cc.identify(h), crypt.name) + self.assertEqual(cc.identify(h, resolve=True, unconfigured=True), crypt) + self.assertHandlerDerivedFrom(cc.identify(h, resolve=True), crypt) + self.assertTrue(cc.verify('test', h)) + self.assertFalse(cc.verify('notest', h)) + + # test default + h = cc.hash("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_41_genconfig(self): + """test genconfig() method""" + cc = CryptContext(schemes=["md5_crypt", "phpass"], + phpass__ident="H", + phpass__default_rounds=7, + admin__phpass__ident="P", + ) + + # uses default scheme + self.assertTrue(cc.genconfig().startswith("$1$")) + + # override scheme + self.assertTrue(cc.genconfig(scheme="phpass").startswith("$H$5")) + + # category override + self.assertTrue(cc.genconfig(scheme="phpass", category="admin").startswith("$P$5")) + self.assertTrue(cc.genconfig(scheme="phpass", category="staff").startswith("$H$5")) + + # override scheme & custom settings + self.assertEqual( + cc.genconfig(scheme="phpass", salt='.'*8, rounds=8, ident='P'), + '$P$6........22zGEuacuPOqEpYPDeR0R/', # NOTE: config string generated w/ rounds=1 + ) + + #-------------------------------------------------------------- + # border cases + #-------------------------------------------------------------- + + # test unicode category strings are accepted under py2 + # this tests basic _get_record() used by hash/genhash/verify. + # we have to omit scheme=xxx so codepath is tested fully + if PY2: + c2 = cc.copy(default="phpass") + self.assertTrue(c2.genconfig(category=u("admin")).startswith("$P$5")) + self.assertTrue(c2.genconfig(category=u("staff")).startswith("$H$5")) + + # throws error without schemes + self.assertRaises(KeyError, CryptContext().genconfig) + self.assertRaises(KeyError, CryptContext().genconfig, scheme='md5_crypt') + + # bad scheme values + self.assertRaises(KeyError, cc.genconfig, scheme="fake") # XXX: should this be ValueError? + self.assertRaises(TypeError, cc.genconfig, scheme=1, category='staff') + self.assertRaises(TypeError, cc.genconfig, scheme=1) + + # bad category values + self.assertRaises(TypeError, cc.genconfig, category=1) + + + def test_42_genhash(self): + """test genhash() method""" + + #-------------------------------------------------------------- + # border cases + #-------------------------------------------------------------- + + # rejects non-string secrets + cc = CryptContext(["des_crypt"]) + hash = cc.hash('stub') + for secret, kwds in self.nonstring_vectors: + self.assertRaises(TypeError, cc.genhash, secret, hash, **kwds) + + # rejects non-string config strings + cc = CryptContext(["des_crypt"]) + for config, kwds in self.nonstring_vectors: + if hash is None: + # NOTE: as of 1.7, genhash is just wrapper for hash(), + # and handles genhash(secret, None) fine. + continue + self.assertRaises(TypeError, cc.genhash, 'secret', config, **kwds) + + # rejects config=None, even if default scheme lacks config string + cc = CryptContext(["mysql323"]) + self.assertRaises(TypeError, cc.genhash, "stub", None) + + # throws error without schemes + self.assertRaises(KeyError, CryptContext().genhash, 'secret', 'hash') + + # bad scheme values + self.assertRaises(KeyError, cc.genhash, 'secret', hash, scheme="fake") # XXX: should this be ValueError? + self.assertRaises(TypeError, cc.genhash, 'secret', hash, scheme=1) + + # bad category values + self.assertRaises(TypeError, cc.genconfig, 'secret', hash, category=1) + + def test_43_hash(self,): + """test hash() method""" + # XXX: what more can we test here that isn't deprecated + # or handled under another test (e.g. context kwds?) + + # respects rounds + cc = CryptContext(**self.sample_4_dict) + hash = cc.hash("password") + self.assertTrue(hash.startswith("$5$rounds=3000$")) + self.assertTrue(cc.verify("password", hash)) + self.assertFalse(cc.verify("passwordx", hash)) + + # make default > max throws error if attempted + # XXX: move this to copy() test? + self.assertRaises(ValueError, cc.copy, + sha256_crypt__default_rounds=4000) + + # rejects non-string secrets + cc = CryptContext(["des_crypt"]) + for secret, kwds in self.nonstring_vectors: + self.assertRaises(TypeError, cc.hash, secret, **kwds) + + # throws error without schemes + self.assertRaises(KeyError, CryptContext().hash, 'secret') + + # bad category values + self.assertRaises(TypeError, cc.hash, 'secret', category=1) + + def test_43_hash_legacy(self, use_16_legacy=False): + """test hash() method -- legacy 'scheme' and settings keywords""" + cc = CryptContext(**self.sample_4_dict) + + # TODO: should migrate these tests elsewhere, or remove them. + # can be replaced with following equivalent: + # + # def wrapper(secret, scheme=None, category=None, **kwds): + # handler = cc.handler(scheme, category) + # if kwds: + # handler = handler.using(**kwds) + # return handler.hash(secret) + # + # need to make sure bits being tested here are tested + # under the tests for the equivalent methods called above, + # and then discard the rest of these under 2.0. + + # hash specific settings + with self.assertWarningList(["passing settings to.*is deprecated"]): + self.assertEqual( + cc.hash("password", scheme="phpass", salt='.'*8), + '$H$5........De04R5Egz0aq8Tf.1eVhY/', + ) + with self.assertWarningList(["passing settings to.*is deprecated"]): + self.assertEqual( + cc.hash("password", scheme="phpass", salt='.'*8, ident="P"), + '$P$5........De04R5Egz0aq8Tf.1eVhY/', + ) + + # NOTE: more thorough job of rounds limits done below. + + # min rounds + with self.assertWarningList(["passing settings to.*is deprecated"]): + self.assertEqual( + cc.hash("password", rounds=1999, salt="nacl"), + '$5$rounds=1999$nacl$nmfwJIxqj0csloAAvSER0B8LU0ERCAbhmMug4Twl609', + ) + + with self.assertWarningList(["passing settings to.*is deprecated"]): + self.assertEqual( + cc.hash("password", rounds=2001, salt="nacl"), + '$5$rounds=2001$nacl$8PdeoPL4aXQnJ0woHhqgIw/efyfCKC2WHneOpnvF.31' + ) + # NOTE: max rounds, etc tested in genconfig() + + # bad scheme values + self.assertRaises(KeyError, cc.hash, 'secret', scheme="fake") # XXX: should this be ValueError? + self.assertRaises(TypeError, cc.hash, 'secret', scheme=1) + + def test_44_identify(self): + """test identify() border cases""" + handlers = ["md5_crypt", "des_crypt", "bsdi_crypt"] + cc = CryptContext(handlers, bsdi_crypt__default_rounds=5) + + # check unknown hash + self.assertEqual(cc.identify('$9$232323123$1287319827'), None) + self.assertRaises(ValueError, cc.identify, '$9$232323123$1287319827', required=True) + + #-------------------------------------------------------------- + # border cases + #-------------------------------------------------------------- + + # rejects non-string hashes + cc = CryptContext(["des_crypt"]) + for hash, kwds in self.nonstring_vectors: + self.assertRaises(TypeError, cc.identify, hash, **kwds) + + # throws error without schemes + cc = CryptContext() + self.assertIs(cc.identify('hash'), None) + self.assertRaises(KeyError, cc.identify, 'hash', required=True) + + # bad category values + self.assertRaises(TypeError, cc.identify, None, category=1) + + def test_45_verify(self): + """test verify() scheme kwd""" + handlers = ["md5_crypt", "des_crypt", "bsdi_crypt"] + cc = CryptContext(handlers, bsdi_crypt__default_rounds=5) + + h = hash.md5_crypt.hash("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') + + #-------------------------------------------------------------- + # border cases + #-------------------------------------------------------------- + + # unknown hash should throw error + self.assertRaises(ValueError, cc.verify, 'stub', '$6$232323123$1287319827') + + # rejects non-string secrets + cc = CryptContext(["des_crypt"]) + h = refhash = cc.hash('stub') + for secret, kwds in self.nonstring_vectors: + self.assertRaises(TypeError, cc.verify, secret, h, **kwds) + + # always treat hash=None as False + self.assertFalse(cc.verify(secret, None)) + + # rejects non-string hashes + cc = CryptContext(["des_crypt"]) + for h, kwds in self.nonstring_vectors: + if h is None: + continue + self.assertRaises(TypeError, cc.verify, 'secret', h, **kwds) + + # throws error without schemes + self.assertRaises(KeyError, CryptContext().verify, 'secret', 'hash') + + # bad scheme values + self.assertRaises(KeyError, cc.verify, 'secret', refhash, scheme="fake") # XXX: should this be ValueError? + self.assertRaises(TypeError, cc.verify, 'secret', refhash, scheme=1) + + # bad category values + self.assertRaises(TypeError, cc.verify, 'secret', refhash, category=1) + + def test_46_needs_update(self): + """test needs_update() method""" + cc = CryptContext(**self.sample_4_dict) + + # check deprecated scheme + self.assertTrue(cc.needs_update('9XXD4trGYeGJA')) + self.assertFalse(cc.needs_update('$1$J8HC2RCr$HcmM.7NxB2weSvlw2FgzU0')) + + # check min rounds + self.assertTrue(cc.needs_update('$5$rounds=1999$jD81UCoo.zI.UETs$Y7qSTQ6mTiU9qZB4fRr43wRgQq4V.5AAf7F97Pzxey/')) + self.assertFalse(cc.needs_update('$5$rounds=2000$228SSRje04cnNCaQ$YGV4RYu.5sNiBvorQDlO0WWQjyJVGKBcJXz3OtyQ2u8')) + + # check max rounds + self.assertFalse(cc.needs_update('$5$rounds=3000$fS9iazEwTKi7QPW4$VasgBC8FqlOvD7x2HhABaMXCTh9jwHclPA9j5YQdns.')) + self.assertTrue(cc.needs_update('$5$rounds=3001$QlFHHifXvpFX4PLs$/0ekt7lSs/lOikSerQ0M/1porEHxYq7W/2hdFpxA3fA')) + + #-------------------------------------------------------------- + # test hash.needs_update() interface + #-------------------------------------------------------------- + check_state = [] + class dummy(uh.StaticHandler): + name = 'dummy' + _hash_prefix = '@' + + @classmethod + def needs_update(cls, hash, secret=None): + check_state.append((hash, secret)) + return secret == "nu" + + def _calc_checksum(self, secret): + from hashlib import md5 + if isinstance(secret, unicode): + secret = secret.encode("utf-8") + return str_to_uascii(md5(secret).hexdigest()) + + # calling needs_update should query callback + ctx = CryptContext([dummy]) + hash = refhash = dummy.hash("test") + self.assertFalse(ctx.needs_update(hash)) + self.assertEqual(check_state, [(hash,None)]) + del check_state[:] + + # now with a password + self.assertFalse(ctx.needs_update(hash, secret='bob')) + self.assertEqual(check_state, [(hash,'bob')]) + del check_state[:] + + # now when it returns True + self.assertTrue(ctx.needs_update(hash, secret='nu')) + self.assertEqual(check_state, [(hash,'nu')]) + del check_state[:] + + #-------------------------------------------------------------- + # border cases + #-------------------------------------------------------------- + + # rejects non-string hashes + cc = CryptContext(["des_crypt"]) + for hash, kwds in self.nonstring_vectors: + self.assertRaises(TypeError, cc.needs_update, hash, **kwds) + + # throws error without schemes + self.assertRaises(KeyError, CryptContext().needs_update, 'hash') + + # bad scheme values + self.assertRaises(KeyError, cc.needs_update, refhash, scheme="fake") # XXX: should this be ValueError? + self.assertRaises(TypeError, cc.needs_update, refhash, scheme=1) + + # bad category values + self.assertRaises(TypeError, cc.needs_update, refhash, category=1) + + def test_47_verify_and_update(self): + """test verify_and_update()""" + cc = CryptContext(**self.sample_4_dict) + + # create some hashes + h1 = cc.handler("des_crypt").hash("password") + h2 = cc.handler("sha256_crypt").hash("password") + + # 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 + #-------------------------------------------------------------- + + # rejects non-string secrets + cc = CryptContext(["des_crypt"]) + hash = refhash = cc.hash('stub') + for secret, kwds in self.nonstring_vectors: + self.assertRaises(TypeError, cc.verify_and_update, secret, hash, **kwds) + + # always treat hash=None as False + self.assertEqual(cc.verify_and_update(secret, None), (False, None)) + + # rejects non-string hashes + cc = CryptContext(["des_crypt"]) + for hash, kwds in self.nonstring_vectors: + if hash is None: + continue + self.assertRaises(TypeError, cc.verify_and_update, 'secret', hash, **kwds) + + # throws error without schemes + self.assertRaises(KeyError, CryptContext().verify_and_update, 'secret', 'hash') + + # bad scheme values + self.assertRaises(KeyError, cc.verify_and_update, 'secret', refhash, scheme="fake") # XXX: should this be ValueError? + self.assertRaises(TypeError, cc.verify_and_update, 'secret', refhash, scheme=1) + + # bad category values + self.assertRaises(TypeError, cc.verify_and_update, 'secret', refhash, category=1) + + def test_48_context_kwds(self): + """hash(), verify(), and verify_and_update() -- discard unused context keywords""" + + # setup test case + # NOTE: postgres_md5 hash supports 'user' context kwd, which is used for this test. + from passlib.hash import des_crypt, md5_crypt, postgres_md5 + des_hash = des_crypt.hash("stub") + pg_root_hash = postgres_md5.hash("stub", user="root") + pg_admin_hash = postgres_md5.hash("stub", user="admin") + + #------------------------------------------------------------ + # case 1: contextual kwds not supported by any hash in CryptContext + #------------------------------------------------------------ + cc1 = CryptContext([des_crypt, md5_crypt]) + self.assertEqual(cc1.context_kwds, set()) + + # des_scrypt should work w/o any contextual kwds + self.assertTrue(des_crypt.identify(cc1.hash("stub")), "des_crypt") + self.assertTrue(cc1.verify("stub", des_hash)) + self.assertEqual(cc1.verify_and_update("stub", des_hash), (True, None)) + + # des_crypt should throw error due to unknown context keyword + with self.assertWarningList(["passing settings to.*is deprecated"]): + self.assertRaises(TypeError, cc1.hash, "stub", user="root") + self.assertRaises(TypeError, cc1.verify, "stub", des_hash, user="root") + self.assertRaises(TypeError, cc1.verify_and_update, "stub", des_hash, user="root") + + #------------------------------------------------------------ + # case 2: at least one contextual kwd supported by non-default hash + #------------------------------------------------------------ + cc2 = CryptContext([des_crypt, postgres_md5]) + self.assertEqual(cc2.context_kwds, set(["user"])) + + # verify des_crypt works w/o "user" kwd + self.assertTrue(des_crypt.identify(cc2.hash("stub")), "des_crypt") + self.assertTrue(cc2.verify("stub", des_hash)) + self.assertEqual(cc2.verify_and_update("stub", des_hash), (True, None)) + + # verify des_crypt ignores "user" kwd + self.assertTrue(des_crypt.identify(cc2.hash("stub", user="root")), "des_crypt") + self.assertTrue(cc2.verify("stub", des_hash, user="root")) + self.assertEqual(cc2.verify_and_update("stub", des_hash, user="root"), (True, None)) + + # verify error with unknown kwd + with self.assertWarningList(["passing settings to.*is deprecated"]): + self.assertRaises(TypeError, cc2.hash, "stub", badkwd="root") + self.assertRaises(TypeError, cc2.verify, "stub", des_hash, badkwd="root") + self.assertRaises(TypeError, cc2.verify_and_update, "stub", des_hash, badkwd="root") + + #------------------------------------------------------------ + # case 3: at least one contextual kwd supported by default hash + #------------------------------------------------------------ + cc3 = CryptContext([postgres_md5, des_crypt], deprecated="auto") + self.assertEqual(cc3.context_kwds, set(["user"])) + + # postgres_md5 should have error w/o context kwd + self.assertRaises(TypeError, cc3.hash, "stub") + self.assertRaises(TypeError, cc3.verify, "stub", pg_root_hash) + self.assertRaises(TypeError, cc3.verify_and_update, "stub", pg_root_hash) + + # postgres_md5 should work w/ context kwd + self.assertEqual(cc3.hash("stub", user="root"), pg_root_hash) + self.assertTrue(cc3.verify("stub", pg_root_hash, user="root")) + self.assertEqual(cc3.verify_and_update("stub", pg_root_hash, user="root"), (True, None)) + + # verify_and_update() should fail against wrong user + self.assertEqual(cc3.verify_and_update("stub", pg_root_hash, user="admin"), (False, None)) + + # verify_and_update() should pass all context kwds through when rehashing + self.assertEqual(cc3.verify_and_update("stub", des_hash, user="root"), + (True, pg_root_hash)) + + #=================================================================== + # rounds options + #=================================================================== + + # TODO: now that rounds generation has moved out of _CryptRecord to HasRounds, + # this should just test that we're passing right options to handler.using(), + # and that resulting handler has right settings. + # Can then just let HasRounds tests (which are a copy of this) deal with things. + + # NOTE: the follow tests check how _CryptRecord handles + # the min/max/default/vary_rounds options, via the output of + # genconfig(). it's assumed hash() takes the same codepath. + + def test_50_rounds_limits(self): + """test rounds limits""" + cc = CryptContext(schemes=["sha256_crypt"], + sha256_crypt__min_rounds=2000, + sha256_crypt__max_rounds=3000, + sha256_crypt__default_rounds=2500, + ) + + # stub digest returned by sha256_crypt's genconfig calls.. + STUB = '...........................................' + + #-------------------------------------------------- + # settings should have been applied to custom handler, + # it should take care of the rest + #-------------------------------------------------- + custom_handler = cc._get_record("sha256_crypt", None) + self.assertEqual(custom_handler.min_desired_rounds, 2000) + self.assertEqual(custom_handler.max_desired_rounds, 3000) + self.assertEqual(custom_handler.default_rounds, 2500) + + #-------------------------------------------------- + # min_rounds + #-------------------------------------------------- + + # set below handler minimum + with self.assertWarningList([PasslibHashWarning]*2): + c2 = cc.copy(sha256_crypt__min_rounds=500, sha256_crypt__max_rounds=None, + sha256_crypt__default_rounds=500) + self.assertEqual(c2.genconfig(salt="nacl"), "$5$rounds=1000$nacl$" + STUB) + + # below policy minimum + # NOTE: formerly issued a warning in passlib 1.6, now just a wrapper for .replace() + with self.assertWarningList([]): + self.assertEqual( + cc.genconfig(rounds=1999, salt="nacl"), '$5$rounds=1999$nacl$' + STUB) + + # equal to policy minimum + self.assertEqual( + cc.genconfig(rounds=2000, salt="nacl"), '$5$rounds=2000$nacl$' + STUB) + + # above policy minimum + self.assertEqual( + cc.genconfig(rounds=2001, salt="nacl"), '$5$rounds=2001$nacl$' + STUB) + + #-------------------------------------------------- + # max rounds + #-------------------------------------------------- + + # set above handler max + with self.assertWarningList([PasslibHashWarning]*2): + c2 = cc.copy(sha256_crypt__max_rounds=int(1e9)+500, sha256_crypt__min_rounds=None, + sha256_crypt__default_rounds=int(1e9)+500) + + self.assertEqual(c2.genconfig(salt="nacl"), "$5$rounds=999999999$nacl$" + STUB) + + # above policy max + # NOTE: formerly issued a warning in passlib 1.6, now just a wrapper for .using() + with self.assertWarningList([]): + self.assertEqual( + cc.genconfig(rounds=3001, salt="nacl"), '$5$rounds=3001$nacl$' + STUB) + + # equal policy max + self.assertEqual( + cc.genconfig(rounds=3000, salt="nacl"), '$5$rounds=3000$nacl$' + STUB) + + # below policy max + self.assertEqual( + cc.genconfig(rounds=2999, salt="nacl"), '$5$rounds=2999$nacl$' + STUB) + + #-------------------------------------------------- + # default_rounds + #-------------------------------------------------- + + # explicit default rounds + self.assertEqual(cc.genconfig(salt="nacl"), '$5$rounds=2500$nacl$' + STUB) + + # fallback default rounds - use handler's + df = hash.sha256_crypt.default_rounds + c2 = cc.copy(sha256_crypt__default_rounds=None, sha256_crypt__max_rounds=df<<1) + self.assertEqual(c2.genconfig(salt="nacl"), '$5$rounds=%d$nacl$%s' % (df, STUB)) + + # fallback default rounds - use handler's, but clipped to max rounds + c2 = cc.copy(sha256_crypt__default_rounds=None, sha256_crypt__max_rounds=3000) + self.assertEqual(c2.genconfig(salt="nacl"), '$5$rounds=3000$nacl$' + STUB) + + # TODO: test default falls back to mx / mn if handler has no default. + + # default rounds - out of bounds + self.assertRaises(ValueError, cc.copy, sha256_crypt__default_rounds=1999) + cc.copy(sha256_crypt__default_rounds=2000) + cc.copy(sha256_crypt__default_rounds=3000) + self.assertRaises(ValueError, cc.copy, sha256_crypt__default_rounds=3001) + + #-------------------------------------------------- + # border cases + #-------------------------------------------------- + + # invalid min/max bounds + c2 = CryptContext(schemes=["sha256_crypt"]) + # NOTE: as of v1.7, these are clipped w/ a warning instead... + # self.assertRaises(ValueError, c2.copy, sha256_crypt__min_rounds=-1) + # self.assertRaises(ValueError, c2.copy, sha256_crypt__max_rounds=-1) + self.assertRaises(ValueError, c2.copy, sha256_crypt__min_rounds=2000, + sha256_crypt__max_rounds=1999) + + # test bad values + self.assertRaises(ValueError, CryptContext, sha256_crypt__min_rounds='x') + self.assertRaises(ValueError, CryptContext, sha256_crypt__max_rounds='x') + self.assertRaises(ValueError, CryptContext, all__vary_rounds='x') + self.assertRaises(ValueError, CryptContext, sha256_crypt__default_rounds='x') + + # test bad types rejected + bad = datetime.datetime.now() # picked cause can't be compared to int + self.assertRaises(TypeError, CryptContext, "sha256_crypt", sha256_crypt__min_rounds=bad) + self.assertRaises(TypeError, CryptContext, "sha256_crypt", sha256_crypt__max_rounds=bad) + self.assertRaises(TypeError, CryptContext, "sha256_crypt", all__vary_rounds=bad) + self.assertRaises(TypeError, CryptContext, "sha256_crypt", sha256_crypt__default_rounds=bad) + + def test_51_linear_vary_rounds(self): + """test linear vary rounds""" + cc = CryptContext(schemes=["sha256_crypt"], + sha256_crypt__min_rounds=1995, + sha256_crypt__max_rounds=2005, + sha256_crypt__default_rounds=2000, + ) + + # test negative + self.assertRaises(ValueError, cc.copy, all__vary_rounds=-1) + self.assertRaises(ValueError, cc.copy, all__vary_rounds="-1%") + self.assertRaises(ValueError, cc.copy, all__vary_rounds="101%") + + # test static + c2 = cc.copy(all__vary_rounds=0) + self.assertEqual(c2._get_record("sha256_crypt", None).vary_rounds, 0) + self.assert_rounds_range(c2, "sha256_crypt", 2000, 2000) + + c2 = cc.copy(all__vary_rounds="0%") + self.assertEqual(c2._get_record("sha256_crypt", None).vary_rounds, 0) + self.assert_rounds_range(c2, "sha256_crypt", 2000, 2000) + + # test absolute + c2 = cc.copy(all__vary_rounds=1) + self.assertEqual(c2._get_record("sha256_crypt", None).vary_rounds, 1) + self.assert_rounds_range(c2, "sha256_crypt", 1999, 2001) + c2 = cc.copy(all__vary_rounds=100) + self.assertEqual(c2._get_record("sha256_crypt", None).vary_rounds, 100) + self.assert_rounds_range(c2, "sha256_crypt", 1995, 2005) + + # test relative + c2 = cc.copy(all__vary_rounds="0.1%") + self.assertEqual(c2._get_record("sha256_crypt", None).vary_rounds, 0.001) + self.assert_rounds_range(c2, "sha256_crypt", 1998, 2002) + c2 = cc.copy(all__vary_rounds="100%") + self.assertEqual(c2._get_record("sha256_crypt", None).vary_rounds, 1.0) + self.assert_rounds_range(c2, "sha256_crypt", 1995, 2005) + + def test_52_log2_vary_rounds(self): + """test log2 vary rounds""" + cc = CryptContext(schemes=["bcrypt"], + bcrypt__min_rounds=15, + bcrypt__max_rounds=25, + bcrypt__default_rounds=20, + ) + + # test negative + self.assertRaises(ValueError, cc.copy, all__vary_rounds=-1) + self.assertRaises(ValueError, cc.copy, all__vary_rounds="-1%") + self.assertRaises(ValueError, cc.copy, all__vary_rounds="101%") + + # test static + c2 = cc.copy(all__vary_rounds=0) + self.assertEqual(c2._get_record("bcrypt", None).vary_rounds, 0) + self.assert_rounds_range(c2, "bcrypt", 20, 20) + + c2 = cc.copy(all__vary_rounds="0%") + self.assertEqual(c2._get_record("bcrypt", None).vary_rounds, 0) + self.assert_rounds_range(c2, "bcrypt", 20, 20) + + # test absolute + c2 = cc.copy(all__vary_rounds=1) + self.assertEqual(c2._get_record("bcrypt", None).vary_rounds, 1) + self.assert_rounds_range(c2, "bcrypt", 19, 21) + c2 = cc.copy(all__vary_rounds=100) + self.assertEqual(c2._get_record("bcrypt", None).vary_rounds, 100) + self.assert_rounds_range(c2, "bcrypt", 15, 25) + + # test relative - should shift over at 50% mark + c2 = cc.copy(all__vary_rounds="1%") + self.assertEqual(c2._get_record("bcrypt", None).vary_rounds, 0.01) + self.assert_rounds_range(c2, "bcrypt", 20, 20) + + c2 = cc.copy(all__vary_rounds="49%") + self.assertEqual(c2._get_record("bcrypt", None).vary_rounds, 0.49) + self.assert_rounds_range(c2, "bcrypt", 20, 20) + + c2 = cc.copy(all__vary_rounds="50%") + self.assertEqual(c2._get_record("bcrypt", None).vary_rounds, 0.5) + self.assert_rounds_range(c2, "bcrypt", 19, 20) + + c2 = cc.copy(all__vary_rounds="100%") + self.assertEqual(c2._get_record("bcrypt", None).vary_rounds, 1.0) + 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.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:") + + #=================================================================== + # harden_verify / min_verify_time + #=================================================================== + def test_harden_verify_parsing(self): + """harden_verify -- parsing""" + warnings.filterwarnings("ignore", ".*harden_verify.*", + category=DeprecationWarning) + + # valid values + ctx = CryptContext(schemes=["sha256_crypt"]) + self.assertEqual(ctx.harden_verify, None) + self.assertEqual(ctx.using(harden_verify="").harden_verify, None) + self.assertEqual(ctx.using(harden_verify="true").harden_verify, None) + self.assertEqual(ctx.using(harden_verify="false").harden_verify, None) + + def test_dummy_verify(self): + """ + dummy_verify() method + """ + # check dummy_verify() takes expected time + expected = 0.05 + accuracy = 0.2 + handler = DelayHash.using() + handler.delay = expected + ctx = CryptContext(schemes=[handler]) + ctx.dummy_verify() # prime the memoized helpers + elapsed, _ = time_call(ctx.dummy_verify) + self.assertAlmostEqual(elapsed, expected, delta=expected * accuracy) + + # TODO: test dummy_verify() invoked by .verify() when hash is None, + # and same for .verify_and_update() + + #=================================================================== + # feature tests + #=================================================================== + def test_61_autodeprecate(self): + """test deprecated='auto' is handled correctly""" + + def getstate(ctx, category=None): + return [ctx.handler(scheme, category).deprecated for scheme in ctx.schemes()] + + # correctly reports default + ctx = CryptContext("sha256_crypt,md5_crypt,des_crypt", deprecated="auto") + self.assertEqual(getstate(ctx, None), [False, True, True]) + self.assertEqual(getstate(ctx, "admin"), [False, True, True]) + + # correctly reports changed default + ctx.update(default="md5_crypt") + self.assertEqual(getstate(ctx, None), [True, False, True]) + self.assertEqual(getstate(ctx, "admin"), [True, False, True]) + + # category default is handled correctly + ctx.update(admin__context__default="des_crypt") + self.assertEqual(getstate(ctx, None), [True, False, True]) + self.assertEqual(getstate(ctx, "admin"), [True, True, False]) + + # handles 1 scheme + ctx = CryptContext(["sha256_crypt"], deprecated="auto") + self.assertEqual(getstate(ctx, None), [False]) + self.assertEqual(getstate(ctx, "admin"), [False]) + + # disallow auto & other deprecated schemes at same time. + self.assertRaises(ValueError, CryptContext, "sha256_crypt,md5_crypt", + deprecated="auto,md5_crypt") + self.assertRaises(ValueError, CryptContext, "sha256_crypt,md5_crypt", + deprecated="md5_crypt,auto") + + def test_disabled_hashes(self): + """disabled hash support""" + # + # init ref info + # + from passlib.exc import UnknownHashError + from passlib.hash import md5_crypt, unix_disabled + + ctx = CryptContext(["des_crypt"]) + ctx2 = CryptContext(["des_crypt", "unix_disabled"]) + h_ref = ctx.hash("foo") + h_other = md5_crypt.hash('foo') + + # + # ctx.disable() + # + + # test w/o disabled hash support + self.assertRaisesRegex(RuntimeError, "no disabled hasher present", + ctx.disable) + self.assertRaisesRegex(RuntimeError, "no disabled hasher present", + ctx.disable, h_ref) + self.assertRaisesRegex(RuntimeError, "no disabled hasher present", + ctx.disable, h_other) + + # test w/ disabled hash support + h_dis = ctx2.disable() + self.assertEqual(h_dis, unix_disabled.default_marker) + h_dis_ref = ctx2.disable(h_ref) + self.assertEqual(h_dis_ref, unix_disabled.default_marker + h_ref) + + h_dis_other = ctx2.disable(h_other) + self.assertEqual(h_dis_other, unix_disabled.default_marker + h_other) + + # don't double-wrap existing disabled hash + self.assertEqual(ctx2.disable(h_dis_ref), h_dis_ref) + + # + # ctx.is_enabled() + # + + # test w/o disabled hash support + self.assertTrue(ctx.is_enabled(h_ref)) + self.assertRaises(UnknownHashError, ctx.is_enabled, h_other) + self.assertRaises(UnknownHashError, ctx.is_enabled, h_dis) + self.assertRaises(UnknownHashError, ctx.is_enabled, h_dis_ref) + + # test w/ disabled hash support + self.assertTrue(ctx2.is_enabled(h_ref)) + self.assertRaises(UnknownHashError, ctx.is_enabled, h_other) + self.assertFalse(ctx2.is_enabled(h_dis)) + self.assertFalse(ctx2.is_enabled(h_dis_ref)) + + # + # ctx.enable() + # + + # test w/o disabled hash support + self.assertRaises(UnknownHashError, ctx.enable, "") + self.assertRaises(TypeError, ctx.enable, None) + self.assertEqual(ctx.enable(h_ref), h_ref) + self.assertRaises(UnknownHashError, ctx.enable, h_other) + self.assertRaises(UnknownHashError, ctx.enable, h_dis) + self.assertRaises(UnknownHashError, ctx.enable, h_dis_ref) + + # test w/ disabled hash support + self.assertRaises(UnknownHashError, ctx.enable, "") + self.assertRaises(TypeError, ctx2.enable, None) + self.assertEqual(ctx2.enable(h_ref), h_ref) + self.assertRaises(UnknownHashError, ctx2.enable, h_other) + self.assertRaisesRegex(ValueError, "cannot restore original hash", + ctx2.enable, h_dis) + self.assertEqual(ctx2.enable(h_dis_ref), h_ref) + + #=================================================================== + # eoc + #=================================================================== + +import hashlib, time + +class DelayHash(uh.StaticHandler): + """dummy hasher which delays by specified amount""" + name = "delay_hash" + checksum_chars = uh.LOWER_HEX_CHARS + checksum_size = 40 + delay = 0 + _hash_prefix = u("$x$") + + def _calc_checksum(self, secret): + time.sleep(self.delay) + if isinstance(secret, unicode): + secret = secret.encode("utf-8") + return str_to_uascii(hashlib.sha1(b"prefix" + secret).hexdigest()) + +#============================================================================= +# 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") + + 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.assertEqual(cc.schemes(), ("dummy_2", "des_crypt")) + self.assertTrue(cc.handler("des_crypt").deprecated) + + self.assertTrue(has_crypt_handler("dummy_2", True)) + + def test_callable_constructor(self): + self.assertFalse(has_crypt_handler("dummy_2")) + register_crypt_handler_path("dummy_2", "passlib.tests.test_context") + + def onload(flag=False): + self.assertTrue(flag) + return dict(schemes=iter(["dummy_2", "des_crypt"]), deprecated=["des_crypt"]) + + cc = LazyCryptContext(onload=onload, flag=True) + + self.assertFalse(has_crypt_handler("dummy_2", True)) + + self.assertEqual(cc.schemes(), ("dummy_2", "des_crypt")) + self.assertTrue(cc.handler("des_crypt").deprecated) + + self.assertTrue(has_crypt_handler("dummy_2", True)) + +#============================================================================= +# eof +#============================================================================= diff --git a/venv/Lib/site-packages/passlib/tests/test_context_deprecated.py b/venv/Lib/site-packages/passlib/tests/test_context_deprecated.py new file mode 100644 index 0000000..0f76624 --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_context_deprecated.py @@ -0,0 +1,743 @@ +"""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 +from logging import getLogger +import os +import warnings +# 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.utils import to_bytes, to_unicode +import passlib.utils.handlers as uh +from passlib.tests.utils import TestCase, set_file +from passlib.registry import (register_crypt_handler_path, + _has_crypt_handler as 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") + warnings.filterwarnings("ignore", + r"the method.*hash_needs_update.*is deprecated") + warnings.filterwarnings("ignore", "The 'all' scheme is deprecated.*") + warnings.filterwarnings("ignore", "bsdi_crypt rounds should be odd") + + def test_00_constructor(self): + """test CryptPolicy() constructor""" + policy = CryptPolicy(**self.sample_config_1pd) + self.assertEqual(policy.to_dict(), self.sample_config_1pd) + + policy = CryptPolicy(self.sample_config_1pd) + self.assertEqual(policy.to_dict(), self.sample_config_1pd) + + self.assertRaises(TypeError, CryptPolicy, {}, {}) + self.assertRaises(TypeError, CryptPolicy, {}, dummy=1) + + # 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""" + path = self.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(), 0) + self.assertEqual(pb.get_min_verify_time('admin'), 0) + + #=================================================================== + # 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) + + s = pa.to_string(encoding="latin-1") + self.assertIsInstance(s, bytes) + + #=================================================================== + # + #=================================================================== + +#============================================================================= +# 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.*") + warnings.filterwarnings("ignore", + r"the method.*hash_needs_update.*is 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) + + # policy kwd + policy = cc.policy + cc = CryptContext(policy=policy) + self.assertEqual(cc.to_dict(), policy.to_dict()) + + cc = CryptContext(policy=policy, default="bsdi_crypt") + self.assertNotEqual(cc.to_dict(), policy.to_dict()) + self.assertEqual(cc.to_dict(), dict(schemes=["md5_crypt","bsdi_crypt","des_crypt"], + default="bsdi_crypt")) + + self.assertRaises(TypeError, setattr, cc, 'policy', None) + self.assertRaises(TypeError, CryptContext, policy='x') + + 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.hash, '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.hash, '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_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')) + + #=================================================================== + # border cases + #=================================================================== + def test_30_nonstring_hash(self): + """test non-string hash values cause error""" + warnings.filterwarnings("ignore", ".*needs_update.*'scheme' keyword is deprecated.*") + + # + # 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, {}), + # NOTE: 'scheme' kwd is deprecated... + (None, {"scheme": "des_crypt"}), + (1, {}), + ((), {}), + ]: + + self.assertRaises(TypeError, cc.hash_needs_update, hash, **kwds) + + cc2 = CryptContext(["mysql323"]) + self.assertRaises(TypeError, cc2.hash_needs_update, None) + + #=================================================================== + # eoc + #=================================================================== + +#============================================================================= +# LazyCryptContext +#============================================================================= +class dummy_2(uh.StaticHandler): + name = "dummy_2" + +class LazyCryptContextTest(TestCase): + descriptionPrefix = "LazyCryptContext" + + def setUp(self): + TestCase.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 +#============================================================================= diff --git a/venv/Lib/site-packages/passlib/tests/test_crypto_builtin_md4.py b/venv/Lib/site-packages/passlib/tests/test_crypto_builtin_md4.py new file mode 100644 index 0000000..0aca1eb --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_crypto_builtin_md4.py @@ -0,0 +1,160 @@ +"""passlib.tests -- unittests for passlib.crypto._md4""" +#============================================================================= +# imports +#============================================================================= +from __future__ import with_statement, division +# core +from binascii import hexlify +import hashlib +# site +# pkg +# module +from passlib.utils.compat import bascii_to_str, PY3, u +from passlib.crypto.digest import lookup_hash +from passlib.tests.utils import TestCase, skipUnless +# local +__all__ = [ + "_Common_MD4_Test", + "MD4_Builtin_Test", + "MD4_SSL_Test", +] +#============================================================================= +# test pure-python MD4 implementation +#============================================================================= +class _Common_MD4_Test(TestCase): + """common code for testing md4 backends""" + + vectors = [ + # input -> hex digest + # test vectors from http://www.faqs.org/rfcs/rfc1320.html - A.5 + (b"", "31d6cfe0d16ae931b73c59d7e0c089c0"), + (b"a", "bde52cb31de33e46245e05fbdbd6fb24"), + (b"abc", "a448017aaf21d8525fc10ae87aa6729d"), + (b"message digest", "d9130a8164549fe818874806e1c7014b"), + (b"abcdefghijklmnopqrstuvwxyz", "d79e1c308aa5bbcdeea8ed63df412da9"), + (b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789", "043f8582f241db351ce627e153e7f0e4"), + (b"12345678901234567890123456789012345678901234567890123456789012345678901234567890", "e33b4ddc9c38f2199c3e7b164fcc0536"), + ] + + def get_md4_const(self): + """ + get md4 constructor -- + overridden by subclasses to use alternate backends. + """ + return lookup_hash("md4").const + + def test_attrs(self): + """informational attributes""" + h = self.get_md4_const()() + self.assertEqual(h.name, "md4") + self.assertEqual(h.digest_size, 16) + self.assertEqual(h.block_size, 64) + + def test_md4_update(self): + """update() method""" + md4 = self.get_md4_const() + h = md4(b'') + self.assertEqual(h.hexdigest(), "31d6cfe0d16ae931b73c59d7e0c089c0") + + h.update(b'a') + self.assertEqual(h.hexdigest(), "bde52cb31de33e46245e05fbdbd6fb24") + + h.update(b'bcdefghijklmnopqrstuvwxyz') + self.assertEqual(h.hexdigest(), "d79e1c308aa5bbcdeea8ed63df412da9") + + if PY3: + # reject unicode, hash should return digest of b'' + h = md4() + self.assertRaises(TypeError, h.update, u('a')) + self.assertEqual(h.hexdigest(), "31d6cfe0d16ae931b73c59d7e0c089c0") + else: + # coerce unicode to ascii, hash should return digest of b'a' + h = md4() + h.update(u('a')) + self.assertEqual(h.hexdigest(), "bde52cb31de33e46245e05fbdbd6fb24") + + def test_md4_hexdigest(self): + """hexdigest() method""" + md4 = self.get_md4_const() + for input, hex in self.vectors: + out = md4(input).hexdigest() + self.assertEqual(out, hex) + + def test_md4_digest(self): + """digest() method""" + md4 = self.get_md4_const() + for input, hex in self.vectors: + out = bascii_to_str(hexlify(md4(input).digest())) + self.assertEqual(out, hex) + + def test_md4_copy(self): + """copy() method""" + md4 = self.get_md4_const() + h = md4(b'abc') + + h2 = h.copy() + h2.update(b'def') + self.assertEqual(h2.hexdigest(), '804e7f1c2586e50b49ac65db5b645131') + + h.update(b'ghi') + self.assertEqual(h.hexdigest(), 'c5225580bfe176f6deeee33dee98732c') + + +#------------------------------------------------------------------------ +# create subclasses to test various backends +#------------------------------------------------------------------------ + +def has_native_md4(): # pragma: no cover -- runtime detection + """ + check if hashlib natively supports md4. + """ + try: + hashlib.new("md4") + return True + except ValueError: + # not supported - ssl probably missing (e.g. ironpython) + return False + + +@skipUnless(has_native_md4(), "hashlib lacks ssl/md4 support") +class MD4_SSL_Test(_Common_MD4_Test): + descriptionPrefix = "hashlib.new('md4')" + + # NOTE: we trust ssl got md4 implementation right, + # this is more to test our test is correct :) + + def setUp(self): + super(MD4_SSL_Test, self).setUp() + + # make sure we're using right constructor. + self.assertEqual(self.get_md4_const().__module__, "hashlib") + + +class MD4_Builtin_Test(_Common_MD4_Test): + descriptionPrefix = "passlib.crypto._md4.md4()" + + def setUp(self): + super(MD4_Builtin_Test, self).setUp() + + if has_native_md4(): + + # Temporarily make lookup_hash() use builtin pure-python implementation, + # by monkeypatching hashlib.new() to ensure we fall back to passlib's md4 class. + orig = hashlib.new + def wrapper(name, *args): + if name == "md4": + raise ValueError("md4 disabled for testing") + return orig(name, *args) + self.patchAttr(hashlib, "new", wrapper) + + # flush cache before & after test, since we're mucking with it. + lookup_hash.clear_cache() + self.addCleanup(lookup_hash.clear_cache) + + # make sure we're using right constructor. + self.assertEqual(self.get_md4_const().__module__, "passlib.crypto._md4") + + +#============================================================================= +# eof +#============================================================================= diff --git a/venv/Lib/site-packages/passlib/tests/test_crypto_des.py b/venv/Lib/site-packages/passlib/tests/test_crypto_des.py new file mode 100644 index 0000000..ab31845 --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_crypto_des.py @@ -0,0 +1,194 @@ +"""passlib.tests -- unittests for passlib.crypto.des""" +#============================================================================= +# imports +#============================================================================= +from __future__ import with_statement, division +# core +from functools import partial +# site +# pkg +# module +from passlib.utils import getrandbytes +from passlib.tests.utils import TestCase + +#============================================================================= +# test DES routines +#============================================================================= +class DesTest(TestCase): + descriptionPrefix = "passlib.crypto.des" + + # test vectors taken from http://www.skepticfiles.org/faq/testdes.htm + des_test_vectors = [ + # key, plaintext, ciphertext + (0x0000000000000000, 0x0000000000000000, 0x8CA64DE9C1B123A7), + (0xFFFFFFFFFFFFFFFF, 0xFFFFFFFFFFFFFFFF, 0x7359B2163E4EDC58), + (0x3000000000000000, 0x1000000000000001, 0x958E6E627A05557B), + (0x1111111111111111, 0x1111111111111111, 0xF40379AB9E0EC533), + (0x0123456789ABCDEF, 0x1111111111111111, 0x17668DFC7292532D), + (0x1111111111111111, 0x0123456789ABCDEF, 0x8A5AE1F81AB8F2DD), + (0x0000000000000000, 0x0000000000000000, 0x8CA64DE9C1B123A7), + (0xFEDCBA9876543210, 0x0123456789ABCDEF, 0xED39D950FA74BCC4), + (0x7CA110454A1A6E57, 0x01A1D6D039776742, 0x690F5B0D9A26939B), + (0x0131D9619DC1376E, 0x5CD54CA83DEF57DA, 0x7A389D10354BD271), + (0x07A1133E4A0B2686, 0x0248D43806F67172, 0x868EBB51CAB4599A), + (0x3849674C2602319E, 0x51454B582DDF440A, 0x7178876E01F19B2A), + (0x04B915BA43FEB5B6, 0x42FD443059577FA2, 0xAF37FB421F8C4095), + (0x0113B970FD34F2CE, 0x059B5E0851CF143A, 0x86A560F10EC6D85B), + (0x0170F175468FB5E6, 0x0756D8E0774761D2, 0x0CD3DA020021DC09), + (0x43297FAD38E373FE, 0x762514B829BF486A, 0xEA676B2CB7DB2B7A), + (0x07A7137045DA2A16, 0x3BDD119049372802, 0xDFD64A815CAF1A0F), + (0x04689104C2FD3B2F, 0x26955F6835AF609A, 0x5C513C9C4886C088), + (0x37D06BB516CB7546, 0x164D5E404F275232, 0x0A2AEEAE3FF4AB77), + (0x1F08260D1AC2465E, 0x6B056E18759F5CCA, 0xEF1BF03E5DFA575A), + (0x584023641ABA6176, 0x004BD6EF09176062, 0x88BF0DB6D70DEE56), + (0x025816164629B007, 0x480D39006EE762F2, 0xA1F9915541020B56), + (0x49793EBC79B3258F, 0x437540C8698F3CFA, 0x6FBF1CAFCFFD0556), + (0x4FB05E1515AB73A7, 0x072D43A077075292, 0x2F22E49BAB7CA1AC), + (0x49E95D6D4CA229BF, 0x02FE55778117F12A, 0x5A6B612CC26CCE4A), + (0x018310DC409B26D6, 0x1D9D5C5018F728C2, 0x5F4C038ED12B2E41), + (0x1C587F1C13924FEF, 0x305532286D6F295A, 0x63FAC0D034D9F793), + (0x0101010101010101, 0x0123456789ABCDEF, 0x617B3A0CE8F07100), + (0x1F1F1F1F0E0E0E0E, 0x0123456789ABCDEF, 0xDB958605F8C8C606), + (0xE0FEE0FEF1FEF1FE, 0x0123456789ABCDEF, 0xEDBFD1C66C29CCC7), + (0x0000000000000000, 0xFFFFFFFFFFFFFFFF, 0x355550B2150E2451), + (0xFFFFFFFFFFFFFFFF, 0x0000000000000000, 0xCAAAAF4DEAF1DBAE), + (0x0123456789ABCDEF, 0x0000000000000000, 0xD5D44FF720683D0D), + (0xFEDCBA9876543210, 0xFFFFFFFFFFFFFFFF, 0x2A2BB008DF97C2F2), + ] + + def test_01_expand(self): + """expand_des_key()""" + from passlib.crypto.des import expand_des_key, shrink_des_key, \ + _KDATA_MASK, INT_56_MASK + + # make sure test vectors are preserved (sans parity bits) + # uses ints, bytes are tested under # 02 + for key1, _, _ in self.des_test_vectors: + key2 = shrink_des_key(key1) + key3 = expand_des_key(key2) + # NOTE: this assumes expand_des_key() sets parity bits to 0 + self.assertEqual(key3, key1 & _KDATA_MASK) + + # type checks + self.assertRaises(TypeError, expand_des_key, 1.0) + + # too large + self.assertRaises(ValueError, expand_des_key, INT_56_MASK+1) + self.assertRaises(ValueError, expand_des_key, b"\x00"*8) + + # too small + self.assertRaises(ValueError, expand_des_key, -1) + self.assertRaises(ValueError, expand_des_key, b"\x00"*6) + + def test_02_shrink(self): + """shrink_des_key()""" + from passlib.crypto.des import expand_des_key, shrink_des_key, INT_64_MASK + rng = self.getRandom() + + # make sure reverse works for some random keys + # uses bytes, ints are tested under # 01 + for i in range(20): + key1 = getrandbytes(rng, 7) + key2 = expand_des_key(key1) + key3 = shrink_des_key(key2) + self.assertEqual(key3, key1) + + # type checks + self.assertRaises(TypeError, shrink_des_key, 1.0) + + # too large + self.assertRaises(ValueError, shrink_des_key, INT_64_MASK+1) + self.assertRaises(ValueError, shrink_des_key, b"\x00"*9) + + # too small + self.assertRaises(ValueError, shrink_des_key, -1) + self.assertRaises(ValueError, shrink_des_key, b"\x00"*7) + + def _random_parity(self, key): + """randomize parity bits""" + from passlib.crypto.des import _KDATA_MASK, _KPARITY_MASK, INT_64_MASK + rng = self.getRandom() + return (key & _KDATA_MASK) | (rng.randint(0,INT_64_MASK) & _KPARITY_MASK) + + def test_03_encrypt_bytes(self): + """des_encrypt_block()""" + from passlib.crypto.des import (des_encrypt_block, shrink_des_key, + _pack64, _unpack64) + + # run through test vectors + for key, plaintext, correct in self.des_test_vectors: + # convert to bytes + key = _pack64(key) + plaintext = _pack64(plaintext) + correct = _pack64(correct) + + # test 64-bit key + result = des_encrypt_block(key, plaintext) + self.assertEqual(result, correct, "key=%r plaintext=%r:" % + (key, plaintext)) + + # test 56-bit version + key2 = shrink_des_key(key) + result = des_encrypt_block(key2, plaintext) + self.assertEqual(result, correct, "key=%r shrink(key)=%r plaintext=%r:" % + (key, key2, plaintext)) + + # test with random parity bits + for _ in range(20): + key3 = _pack64(self._random_parity(_unpack64(key))) + result = des_encrypt_block(key3, plaintext) + self.assertEqual(result, correct, "key=%r rndparity(key)=%r plaintext=%r:" % + (key, key3, plaintext)) + + # check invalid keys + stub = b'\x00' * 8 + self.assertRaises(TypeError, des_encrypt_block, 0, stub) + self.assertRaises(ValueError, des_encrypt_block, b'\x00'*6, stub) + + # check invalid input + self.assertRaises(TypeError, des_encrypt_block, stub, 0) + self.assertRaises(ValueError, des_encrypt_block, stub, b'\x00'*7) + + # check invalid salts + self.assertRaises(ValueError, des_encrypt_block, stub, stub, salt=-1) + self.assertRaises(ValueError, des_encrypt_block, stub, stub, salt=1<<24) + + # check invalid rounds + self.assertRaises(ValueError, des_encrypt_block, stub, stub, 0, rounds=0) + + def test_04_encrypt_ints(self): + """des_encrypt_int_block()""" + from passlib.crypto.des import des_encrypt_int_block + + # run through test vectors + for key, plaintext, correct in self.des_test_vectors: + # test 64-bit key + result = des_encrypt_int_block(key, plaintext) + self.assertEqual(result, correct, "key=%r plaintext=%r:" % + (key, plaintext)) + + # test with random parity bits + for _ in range(20): + key3 = self._random_parity(key) + result = des_encrypt_int_block(key3, plaintext) + self.assertEqual(result, correct, "key=%r rndparity(key)=%r plaintext=%r:" % + (key, key3, plaintext)) + + # check invalid keys + self.assertRaises(TypeError, des_encrypt_int_block, b'\x00', 0) + self.assertRaises(ValueError, des_encrypt_int_block, -1, 0) + + # check invalid input + self.assertRaises(TypeError, des_encrypt_int_block, 0, b'\x00') + self.assertRaises(ValueError, des_encrypt_int_block, 0, -1) + + # check invalid salts + self.assertRaises(ValueError, des_encrypt_int_block, 0, 0, salt=-1) + self.assertRaises(ValueError, des_encrypt_int_block, 0, 0, salt=1<<24) + + # check invalid rounds + self.assertRaises(ValueError, des_encrypt_int_block, 0, 0, 0, rounds=0) + +#============================================================================= +# eof +#============================================================================= diff --git a/venv/Lib/site-packages/passlib/tests/test_crypto_digest.py b/venv/Lib/site-packages/passlib/tests/test_crypto_digest.py new file mode 100644 index 0000000..461d209 --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_crypto_digest.py @@ -0,0 +1,544 @@ +"""tests for passlib.utils.(des|pbkdf2|md4)""" +#============================================================================= +# imports +#============================================================================= +from __future__ import with_statement, division +# core +from binascii import hexlify +import hashlib +import warnings +# site +# pkg +# module +from passlib.exc import UnknownHashError +from passlib.utils.compat import PY3, u, JYTHON +from passlib.tests.utils import TestCase, TEST_MODE, skipUnless, hb + +#============================================================================= +# test assorted crypto helpers +#============================================================================= +class HashInfoTest(TestCase): + """test various crypto functions""" + descriptionPrefix = "passlib.crypto.digest" + + #: list of formats norm_hash_name() should support + norm_hash_formats = ["hashlib", "iana"] + + #: test cases for norm_hash_name() + #: each row contains (iana name, hashlib name, ... 0+ unnormalized names) + norm_hash_samples = [ + # real hashes + ("md5", "md5", "SCRAM-MD5-PLUS", "MD-5"), + ("sha1", "sha-1", "SCRAM-SHA-1", "SHA1"), + ("sha256", "sha-256", "SHA_256", "sha2-256"), + ("ripemd160", "ripemd-160", "SCRAM-RIPEMD-160", "RIPEmd160", + # NOTE: there was an older "RIPEMD" & "RIPEMD-128", but python treates "RIPEMD" + # as alias for "RIPEMD-160" + "ripemd", "SCRAM-RIPEMD"), + + # fake hashes (to check if fallback normalization behaves sanely) + ("sha4_256", "sha4-256", "SHA4-256", "SHA-4-256"), + ("test128", "test-128", "TEST128"), + ("test2", "test2", "TEST-2"), + ("test3_128", "test3-128", "TEST-3-128"), + ] + + def test_norm_hash_name(self): + """norm_hash_name()""" + from itertools import chain + from passlib.crypto.digest import norm_hash_name, _known_hash_names + + # snapshot warning state, ignore unknown hash warnings + ctx = warnings.catch_warnings() + ctx.__enter__() + self.addCleanup(ctx.__exit__) + warnings.filterwarnings("ignore", '.*unknown hash') + warnings.filterwarnings("ignore", '.*unsupported hash') + + # test string types + self.assertEqual(norm_hash_name(u("MD4")), "md4") + self.assertEqual(norm_hash_name(b"MD4"), "md4") + self.assertRaises(TypeError, norm_hash_name, None) + + # test selected results + for row in chain(_known_hash_names, self.norm_hash_samples): + for idx, format in enumerate(self.norm_hash_formats): + correct = row[idx] + for value in row: + result = norm_hash_name(value, format) + self.assertEqual(result, correct, + "name=%r, format=%r:" % (value, + format)) + + def test_lookup_hash_ctor(self): + """lookup_hash() -- constructor""" + from passlib.crypto.digest import lookup_hash + + # invalid/unknown names should be rejected + self.assertRaises(ValueError, lookup_hash, "new") + self.assertRaises(ValueError, lookup_hash, "__name__") + self.assertRaises(ValueError, lookup_hash, "sha4") + + # 1. should return hashlib builtin if found + self.assertEqual(lookup_hash("md5"), (hashlib.md5, 16, 64)) + + # 2. should return wrapper around hashlib.new() if found + try: + hashlib.new("sha") + has_sha = True + except ValueError: + has_sha = False + if has_sha: + record = lookup_hash("sha") + const = record[0] + self.assertEqual(record, (const, 20, 64)) + self.assertEqual(hexlify(const(b"abc").digest()), + b"0164b8a914cd2a5e74c4f7ff082c4d97f1edf880") + + else: + self.assertRaises(ValueError, lookup_hash, "sha") + + # 3. should fall back to builtin md4 + try: + hashlib.new("md4") + has_md4 = True + except ValueError: + has_md4 = False + record = lookup_hash("md4") + const = record[0] + if not has_md4: + from passlib.crypto._md4 import md4 + self.assertIs(const, md4) + self.assertEqual(record, (const, 16, 64)) + self.assertEqual(hexlify(const(b"abc").digest()), + b"a448017aaf21d8525fc10ae87aa6729d") + + # should memoize records + self.assertIs(lookup_hash("md5"), lookup_hash("md5")) + + def test_lookup_hash_w_unknown_name(self): + """lookup_hash() -- unknown hash name""" + from passlib.crypto.digest import lookup_hash + + # unknown names should be rejected by default + self.assertRaises(UnknownHashError, lookup_hash, "xxx256") + + # required=False should return stub record instead + info = lookup_hash("xxx256", required=False) + self.assertFalse(info.supported) + self.assertRaisesRegex(UnknownHashError, "unknown hash: 'xxx256'", info.const) + self.assertEqual(info.name, "xxx256") + self.assertEqual(info.digest_size, None) + self.assertEqual(info.block_size, None) + + # should cache stub records + info2 = lookup_hash("xxx256", required=False) + self.assertIs(info2, info) + + def test_mock_fips_mode(self): + """ + lookup_hash() -- test set_mock_fips_mode() + """ + from passlib.crypto.digest import lookup_hash, _set_mock_fips_mode + + # check if md5 is available so we can test mock helper + if not lookup_hash("md5", required=False).supported: + raise self.skipTest("md5 not supported") + + # enable monkeypatch to mock up fips mode + _set_mock_fips_mode() + self.addCleanup(_set_mock_fips_mode, False) + + pat = "'md5' hash disabled for fips" + self.assertRaisesRegex(UnknownHashError, pat, lookup_hash, "md5") + + info = lookup_hash("md5", required=False) + self.assertRegex(info.error_text, pat) + self.assertRaisesRegex(UnknownHashError, pat, info.const) + + # should use hardcoded fallback info + self.assertEqual(info.digest_size, 16) + self.assertEqual(info.block_size, 64) + + def test_lookup_hash_metadata(self): + """lookup_hash() -- metadata""" + + from passlib.crypto.digest import lookup_hash + + # quick test of metadata using known reference - sha256 + info = lookup_hash("sha256") + self.assertEqual(info.name, "sha256") + self.assertEqual(info.iana_name, "sha-256") + self.assertEqual(info.block_size, 64) + self.assertEqual(info.digest_size, 32) + self.assertIs(lookup_hash("SHA2-256"), info) + + # quick test of metadata using known reference - md5 + info = lookup_hash("md5") + self.assertEqual(info.name, "md5") + self.assertEqual(info.iana_name, "md5") + self.assertEqual(info.block_size, 64) + self.assertEqual(info.digest_size, 16) + + def test_lookup_hash_alt_types(self): + """lookup_hash() -- alternate types""" + + from passlib.crypto.digest import lookup_hash + + info = lookup_hash("sha256") + self.assertIs(lookup_hash(info), info) + self.assertIs(lookup_hash(info.const), info) + + self.assertRaises(TypeError, lookup_hash, 123) + + # TODO: write full test of compile_hmac() -- currently relying on pbkdf2_hmac() tests + +#============================================================================= +# test PBKDF1 support +#============================================================================= +class Pbkdf1_Test(TestCase): + """test kdf helpers""" + descriptionPrefix = "passlib.crypto.digest.pbkdf1" + + pbkdf1_tests = [ + # (password, salt, rounds, keylen, hash, result) + + # + # from http://www.di-mgt.com.au/cryptoKDFs.html + # + (b'password', hb('78578E5A5D63CB06'), 1000, 16, 'sha1', hb('dc19847e05c64d2faf10ebfb4a3d2a20')), + + # + # custom + # + (b'password', b'salt', 1000, 0, 'md5', b''), + (b'password', b'salt', 1000, 1, 'md5', hb('84')), + (b'password', b'salt', 1000, 8, 'md5', hb('8475c6a8531a5d27')), + (b'password', b'salt', 1000, 16, 'md5', hb('8475c6a8531a5d27e386cd496457812c')), + (b'password', b'salt', 1000, None, 'md5', hb('8475c6a8531a5d27e386cd496457812c')), + (b'password', b'salt', 1000, None, 'sha1', hb('4a8fd48e426ed081b535be5769892fa396293efb')), + ] + if not JYTHON: # FIXME: find out why not jython, or reenable this. + pbkdf1_tests.append( + (b'password', b'salt', 1000, None, 'md4', hb('f7f2e91100a8f96190f2dd177cb26453')) + ) + + def test_known(self): + """test reference vectors""" + from passlib.crypto.digest import pbkdf1 + for secret, salt, rounds, keylen, digest, correct in self.pbkdf1_tests: + result = pbkdf1(digest, secret, salt, rounds, keylen) + self.assertEqual(result, correct) + + def test_border(self): + """test border cases""" + from passlib.crypto.digest import pbkdf1 + def helper(secret=b'secret', salt=b'salt', rounds=1, keylen=1, hash='md5'): + return pbkdf1(hash, secret, salt, rounds, keylen) + helper() + + # salt/secret wrong type + self.assertRaises(TypeError, helper, secret=1) + self.assertRaises(TypeError, helper, salt=1) + + # non-existent hashes + self.assertRaises(ValueError, helper, hash='missing') + + # rounds < 1 and wrong type + self.assertRaises(ValueError, helper, rounds=0) + self.assertRaises(TypeError, helper, rounds='1') + + # keylen < 0, keylen > block_size, and wrong type + self.assertRaises(ValueError, helper, keylen=-1) + self.assertRaises(ValueError, helper, keylen=17, hash='md5') + self.assertRaises(TypeError, helper, keylen='1') + +#============================================================================= +# test PBKDF2-HMAC support +#============================================================================= + +# import the test subject +from passlib.crypto.digest import pbkdf2_hmac, PBKDF2_BACKENDS + +# NOTE: relying on tox to verify this works under all the various backends. +class Pbkdf2Test(TestCase): + """test pbkdf2() support""" + descriptionPrefix = "passlib.crypto.digest.pbkdf2_hmac() " % ", ".join(PBKDF2_BACKENDS) + + pbkdf2_test_vectors = [ + # (result, secret, salt, rounds, keylen, digest="sha1") + + # + # from rfc 3962 + # + + # test case 1 / 128 bit + ( + hb("cdedb5281bb2f801565a1122b2563515"), + b"password", b"ATHENA.MIT.EDUraeburn", 1, 16 + ), + + # test case 2 / 128 bit + ( + hb("01dbee7f4a9e243e988b62c73cda935d"), + b"password", b"ATHENA.MIT.EDUraeburn", 2, 16 + ), + + # test case 2 / 256 bit + ( + hb("01dbee7f4a9e243e988b62c73cda935da05378b93244ec8f48a99e61ad799d86"), + b"password", b"ATHENA.MIT.EDUraeburn", 2, 32 + ), + + # test case 3 / 256 bit + ( + hb("5c08eb61fdf71e4e4ec3cf6ba1f5512ba7e52ddbc5e5142f708a31e2e62b1e13"), + b"password", b"ATHENA.MIT.EDUraeburn", 1200, 32 + ), + + # test case 4 / 256 bit + ( + hb("d1daa78615f287e6a1c8b120d7062a493f98d203e6be49a6adf4fa574b6e64ee"), + b"password", b'\x12\x34\x56\x78\x78\x56\x34\x12', 5, 32 + ), + + # test case 5 / 256 bit + ( + hb("139c30c0966bc32ba55fdbf212530ac9c5ec59f1a452f5cc9ad940fea0598ed1"), + b"X"*64, b"pass phrase equals block size", 1200, 32 + ), + + # test case 6 / 256 bit + ( + hb("9ccad6d468770cd51b10e6a68721be611a8b4d282601db3b36be9246915ec82a"), + b"X"*65, b"pass phrase exceeds block size", 1200, 32 + ), + + # + # from rfc 6070 + # + ( + hb("0c60c80f961f0e71f3a9b524af6012062fe037a6"), + b"password", b"salt", 1, 20, + ), + + ( + hb("ea6c014dc72d6f8ccd1ed92ace1d41f0d8de8957"), + b"password", b"salt", 2, 20, + ), + + ( + hb("4b007901b765489abead49d926f721d065a429c1"), + b"password", b"salt", 4096, 20, + ), + + # just runs too long - could enable if ALL option is set + ##( + ## + ## hb("eefe3d61cd4da4e4e9945b3d6ba2158c2634e984"), + ## "password", "salt", 16777216, 20, + ##), + + ( + hb("3d2eec4fe41c849b80c8d83662c0e44a8b291a964cf2f07038"), + b"passwordPASSWORDpassword", + b"saltSALTsaltSALTsaltSALTsaltSALTsalt", + 4096, 25, + ), + + ( + hb("56fa6aa75548099dcc37d7f03425e0c3"), + b"pass\00word", b"sa\00lt", 4096, 16, + ), + + # + # from example in http://grub.enbug.org/Authentication + # + ( + hb("887CFF169EA8335235D8004242AA7D6187A41E3187DF0CE14E256D85ED" + "97A97357AAA8FF0A3871AB9EEFF458392F462F495487387F685B7472FC" + "6C29E293F0A0"), + b"hello", + hb("9290F727ED06C38BA4549EF7DE25CF5642659211B7FC076F2D28FEFD71" + "784BB8D8F6FB244A8CC5C06240631B97008565A120764C0EE9C2CB0073" + "994D79080136"), + 10000, 64, "sha512" + ), + + # + # test vectors from fastpbkdf2 + # + ( + hb('55ac046e56e3089fec1691c22544b605f94185216dde0465e68b9d57c20dacbc' + '49ca9cccf179b645991664b39d77ef317c71b845b1e30bd509112041d3a19783'), + b'passwd', b'salt', 1, 64, 'sha256', + ), + + ( + hb('4ddcd8f60b98be21830cee5ef22701f9641a4418d04c0414aeff08876b34ab56' + 'a1d425a1225833549adb841b51c9b3176a272bdebba1d078478f62b397f33c8d'), + b'Password', b'NaCl', 80000, 64, 'sha256', + ), + + ( + hb('120fb6cffcf8b32c43e7225256c4f837a86548c92ccc35480805987cb70be17b'), + b'password', b'salt', 1, 32, 'sha256', + ), + + ( + hb('ae4d0c95af6b46d32d0adff928f06dd02a303f8ef3c251dfd6e2d85a95474c43'), + b'password', b'salt', 2, 32, 'sha256', + ), + + ( + hb('c5e478d59288c841aa530db6845c4c8d962893a001ce4e11a4963873aa98134a'), + b'password', b'salt', 4096, 32, 'sha256', + ), + + ( + hb('348c89dbcbd32b2f32d814b8116e84cf2b17347ebc1800181c4e2a1fb8dd53e1c' + '635518c7dac47e9'), + b'passwordPASSWORDpassword', b'saltSALTsaltSALTsaltSALTsaltSALTsalt', + 4096, 40, 'sha256', + ), + + ( + hb('9e83f279c040f2a11aa4a02b24c418f2d3cb39560c9627fa4f47e3bcc2897c3d'), + b'', b'salt', 1024, 32, 'sha256', + ), + + ( + hb('ea5808411eb0c7e830deab55096cee582761e22a9bc034e3ece925225b07bf46'), + b'password', b'', 1024, 32, 'sha256', + ), + + ( + hb('89b69d0516f829893c696226650a8687'), + b'pass\x00word', b'sa\x00lt', 4096, 16, 'sha256', + ), + + ( + hb('867f70cf1ade02cff3752599a3a53dc4af34c7a669815ae5d513554e1c8cf252'), + b'password', b'salt', 1, 32, 'sha512', + ), + + ( + hb('e1d9c16aa681708a45f5c7c4e215ceb66e011a2e9f0040713f18aefdb866d53c'), + b'password', b'salt', 2, 32, 'sha512', + ), + + ( + hb('d197b1b33db0143e018b12f3d1d1479e6cdebdcc97c5c0f87f6902e072f457b5'), + b'password', b'salt', 4096, 32, 'sha512', + ), + + ( + hb('6e23f27638084b0f7ea1734e0d9841f55dd29ea60a834466f3396bac801fac1eeb' + '63802f03a0b4acd7603e3699c8b74437be83ff01ad7f55dac1ef60f4d56480c35e' + 'e68fd52c6936'), + b'passwordPASSWORDpassword', b'saltSALTsaltSALTsaltSALTsaltSALTsalt', + 1, 72, 'sha512', + ), + + ( + hb('0c60c80f961f0e71f3a9b524af6012062fe037a6'), + b'password', b'salt', 1, 20, 'sha1', + ), + + # + # custom tests + # + ( + hb('e248fb6b13365146f8ac6307cc222812'), + b"secret", b"salt", 10, 16, "sha1", + ), + ( + hb('e248fb6b13365146f8ac6307cc2228127872da6d'), + b"secret", b"salt", 10, None, "sha1", + ), + ( + hb('b1d5485772e6f76d5ebdc11b38d3eff0a5b2bd50dc11f937e86ecacd0cd40d1b' + '9113e0734e3b76a3'), + b"secret", b"salt", 62, 40, "md5", + ), + ( + hb('ea014cc01f78d3883cac364bb5d054e2be238fb0b6081795a9d84512126e3129' + '062104d2183464c4'), + b"secret", b"salt", 62, 40, "md4", + ), + ] + + def test_known(self): + """test reference vectors""" + for row in self.pbkdf2_test_vectors: + correct, secret, salt, rounds, keylen = row[:5] + digest = row[5] if len(row) == 6 else "sha1" + result = pbkdf2_hmac(digest, secret, salt, rounds, keylen) + self.assertEqual(result, correct) + + def test_backends(self): + """verify expected backends are present""" + from passlib.crypto.digest import PBKDF2_BACKENDS + + # check for fastpbkdf2 + try: + import fastpbkdf2 + has_fastpbkdf2 = True + except ImportError: + has_fastpbkdf2 = False + self.assertEqual("fastpbkdf2" in PBKDF2_BACKENDS, has_fastpbkdf2) + + # check for hashlib + try: + from hashlib import pbkdf2_hmac + has_hashlib_ssl = pbkdf2_hmac.__module__ != "hashlib" + except ImportError: + has_hashlib_ssl = False + self.assertEqual("hashlib-ssl" in PBKDF2_BACKENDS, has_hashlib_ssl) + + # check for appropriate builtin + from passlib.utils.compat import PY3 + if PY3: + self.assertIn("builtin-from-bytes", PBKDF2_BACKENDS) + else: + # XXX: only true as long as this is preferred over hexlify + self.assertIn("builtin-unpack", PBKDF2_BACKENDS) + + def test_border(self): + """test border cases""" + def helper(secret=b'password', salt=b'salt', rounds=1, keylen=None, digest="sha1"): + return pbkdf2_hmac(digest, secret, salt, rounds, keylen) + helper() + + # invalid rounds + self.assertRaises(ValueError, helper, rounds=-1) + self.assertRaises(ValueError, helper, rounds=0) + self.assertRaises(TypeError, helper, rounds='x') + + # invalid keylen + helper(keylen=1) + self.assertRaises(ValueError, helper, keylen=-1) + self.assertRaises(ValueError, helper, keylen=0) + # NOTE: hashlib actually throws error for keylen>=MAX_SINT32, + # but pbkdf2 forbids anything > MAX_UINT32 * digest_size + self.assertRaises(OverflowError, helper, keylen=20*(2**32-1)+1) + self.assertRaises(TypeError, helper, keylen='x') + + # invalid secret/salt type + self.assertRaises(TypeError, helper, salt=5) + self.assertRaises(TypeError, helper, secret=5) + + # invalid hash + self.assertRaises(ValueError, helper, digest='foo') + self.assertRaises(TypeError, helper, digest=5) + + def test_default_keylen(self): + """test keylen==None""" + def helper(secret=b'password', salt=b'salt', rounds=1, keylen=None, digest="sha1"): + return pbkdf2_hmac(digest, secret, salt, rounds, keylen) + self.assertEqual(len(helper(digest='sha1')), 20) + self.assertEqual(len(helper(digest='sha256')), 32) + +#============================================================================= +# eof +#=============================================================================