path: root/passlib/tests/test_context.py

"""tests for passlib.pwhash -- (c) Assurance Technologies 2003-2009"""
#=========================================================
#imports
#=========================================================
from __future__ import with_statement
#core
import hashlib
from logging import getLogger
import os
import time
import warnings
import sys
#site
try:
    from pkg_resources import resource_filename
except ImportError:
    resource_filename = None
#pkg
from passlib import hash
from passlib.context import CryptContext, CryptPolicy, LazyCryptContext
from passlib.utils import to_bytes, to_unicode, PasslibPolicyWarning
from passlib.utils.compat import irange
import passlib.utils.handlers as uh
from passlib.tests.utils import TestCase, mktemp, catch_warnings, \
    gae_env, set_file
from passlib.registry import register_crypt_handler_path, has_crypt_handler, \
    _unload_handler_name as unload_handler_name
#module
log = getLogger(__name__)

#=========================================================
#
#=========================================================
class CryptPolicyTest(TestCase):
    "test CryptPolicy object"

    #TODO: need to test user categories w/in all this

    case_prefix = "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",
        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_1pid = {
        "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_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.
        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 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",
        all__vary_rounds = "10%",
        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",
        all__vary_rounds = "10%",
        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" ],
         all__vary_rounds = "10%",
        sha512_crypt__max_rounds = 20000,
        admin__all__vary_rounds = "5%",
        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 test_00_constructor(self):
        "test CryptPolicy() constructor"
        policy = CryptPolicy(**self.sample_config_1pd)
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #check key with too many separators is rejected
        self.assertRaises(TypeError, CryptPolicy,
            schemes = [ "des_crypt", "md5_crypt", "bsdi_crypt", "sha512_crypt"],
            bad__key__bsdi_crypt__max_rounds = 30000,
            )

        #check nameless handler rejected
        self.assertRaises(ValueError, CryptPolicy, schemes=[uh.StaticHandler])

        #check name conflicts are rejected
        class dummy_1(uh.StaticHandler):
            name = 'dummy_1'
        self.assertRaises(KeyError, CryptPolicy, schemes=[dummy_1, dummy_1])

        #with unknown deprecated value
        self.assertRaises(KeyError, CryptPolicy,
                          schemes=['des_crypt'],
                          deprecated=['md5_crypt'])

        #with unknown default value
        self.assertRaises(KeyError, CryptPolicy,
                          schemes=['des_crypt'],
                          default='md5_crypt')

    def test_01_from_path_simple(self):
        "test CryptPolicy.from_path() constructor"
        #NOTE: this is separate so it can also run under GAE

        #test preset stored in existing file
        path = self.sample_config_1s_path
        policy = CryptPolicy.from_path(path)
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #test if path missing
        self.assertRaises(EnvironmentError, CryptPolicy.from_path, path + 'xxx')

    def test_01_from_path(self):
        "test CryptPolicy.from_path() constructor with encodings"
        if gae_env:
            return self.skipTest("GAE doesn't offer read/write filesystem access")

        path = mktemp()

        #test "\n" linesep
        set_file(path, self.sample_config_1s)
        policy = CryptPolicy.from_path(path)
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #test "\r\n" linesep
        set_file(path, self.sample_config_1s.replace("\n","\r\n"))
        policy = CryptPolicy.from_path(path)
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #test with custom encoding
        uc2 = to_bytes(self.sample_config_1s, "utf-16", source_encoding="utf-8")
        set_file(path, uc2)
        policy = CryptPolicy.from_path(path, encoding="utf-16")
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

    def test_02_from_string(self):
        "test CryptPolicy.from_string() constructor"
        #test "\n" linesep
        policy = CryptPolicy.from_string(self.sample_config_1s)
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #test "\r\n" linesep
        policy = CryptPolicy.from_string(
            self.sample_config_1s.replace("\n","\r\n"))
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #test with unicode
        data = to_unicode(self.sample_config_1s)
        policy = CryptPolicy.from_string(data)
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #test with non-ascii-compatible encoding
        uc2 = to_bytes(self.sample_config_1s, "utf-16", source_encoding="utf-8")
        policy = CryptPolicy.from_string(uc2, encoding="utf-16")
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #test category specific options
        policy = CryptPolicy.from_string(self.sample_config_4s)
        self.assertEqual(policy.to_dict(), self.sample_config_4pd)

    def test_03_from_source(self):
        "test CryptPolicy.from_source() constructor"
        #pass it a path
        policy = CryptPolicy.from_source(self.sample_config_1s_path)
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #pass it a string
        policy = CryptPolicy.from_source(self.sample_config_1s)
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #pass it a dict (NOTE: make a copy to detect in-place modifications)
        policy = CryptPolicy.from_source(self.sample_config_1pd.copy())
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #pass it existing policy
        p2 = CryptPolicy.from_source(policy)
        self.assertIs(policy, p2)

        #pass it something wrong
        self.assertRaises(TypeError, CryptPolicy.from_source, 1)
        self.assertRaises(TypeError, CryptPolicy.from_source, [])

    def test_04_from_sources(self):
        "test CryptPolicy.from_sources() constructor"

        #pass it empty list
        self.assertRaises(ValueError, CryptPolicy.from_sources, [])

        #pass it one-element list
        policy = CryptPolicy.from_sources([self.sample_config_1s])
        self.assertEqual(policy.to_dict(), self.sample_config_1pd)

        #pass multiple sources
        policy = CryptPolicy.from_sources(
            [
            self.sample_config_1s_path,
            self.sample_config_2s,
            self.sample_config_3pd,
            ])
        self.assertEqual(policy.to_dict(), self.sample_config_123pd)

    def test_05_replace(self):
        "test CryptPolicy.replace() constructor"

        p1 = CryptPolicy(**self.sample_config_1pd)

        #check overlaying sample 2
        p2 = p1.replace(**self.sample_config_2pd)
        self.assertEqual(p2.to_dict(), self.sample_config_12pd)

        #check repeating overlay makes no change
        p2b = p2.replace(**self.sample_config_2pd)
        self.assertEqual(p2b.to_dict(), self.sample_config_12pd)

        #check overlaying sample 3
        p3 = p2.replace(self.sample_config_3pd)
        self.assertEqual(p3.to_dict(), self.sample_config_123pd)

    def test_06_forbidden(self):
        "test CryptPolicy() forbidden kwds"

        #salt not allowed to be set
        self.assertRaises(KeyError, CryptPolicy,
            schemes=["des_crypt"],
            des_crypt__salt="xx",
        )
        self.assertRaises(KeyError, CryptPolicy,
            schemes=["des_crypt"],
            all__salt="xx",
        )

        #schemes not allowed for category
        self.assertRaises(KeyError, CryptPolicy,
            schemes=["des_crypt"],
            user__context__schemes=["md5_crypt"],
        )

    #=========================================================
    #reading
    #=========================================================
    def test_10_has_schemes(self):
        "test has_schemes() method"

        p1 = CryptPolicy(**self.sample_config_1pd)
        self.assertTrue(p1.has_schemes())

        p3 = CryptPolicy(**self.sample_config_3pd)
        self.assertTrue(not p3.has_schemes())

    def test_11_iter_handlers(self):
        "test iter_handlers() method"

        p1 = CryptPolicy(**self.sample_config_1pd)
        s = self.sample_config_1prd['schemes']
        self.assertEqual(list(p1.iter_handlers()), s)

        p3 = CryptPolicy(**self.sample_config_3pd)
        self.assertEqual(list(p3.iter_handlers()), [])

    def test_12_get_handler(self):
        "test get_handler() method"

        p1 = CryptPolicy(**self.sample_config_1pd)

        #check by name
        self.assertIs(p1.get_handler("bsdi_crypt"), hash.bsdi_crypt)

        #check by missing name
        self.assertIs(p1.get_handler("sha256_crypt"), None)
        self.assertRaises(KeyError, p1.get_handler, "sha256_crypt", required=True)

        #check default
        self.assertIs(p1.get_handler(), hash.md5_crypt)

    def test_13_get_options(self):
        "test get_options() method"

        p12 = CryptPolicy(**self.sample_config_12pd)

        self.assertEqual(p12.get_options("bsdi_crypt"),dict(
            vary_rounds = "10%",
            min_rounds = 29000,
            max_rounds = 35000,
            default_rounds = 31000,
        ))

        self.assertEqual(p12.get_options("sha512_crypt"),dict(
            vary_rounds = "10%",
            min_rounds = 45000,
            max_rounds = 50000,
        ))

        p4 = CryptPolicy.from_string(self.sample_config_4s)
        self.assertEqual(p4.get_options("sha512_crypt"), dict(
            vary_rounds="10%",
            max_rounds=20000,
        ))

        self.assertEqual(p4.get_options("sha512_crypt", "user"), dict(
            vary_rounds="10%",
            max_rounds=20000,
        ))

        self.assertEqual(p4.get_options("sha512_crypt", "admin"), dict(
            vary_rounds="5%",
            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"
        pa = CryptPolicy()
        self.assertEqual(pa.get_min_verify_time(), 0)
        self.assertEqual(pa.get_min_verify_time('admin'), 0)

        pb = pa.replace(min_verify_time=.1)
        self.assertEqual(pb.get_min_verify_time(), .1)
        self.assertEqual(pb.get_min_verify_time('admin'), .1)

        pc = pa.replace(admin__context__min_verify_time=.2)
        self.assertEqual(pc.get_min_verify_time(), 0)
        self.assertEqual(pc.get_min_verify_time('admin'), .2)

        pd = pb.replace(admin__context__min_verify_time=.2)
        self.assertEqual(pd.get_min_verify_time(), .1)
        self.assertEqual(pd.get_min_verify_time('admin'), .2)

    #=========================================================
    #serialization
    #=========================================================
    def test_20_iter_config(self):
        "test iter_config() method"
        p5 = CryptPolicy(**self.sample_config_5pd)
        self.assertEqual(dict(p5.iter_config()), self.sample_config_5pd)
        self.assertEqual(dict(p5.iter_config(resolve=True)), self.sample_config_5prd)
        self.assertEqual(dict(p5.iter_config(ini=True)), self.sample_config_5pid)

    def test_21_to_dict(self):
        "test to_dict() method"
        p5 = CryptPolicy(**self.sample_config_5pd)
        self.assertEqual(p5.to_dict(), self.sample_config_5pd)
        self.assertEqual(p5.to_dict(resolve=True), self.sample_config_5prd)

    def test_22_to_string(self):
        "test to_string() method"
        pa = CryptPolicy(**self.sample_config_5pd)
        s = pa.to_string() #NOTE: can't compare string directly, ordering etc may not match
        pb = CryptPolicy.from_string(s)
        self.assertEqual(pb.to_dict(), self.sample_config_5pd)

    #=========================================================
    #
    #=========================================================

#=========================================================
#CryptContext
#=========================================================
class CryptContextTest(TestCase):
    "test CryptContext object's behavior"
    case_prefix = "CryptContext"

    #=========================================================
    #constructor
    #=========================================================
    def test_00_constructor(self):
        "test constructor"
        #create crypt context using handlers
        cc = CryptContext([hash.md5_crypt, hash.bsdi_crypt, hash.des_crypt])
        c,b,a = cc.policy.iter_handlers()
        self.assertIs(a, hash.des_crypt)
        self.assertIs(b, hash.bsdi_crypt)
        self.assertIs(c, hash.md5_crypt)

        #create context using names
        cc = CryptContext(["md5_crypt", "bsdi_crypt", "des_crypt"])
        c,b,a = cc.policy.iter_handlers()
        self.assertIs(a, hash.des_crypt)
        self.assertIs(b, hash.bsdi_crypt)
        self.assertIs(c, hash.md5_crypt)

        #TODO: test policy & other options

    def test_01_replace(self):
        "test replace()"

        cc = CryptContext(["md5_crypt", "bsdi_crypt", "des_crypt"])
        self.assertIs(cc.policy.get_handler(), hash.md5_crypt)

        cc2 = cc.replace()
        self.assertIsNot(cc2, cc)
        self.assertIs(cc2.policy, cc.policy)

        cc3 = cc.replace(default="bsdi_crypt")
        self.assertIsNot(cc3, cc)
        self.assertIsNot(cc3.policy, cc.policy)
        self.assertIs(cc3.policy.get_handler(), hash.bsdi_crypt)

    def test_02_no_handlers(self):
        "test no handlers"

        #check constructor...
        cc = CryptContext()
        self.assertRaises(KeyError, cc.identify, 'hash', required=True)
        self.assertRaises(KeyError, cc.encrypt, 'secret')
        self.assertRaises(KeyError, cc.verify, 'secret', 'hash')

        #check updating policy after the fact...
        cc = CryptContext(['md5_crypt'])
        p = CryptPolicy(schemes=[])
        cc.policy = p

        self.assertRaises(KeyError, cc.identify, 'hash', required=True)
        self.assertRaises(KeyError, cc.encrypt, 'secret')
        self.assertRaises(KeyError, cc.verify, 'secret', 'hash')

    #=========================================================
    #policy adaptation
    #=========================================================
    sample_policy_1 = dict(
            schemes = [ "des_crypt", "md5_crypt", "nthash", "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,
            nthash__ident = "NT",
    )

    def test_10_01_genconfig_settings(self):
        "test genconfig() settings"
        cc = CryptContext(policy=None,
                          schemes=["md5_crypt", "nthash"],
                          nthash__ident="NT",
                         )

        # hash specific settings
        self.assertTrue(cc.genconfig().startswith("$1$"))
        self.assertEqual(
            cc.genconfig(scheme="nthash"),
            '$NT$00000000000000000000000000000000',
            )
        self.assertEqual(
            cc.genconfig(scheme="nthash", ident="3"),
            '$3$$00000000000000000000000000000000',
            )

    def test_10_02_genconfig_rounds_limits(self):
        "test genconfig() policy rounds limits"
        cc = CryptContext(policy=None,
                          schemes=["sha256_crypt"],
                          all__min_rounds=2000,
                          all__max_rounds=3000,
                          all__default_rounds=2500,
                          )

        # min rounds
        with catch_warnings(record=True) as wlog:

            # set below handler min
            c2 = cc.replace(all__min_rounds=500, all__max_rounds=None,
                            all__default_rounds=500)
            self.assertWarningMatches(wlog.pop(), category=PasslibPolicyWarning)
            self.assertWarningMatches(wlog.pop(), category=PasslibPolicyWarning)
            self.assertEqual(c2.genconfig(salt="nacl"), "$5$rounds=1000$nacl$")
            self.assertNoWarnings(wlog)

            # below
            self.assertEqual(
                cc.genconfig(rounds=1999, salt="nacl"),
                '$5$rounds=2000$nacl$',
                )
            self.assertWarningMatches(wlog.pop(), category=PasslibPolicyWarning)
            self.assertNoWarnings(wlog)

            # equal
            self.assertEqual(
                cc.genconfig(rounds=2000, salt="nacl"),
                '$5$rounds=2000$nacl$',
                )
            self.assertNoWarnings(wlog)

            # above
            self.assertEqual(
                cc.genconfig(rounds=2001, salt="nacl"),
                '$5$rounds=2001$nacl$'
                )
            self.assertNoWarnings(wlog)

        # max rounds
        with catch_warnings(record=True) as wlog:
            # set above handler max
            c2 = cc.replace(all__max_rounds=int(1e9)+500, all__min_rounds=None,
                            all__default_rounds=int(1e9)+500)
            self.assertWarningMatches(wlog.pop(), category=PasslibPolicyWarning)
            self.assertWarningMatches(wlog.pop(), category=PasslibPolicyWarning)
            self.assertEqual(c2.genconfig(salt="nacl"),
                             "$5$rounds=999999999$nacl$")
            self.assertNoWarnings(wlog)

            # above
            self.assertEqual(
                cc.genconfig(rounds=3001, salt="nacl"),
                '$5$rounds=3000$nacl$'
                )
            self.assertWarningMatches(wlog.pop(), category=PasslibPolicyWarning)
            self.assertNoWarnings(wlog)

            # equal
            self.assertEqual(
                cc.genconfig(rounds=3000, salt="nacl"),
                '$5$rounds=3000$nacl$'
                )
            self.assertNoWarnings(wlog)

            # below
            self.assertEqual(
                cc.genconfig(rounds=2999, salt="nacl"),
                '$5$rounds=2999$nacl$',
                )
            self.assertNoWarnings(wlog)

        # explicit default rounds
        self.assertEqual(cc.genconfig(salt="nacl"), '$5$rounds=2500$nacl$')

        # fallback default rounds - use handler's
        c2 = cc.replace(all__default_rounds=None, all__max_rounds=50000)
        self.assertEqual(c2.genconfig(salt="nacl"), '$5$rounds=40000$nacl$')

        # fallback default rounds - use handler's, but clipped to max rounds
        c2 = cc.replace(all__default_rounds=None, all__max_rounds=3000)
        self.assertEqual(c2.genconfig(salt="nacl"), '$5$rounds=3000$nacl$')

        # TODO: test default falls back to mx / mn if handler has no default.

        #default rounds - out of bounds
        self.assertRaises(ValueError, cc.replace, all__default_rounds=1999)
        cc.policy.replace(all__default_rounds=2000)
        cc.policy.replace(all__default_rounds=3000)
        self.assertRaises(ValueError, cc.replace, all__default_rounds=3001)

        # invalid min/max bounds
        c2 = CryptContext(policy=None, schemes=["sha256_crypt"])
        self.assertRaises(ValueError, c2.replace, all__min_rounds=-1)
        self.assertRaises(ValueError, c2.replace, all__max_rounds=-1)
        self.assertRaises(ValueError, c2.replace, all__min_rounds=2000,
                          all__max_rounds=1999)

    def test_10_03_genconfig_linear_vary_rounds(self):
        "test genconfig() linear vary rounds"
        cc = CryptContext(policy=None,
                          schemes=["sha256_crypt"],
                          all__min_rounds=1995,
                          all__max_rounds=2005,
                          all__default_rounds=2000,
                          )

        # test negative
        self.assertRaises(ValueError, cc.replace, all__vary_rounds=-1)
        self.assertRaises(ValueError, cc.replace, all__vary_rounds="-1%")
        self.assertRaises(ValueError, cc.replace, all__vary_rounds="101%")

        # test static
        c2 = cc.replace(all__vary_rounds=0)
        self.assert_rounds_range(c2, "sha256_crypt", 2000, 2000)

        c2 = cc.replace(all__vary_rounds="0%")
        self.assert_rounds_range(c2, "sha256_crypt", 2000, 2000)

        # test absolute
        c2 = cc.replace(all__vary_rounds=1)
        self.assert_rounds_range(c2, "sha256_crypt", 1999, 2001)
        c2 = cc.replace(all__vary_rounds=100)
        self.assert_rounds_range(c2, "sha256_crypt", 1995, 2005)

        # test relative
        c2 = cc.replace(all__vary_rounds="0.1%")
        self.assert_rounds_range(c2, "sha256_crypt", 1998, 2002)
        c2 = cc.replace(all__vary_rounds="100%")
        self.assert_rounds_range(c2, "sha256_crypt", 1995, 2005)

    def test_10_03_genconfig_log2_vary_rounds(self):
        "test genconfig() log2 vary rounds"
        cc = CryptContext(policy=None,
                          schemes=["bcrypt"],
                          all__min_rounds=15,
                          all__max_rounds=25,
                          all__default_rounds=20,
                          )

        # test negative
        self.assertRaises(ValueError, cc.replace, all__vary_rounds=-1)
        self.assertRaises(ValueError, cc.replace, all__vary_rounds="-1%")
        self.assertRaises(ValueError, cc.replace, all__vary_rounds="101%")

        # test static
        c2 = cc.replace(all__vary_rounds=0)
        self.assert_rounds_range(c2, "bcrypt", 20, 20)

        c2 = cc.replace(all__vary_rounds="0%")
        self.assert_rounds_range(c2, "bcrypt", 20, 20)

        # test absolute
        c2 = cc.replace(all__vary_rounds=1)
        self.assert_rounds_range(c2, "bcrypt", 19, 21)
        c2 = cc.replace(all__vary_rounds=100)
        self.assert_rounds_range(c2, "bcrypt", 15, 25)

        # test relative - should shift over at 50% mark
        c2 = cc.replace(all__vary_rounds="1%")
        self.assert_rounds_range(c2, "bcrypt", 20, 20)

        c2 = cc.replace(all__vary_rounds="49%")
        self.assert_rounds_range(c2, "bcrypt", 20, 20)

        c2 = cc.replace(all__vary_rounds="50%")
        self.assert_rounds_range(c2, "bcrypt", 19, 20)

        c2 = cc.replace(all__vary_rounds="100%")
        self.assert_rounds_range(c2, "bcrypt", 15, 21)

    def assert_rounds_range(self, context, scheme, lower, upper, salt="."*22):
        "helper to check vary_rounds covers specified range"
        # NOTE: this runs enough times the min and max *should* be hit,
        # though there's a faint chance it will randomly fail.
        handler = context.policy.get_handler(scheme)
        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 lower bound:")
        self.assertEqual(max(seen), upper, "vary_rounds upper bound:")

    def test_11_encrypt_settings(self):
        "test encrypt() honors policy settings"
        cc = CryptContext(**self.sample_policy_1)

        # hash specific settings
        self.assertEqual(
            cc.encrypt("password", scheme="nthash"),
            '$NT$8846f7eaee8fb117ad06bdd830b7586c',
            )
        self.assertEqual(
            cc.encrypt("password", scheme="nthash", ident="3"),
            '$3$$8846f7eaee8fb117ad06bdd830b7586c',
            )

        # NOTE: more thorough job of rounds limits done in genconfig() test,
        # which is much cheaper, and shares the same codebase.

        # min rounds
        with catch_warnings(record=True) as wlog:
            self.assertEqual(
                cc.encrypt("password", rounds=1999, salt="nacl"),
                '$5$rounds=2000$nacl$9/lTZ5nrfPuz8vphznnmHuDGFuvjSNvOEDsGmGfsS97',
                )
            self.assertWarningMatches(wlog.pop(), category=PasslibPolicyWarning)
            self.assertFalse(wlog)

            self.assertEqual(
                cc.encrypt("password", rounds=2001, salt="nacl"),
                '$5$rounds=2001$nacl$8PdeoPL4aXQnJ0woHhqgIw/efyfCKC2WHneOpnvF.31'
                )
            self.assertFalse(wlog)

        # max rounds, etc tested in genconfig()

        # make default > max throws error if attempted
        self.assertRaises(ValueError, cc.replace,
                          sha256_crypt__default_rounds=4000)

    def test_12_hash_needs_update(self):
        "test hash_needs_update() method"
        cc = CryptContext(**self.sample_policy_1)

        #check deprecated scheme
        self.assertTrue(cc.hash_needs_update('9XXD4trGYeGJA'))
        self.assertFalse(cc.hash_needs_update('$1$J8HC2RCr$HcmM.7NxB2weSvlw2FgzU0'))

        #check min rounds
        self.assertTrue(cc.hash_needs_update('$5$rounds=1999$jD81UCoo.zI.UETs$Y7qSTQ6mTiU9qZB4fRr43wRgQq4V.5AAf7F97Pzxey/'))
        self.assertFalse(cc.hash_needs_update('$5$rounds=2000$228SSRje04cnNCaQ$YGV4RYu.5sNiBvorQDlO0WWQjyJVGKBcJXz3OtyQ2u8'))

        #check max rounds
        self.assertFalse(cc.hash_needs_update('$5$rounds=3000$fS9iazEwTKi7QPW4$VasgBC8FqlOvD7x2HhABaMXCTh9jwHclPA9j5YQdns.'))
        self.assertTrue(cc.hash_needs_update('$5$rounds=3001$QlFHHifXvpFX4PLs$/0ekt7lSs/lOikSerQ0M/1porEHxYq7W/2hdFpxA3fA'))

    #=========================================================
    #identify
    #=========================================================
    def test_20_basic(self):
        "test basic encrypt/identify/verify functionality"
        handlers = [hash.md5_crypt, hash.des_crypt, hash.bsdi_crypt]
        cc = CryptContext(handlers, policy=None)

        #run through handlers
        for crypt in handlers:
            h = cc.encrypt("test", scheme=crypt.name)
            self.assertEqual(cc.identify(h), crypt.name)
            self.assertEqual(cc.identify(h, resolve=True), crypt)
            self.assertTrue(cc.verify('test', h))
            self.assertTrue(not cc.verify('notest', h))

        #test default
        h = cc.encrypt("test")
        self.assertEqual(cc.identify(h), "md5_crypt")

        #test genhash
        h = cc.genhash('secret', cc.genconfig())
        self.assertEqual(cc.identify(h), 'md5_crypt')

        h = cc.genhash('secret', cc.genconfig(), scheme='md5_crypt')
        self.assertEqual(cc.identify(h), 'md5_crypt')

        self.assertRaises(ValueError, cc.genhash, 'secret', cc.genconfig(), scheme="des_crypt")

    def test_21_identify(self):
        "test identify() border cases"
        handlers = ["md5_crypt", "des_crypt", "bsdi_crypt"]
        cc = CryptContext(handlers, policy=None)

        #check unknown hash
        self.assertEqual(cc.identify('$9$232323123$1287319827'), None)
        self.assertRaises(ValueError, cc.identify, '$9$232323123$1287319827', required=True)

        #make sure "None" is accepted
        self.assertEqual(cc.identify(None), None)
        self.assertRaises(ValueError, cc.identify, None, required=True)

    def test_22_verify(self):
        "test verify() scheme kwd"
        handlers = ["md5_crypt", "des_crypt", "bsdi_crypt"]
        cc = CryptContext(handlers, policy=None)

        h = hash.md5_crypt.encrypt("test")

        #check base verify
        self.assertTrue(cc.verify("test", h))
        self.assertTrue(not cc.verify("notest", h))

        #check verify using right alg
        self.assertTrue(cc.verify('test', h, scheme='md5_crypt'))
        self.assertTrue(not cc.verify('notest', h, scheme='md5_crypt'))

        #check verify using wrong alg
        self.assertRaises(ValueError, cc.verify, 'test', h, scheme='bsdi_crypt')

    def test_23_verify_empty_hash(self):
        "test verify() allows hash=None"
        handlers = [hash.md5_crypt, hash.des_crypt, hash.bsdi_crypt]
        cc = CryptContext(handlers, policy=None)
        self.assertTrue(not cc.verify("test", None))
        for handler in handlers:
            self.assertTrue(not cc.verify("test", None, scheme=handler.name))

    def test_24_min_verify_time(self):
        "test verify() honors min_verify_time"
        #NOTE: this whole test assumes time.sleep() and time.time()
        #      have at least 1ms accuracy
        delta = .1
        min_delay = delta
        min_verify_time = min_delay + 2*delta
        max_delay = min_verify_time + 2*delta

        class TimedHash(uh.StaticHandler):
            "psuedo hash that takes specified amount of time"
            name = "timed_hash"
            delay = 0

            @classmethod
            def identify(cls, hash):
                return True

            @classmethod
            def genhash(cls, secret, hash):
                time.sleep(cls.delay)
                return hash or 'x'

        cc = CryptContext([TimedHash], min_verify_time=min_verify_time)

        def timecall(func, *args, **kwds):
            start = time.time()
            result = func(*args, **kwds)
            end = time.time()
            return end-start, result

        #verify hashing works
        TimedHash.delay = min_delay
        elapsed, _ = timecall(TimedHash.genhash, 'stub', 'stub')
        self.assertAlmostEqual(elapsed, min_delay, delta=delta)

        #ensure min verify time is honored
        elapsed, _ = timecall(cc.verify, "stub", "stub")
        self.assertAlmostEqual(elapsed, min_verify_time, delta=delta)

        #ensure taking longer emits a warning.
        TimedHash.delay = max_delay
        with catch_warnings(record=True) as wlog:
            warnings.simplefilter("always")
            elapsed, _ = timecall(cc.verify, "stub", "stub")
        self.assertAlmostEqual(elapsed, max_delay, delta=delta)
        self.assertEqual(len(wlog), 1)
        self.assertWarningMatches(wlog[0],
            message_re="CryptContext: verify exceeded min_verify_time")

    def test_25_verify_and_update(self):
        "test verify_and_update()"
        cc = CryptContext(**self.sample_policy_1)

        #create some hashes
        h1 = cc.encrypt("password", scheme="des_crypt")
        h2 = cc.encrypt("password", scheme="sha256_crypt")

        #check bad password, deprecated hash
        ok, new_hash = cc.verify_and_update("wrongpass", h1)
        self.assertFalse(ok)
        self.assertIs(new_hash, None)

        #check bad password, good hash
        ok, new_hash = cc.verify_and_update("wrongpass", h2)
        self.assertFalse(ok)
        self.assertIs(new_hash, None)

        #check right password, deprecated hash
        ok, new_hash = cc.verify_and_update("password", h1)
        self.assertTrue(ok)
        self.assertTrue(cc.identify(new_hash), "sha256_crypt")

        #check right password, good hash
        ok, new_hash = cc.verify_and_update("password", h2)
        self.assertTrue(ok)
        self.assertIs(new_hash, None)

    #=========================================================
    # other
    #=========================================================
    def test_90_bcrypt_normhash(self):
        "teset verify_and_update / hash_needs_update corrects bcrypt padding"
        # see issue 25.
        bcrypt = hash.bcrypt

        PASS1 = "loppux"
        BAD1  = "$2a$12$oaQbBqq8JnSM1NHRPQGXORm4GCUMqp7meTnkft4zgSnrbhoKdDV0C"
        GOOD1 = "$2a$12$oaQbBqq8JnSM1NHRPQGXOOm4GCUMqp7meTnkft4zgSnrbhoKdDV0C"
        ctx = CryptContext(["bcrypt"])

        with catch_warnings(record=True) as wlog:
            warnings.simplefilter("always")

            self.assertTrue(ctx.hash_needs_update(BAD1))
            self.assertFalse(ctx.hash_needs_update(GOOD1))

            if bcrypt.has_backend():
                self.assertEquals(ctx.verify_and_update(PASS1,GOOD1), (True,None))
                self.assertEquals(ctx.verify_and_update("x",BAD1), (False,None))
                res = ctx.verify_and_update(PASS1, BAD1)
                self.assertTrue(res[0] and res[1] and res[1] != BAD1)

    #=========================================================
    #eoc
    #=========================================================

#=========================================================
#LazyCryptContext
#=========================================================
class dummy_2(uh.StaticHandler):
    name = "dummy_2"

class LazyCryptContextTest(TestCase):
    case_prefix = "LazyCryptContext"

    def setUp(self):
        unload_handler_name("dummy_2")

    def tearDown(self):
        unload_handler_name("dummy_2")

    def test_kwd_constructor(self):
        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):
        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
#=========================================================