From 717a5d6b7cf98c4c901731ac9e71952b299bc3fa Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 18:52:00 +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 --- .../site-packages/passlib/tests/test_hosts.py | 97 + .../site-packages/passlib/tests/test_pwd.py | 205 +++ .../passlib/tests/test_registry.py | 228 +++ .../site-packages/passlib/tests/test_totp.py | 1604 +++++++++++++++++ .../site-packages/passlib/tests/test_utils.py | 1171 ++++++++++++ 5 files changed, 3305 insertions(+) create mode 100644 venv/Lib/site-packages/passlib/tests/test_hosts.py create mode 100644 venv/Lib/site-packages/passlib/tests/test_pwd.py create mode 100644 venv/Lib/site-packages/passlib/tests/test_registry.py create mode 100644 venv/Lib/site-packages/passlib/tests/test_totp.py create mode 100644 venv/Lib/site-packages/passlib/tests/test_utils.py diff --git a/venv/Lib/site-packages/passlib/tests/test_hosts.py b/venv/Lib/site-packages/passlib/tests/test_hosts.py new file mode 100644 index 0000000..cbf93ab --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_hosts.py @@ -0,0 +1,97 @@ +"""test passlib.hosts""" +#============================================================================= +# imports +#============================================================================= +from __future__ import with_statement +# core +import logging; log = logging.getLogger(__name__) +# site +# pkg +from passlib import hosts, hash as hashmod +from passlib.utils import unix_crypt_schemes +from passlib.tests.utils import TestCase +# module + +#============================================================================= +# test predefined app contexts +#============================================================================= +class HostsTest(TestCase): + """perform general tests to make sure contexts work""" + # NOTE: these tests are not really comprehensive, + # since they would do little but duplicate + # the presets in apps.py + # + # they mainly try to ensure no typos + # or dynamic behavior foul-ups. + + def check_unix_disabled(self, ctx): + for hash in [ + "", + "!", + "*", + "!$1$TXl/FX/U$BZge.lr.ux6ekjEjxmzwz0", + ]: + self.assertEqual(ctx.identify(hash), 'unix_disabled') + self.assertFalse(ctx.verify('test', hash)) + + def test_linux_context(self): + ctx = hosts.linux_context + for hash in [ + ('$6$rounds=41128$VoQLvDjkaZ6L6BIE$4pt.1Ll1XdDYduEwEYPCMOBiR6W6' + 'znsyUEoNlcVXpv2gKKIbQolgmTGe6uEEVJ7azUxuc8Tf7zV9SD2z7Ij751'), + ('$5$rounds=31817$iZGmlyBQ99JSB5n6$p4E.pdPBWx19OajgjLRiOW0itGny' + 'xDGgMlDcOsfaI17'), + '$1$TXl/FX/U$BZge.lr.ux6ekjEjxmzwz0', + 'kAJJz.Rwp0A/I', + ]: + self.assertTrue(ctx.verify("test", hash)) + self.check_unix_disabled(ctx) + + def test_bsd_contexts(self): + for ctx in [ + hosts.freebsd_context, + hosts.openbsd_context, + hosts.netbsd_context, + ]: + for hash in [ + '$1$TXl/FX/U$BZge.lr.ux6ekjEjxmzwz0', + 'kAJJz.Rwp0A/I', + ]: + self.assertTrue(ctx.verify("test", hash)) + h1 = "$2a$04$yjDgE74RJkeqC0/1NheSSOrvKeu9IbKDpcQf/Ox3qsrRS/Kw42qIS" + if hashmod.bcrypt.has_backend(): + self.assertTrue(ctx.verify("test", h1)) + else: + self.assertEqual(ctx.identify(h1), "bcrypt") + self.check_unix_disabled(ctx) + + def test_host_context(self): + ctx = getattr(hosts, "host_context", None) + if not ctx: + return self.skipTest("host_context not available on this platform") + + # validate schemes is non-empty, + # and contains unix_disabled + at least one real scheme + schemes = list(ctx.schemes()) + self.assertTrue(schemes, "appears to be unix system, but no known schemes supported by crypt") + self.assertTrue('unix_disabled' in schemes) + schemes.remove("unix_disabled") + self.assertTrue(schemes, "should have schemes beside fallback scheme") + self.assertTrue(set(unix_crypt_schemes).issuperset(schemes)) + + # check for hash support + self.check_unix_disabled(ctx) + for scheme, hash in [ + ("sha512_crypt", ('$6$rounds=41128$VoQLvDjkaZ6L6BIE$4pt.1Ll1XdDYduEwEYPCMOBiR6W6' + 'znsyUEoNlcVXpv2gKKIbQolgmTGe6uEEVJ7azUxuc8Tf7zV9SD2z7Ij751')), + ("sha256_crypt", ('$5$rounds=31817$iZGmlyBQ99JSB5n6$p4E.pdPBWx19OajgjLRiOW0itGny' + 'xDGgMlDcOsfaI17')), + ("md5_crypt", '$1$TXl/FX/U$BZge.lr.ux6ekjEjxmzwz0'), + ("des_crypt", 'kAJJz.Rwp0A/I'), + ]: + if scheme in schemes: + self.assertTrue(ctx.verify("test", hash)) + +#============================================================================= +# eof +#============================================================================= diff --git a/venv/Lib/site-packages/passlib/tests/test_pwd.py b/venv/Lib/site-packages/passlib/tests/test_pwd.py new file mode 100644 index 0000000..2c983cd --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_pwd.py @@ -0,0 +1,205 @@ +"""passlib.tests -- tests for passlib.pwd""" +#============================================================================= +# imports +#============================================================================= +# core +import itertools +import logging; log = logging.getLogger(__name__) +# site +# pkg +from passlib.tests.utils import TestCase +# local +__all__ = [ + "UtilsTest", + "GenerateTest", + "StrengthTest", +] + +#============================================================================= +# +#============================================================================= +class UtilsTest(TestCase): + """test internal utilities""" + descriptionPrefix = "passlib.pwd" + + def test_self_info_rate(self): + """_self_info_rate()""" + from passlib.pwd import _self_info_rate + + self.assertEqual(_self_info_rate(""), 0) + + self.assertEqual(_self_info_rate("a" * 8), 0) + + self.assertEqual(_self_info_rate("ab"), 1) + self.assertEqual(_self_info_rate("ab" * 8), 1) + + self.assertEqual(_self_info_rate("abcd"), 2) + self.assertEqual(_self_info_rate("abcd" * 8), 2) + self.assertAlmostEqual(_self_info_rate("abcdaaaa"), 1.5488, places=4) + + # def test_total_self_info(self): + # """_total_self_info()""" + # from passlib.pwd import _total_self_info + # + # self.assertEqual(_total_self_info(""), 0) + # + # self.assertEqual(_total_self_info("a" * 8), 0) + # + # self.assertEqual(_total_self_info("ab"), 2) + # self.assertEqual(_total_self_info("ab" * 8), 16) + # + # self.assertEqual(_total_self_info("abcd"), 8) + # self.assertEqual(_total_self_info("abcd" * 8), 64) + # self.assertAlmostEqual(_total_self_info("abcdaaaa"), 12.3904, places=4) + +#============================================================================= +# word generation +#============================================================================= + +# import subject +from passlib.pwd import genword, default_charsets +ascii_62 = default_charsets['ascii_62'] +hex = default_charsets['hex'] + +class WordGeneratorTest(TestCase): + """test generation routines""" + descriptionPrefix = "passlib.pwd.genword()" + + def setUp(self): + super(WordGeneratorTest, self).setUp() + + # patch some RNG references so they're reproducible. + from passlib.pwd import SequenceGenerator + self.patchAttr(SequenceGenerator, "rng", + self.getRandom("pwd generator")) + + def assertResultContents(self, results, count, chars, unique=True): + """check result list matches expected count & charset""" + self.assertEqual(len(results), count) + if unique: + if unique is True: + unique = count + self.assertEqual(len(set(results)), unique) + self.assertEqual(set("".join(results)), set(chars)) + + def test_general(self): + """general behavior""" + + # basic usage + result = genword() + self.assertEqual(len(result), 9) + + # malformed keyword should have useful error. + self.assertRaisesRegex(TypeError, "(?i)unexpected keyword.*badkwd", genword, badkwd=True) + + def test_returns(self): + """'returns' keyword""" + # returns=int option + results = genword(returns=5000) + self.assertResultContents(results, 5000, ascii_62) + + # returns=iter option + gen = genword(returns=iter) + results = [next(gen) for _ in range(5000)] + self.assertResultContents(results, 5000, ascii_62) + + # invalid returns option + self.assertRaises(TypeError, genword, returns='invalid-type') + + def test_charset(self): + """'charset' & 'chars' options""" + # charset option + results = genword(charset="hex", returns=5000) + self.assertResultContents(results, 5000, hex) + + # chars option + # there are 3**3=27 possible combinations + results = genword(length=3, chars="abc", returns=5000) + self.assertResultContents(results, 5000, "abc", unique=27) + + # chars + charset + self.assertRaises(TypeError, genword, chars='abc', charset='hex') + + # TODO: test rng option + +#============================================================================= +# phrase generation +#============================================================================= + +# import subject +from passlib.pwd import genphrase +simple_words = ["alpha", "beta", "gamma"] + +class PhraseGeneratorTest(TestCase): + """test generation routines""" + descriptionPrefix = "passlib.pwd.genphrase()" + + def assertResultContents(self, results, count, words, unique=True, sep=" "): + """check result list matches expected count & charset""" + self.assertEqual(len(results), count) + if unique: + if unique is True: + unique = count + self.assertEqual(len(set(results)), unique) + out = set(itertools.chain.from_iterable(elem.split(sep) for elem in results)) + self.assertEqual(out, set(words)) + + def test_general(self): + """general behavior""" + + # basic usage + result = genphrase() + self.assertEqual(len(result.split(" ")), 4) # 48 / log(7776, 2) ~= 3.7 -> 4 + + # malformed keyword should have useful error. + self.assertRaisesRegex(TypeError, "(?i)unexpected keyword.*badkwd", genphrase, badkwd=True) + + def test_entropy(self): + """'length' & 'entropy' keywords""" + + # custom entropy + result = genphrase(entropy=70) + self.assertEqual(len(result.split(" ")), 6) # 70 / log(7776, 2) ~= 5.4 -> 6 + + # custom length + result = genphrase(length=3) + self.assertEqual(len(result.split(" ")), 3) + + # custom length < entropy + result = genphrase(length=3, entropy=48) + self.assertEqual(len(result.split(" ")), 4) + + # custom length > entropy + result = genphrase(length=4, entropy=12) + self.assertEqual(len(result.split(" ")), 4) + + def test_returns(self): + """'returns' keyword""" + # returns=int option + results = genphrase(returns=1000, words=simple_words) + self.assertResultContents(results, 1000, simple_words) + + # returns=iter option + gen = genphrase(returns=iter, words=simple_words) + results = [next(gen) for _ in range(1000)] + self.assertResultContents(results, 1000, simple_words) + + # invalid returns option + self.assertRaises(TypeError, genphrase, returns='invalid-type') + + def test_wordset(self): + """'wordset' & 'words' options""" + # wordset option + results = genphrase(words=simple_words, returns=5000) + self.assertResultContents(results, 5000, simple_words) + + # words option + results = genphrase(length=3, words=simple_words, returns=5000) + self.assertResultContents(results, 5000, simple_words, unique=3**3) + + # words + wordset + self.assertRaises(TypeError, genphrase, words=simple_words, wordset='bip39') + +#============================================================================= +# eof +#============================================================================= diff --git a/venv/Lib/site-packages/passlib/tests/test_registry.py b/venv/Lib/site-packages/passlib/tests/test_registry.py new file mode 100644 index 0000000..8cec48d --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_registry.py @@ -0,0 +1,228 @@ +"""tests for passlib.hash -- (c) Assurance Technologies 2003-2009""" +#============================================================================= +# imports +#============================================================================= +from __future__ import with_statement +# core +from logging import getLogger +import warnings +import sys +# site +# pkg +from passlib import hash, registry, exc +from passlib.registry import register_crypt_handler, register_crypt_handler_path, \ + get_crypt_handler, list_crypt_handlers, _unload_handler_name as unload_handler_name +import passlib.utils.handlers as uh +from passlib.tests.utils import TestCase +# module +log = getLogger(__name__) + +#============================================================================= +# dummy handlers +# +# NOTE: these are defined outside of test case +# since they're used by test_register_crypt_handler_path(), +# which needs them to be available as module globals. +#============================================================================= +class dummy_0(uh.StaticHandler): + name = "dummy_0" + +class alt_dummy_0(uh.StaticHandler): + name = "dummy_0" + +dummy_x = 1 + +#============================================================================= +# test registry +#============================================================================= +class RegistryTest(TestCase): + + descriptionPrefix = "passlib.registry" + + def setUp(self): + super(RegistryTest, self).setUp() + + # backup registry state & restore it after test. + locations = dict(registry._locations) + handlers = dict(registry._handlers) + def restore(): + registry._locations.clear() + registry._locations.update(locations) + registry._handlers.clear() + registry._handlers.update(handlers) + self.addCleanup(restore) + + def test_hash_proxy(self): + """test passlib.hash proxy object""" + # check dir works + dir(hash) + + # check repr works + repr(hash) + + # check non-existent attrs raise error + self.assertRaises(AttributeError, getattr, hash, 'fooey') + + # GAE tries to set __loader__, + # make sure that doesn't call register_crypt_handler. + old = getattr(hash, "__loader__", None) + test = object() + hash.__loader__ = test + self.assertIs(hash.__loader__, test) + if old is None: + del hash.__loader__ + self.assertFalse(hasattr(hash, "__loader__")) + else: + hash.__loader__ = old + self.assertIs(hash.__loader__, old) + + # check storing attr calls register_crypt_handler + class dummy_1(uh.StaticHandler): + name = "dummy_1" + hash.dummy_1 = dummy_1 + self.assertIs(get_crypt_handler("dummy_1"), dummy_1) + + # check storing under wrong name results in error + self.assertRaises(ValueError, setattr, hash, "dummy_1x", dummy_1) + + def test_register_crypt_handler_path(self): + """test register_crypt_handler_path()""" + # NOTE: this messes w/ internals of registry, shouldn't be used publically. + paths = registry._locations + + # check namespace is clear + self.assertTrue('dummy_0' not in paths) + self.assertFalse(hasattr(hash, 'dummy_0')) + + # check invalid names are rejected + self.assertRaises(ValueError, register_crypt_handler_path, + "dummy_0", ".test_registry") + self.assertRaises(ValueError, register_crypt_handler_path, + "dummy_0", __name__ + ":dummy_0:xxx") + self.assertRaises(ValueError, register_crypt_handler_path, + "dummy_0", __name__ + ":dummy_0.xxx") + + # try lazy load + register_crypt_handler_path('dummy_0', __name__) + self.assertTrue('dummy_0' in list_crypt_handlers()) + self.assertTrue('dummy_0' not in list_crypt_handlers(loaded_only=True)) + self.assertIs(hash.dummy_0, dummy_0) + self.assertTrue('dummy_0' in list_crypt_handlers(loaded_only=True)) + unload_handler_name('dummy_0') + + # try lazy load w/ alt + register_crypt_handler_path('dummy_0', __name__ + ':alt_dummy_0') + self.assertIs(hash.dummy_0, alt_dummy_0) + unload_handler_name('dummy_0') + + # check lazy load w/ wrong type fails + register_crypt_handler_path('dummy_x', __name__) + self.assertRaises(TypeError, get_crypt_handler, 'dummy_x') + + # check lazy load w/ wrong name fails + register_crypt_handler_path('alt_dummy_0', __name__) + self.assertRaises(ValueError, get_crypt_handler, "alt_dummy_0") + unload_handler_name("alt_dummy_0") + + # TODO: check lazy load which calls register_crypt_handler (warning should be issued) + sys.modules.pop("passlib.tests._test_bad_register", None) + register_crypt_handler_path("dummy_bad", "passlib.tests._test_bad_register") + with warnings.catch_warnings(): + warnings.filterwarnings("ignore", "xxxxxxxxxx", DeprecationWarning) + h = get_crypt_handler("dummy_bad") + from passlib.tests import _test_bad_register as tbr + self.assertIs(h, tbr.alt_dummy_bad) + + def test_register_crypt_handler(self): + """test register_crypt_handler()""" + + self.assertRaises(TypeError, register_crypt_handler, {}) + + self.assertRaises(ValueError, register_crypt_handler, type('x', (uh.StaticHandler,), dict(name=None))) + self.assertRaises(ValueError, register_crypt_handler, type('x', (uh.StaticHandler,), dict(name="AB_CD"))) + self.assertRaises(ValueError, register_crypt_handler, type('x', (uh.StaticHandler,), dict(name="ab-cd"))) + self.assertRaises(ValueError, register_crypt_handler, type('x', (uh.StaticHandler,), dict(name="ab__cd"))) + self.assertRaises(ValueError, register_crypt_handler, type('x', (uh.StaticHandler,), dict(name="default"))) + + class dummy_1(uh.StaticHandler): + name = "dummy_1" + + class dummy_1b(uh.StaticHandler): + name = "dummy_1" + + self.assertTrue('dummy_1' not in list_crypt_handlers()) + + register_crypt_handler(dummy_1) + register_crypt_handler(dummy_1) + self.assertIs(get_crypt_handler("dummy_1"), dummy_1) + + self.assertRaises(KeyError, register_crypt_handler, dummy_1b) + self.assertIs(get_crypt_handler("dummy_1"), dummy_1) + + register_crypt_handler(dummy_1b, force=True) + self.assertIs(get_crypt_handler("dummy_1"), dummy_1b) + + self.assertTrue('dummy_1' in list_crypt_handlers()) + + def test_get_crypt_handler(self): + """test get_crypt_handler()""" + + class dummy_1(uh.StaticHandler): + name = "dummy_1" + + # without available handler + self.assertRaises(KeyError, get_crypt_handler, "dummy_1") + self.assertIs(get_crypt_handler("dummy_1", None), None) + + # already loaded handler + register_crypt_handler(dummy_1) + self.assertIs(get_crypt_handler("dummy_1"), dummy_1) + + with warnings.catch_warnings(): + warnings.filterwarnings("ignore", "handler names should be lower-case, and use underscores instead of hyphens:.*", UserWarning) + + # already loaded handler, using incorrect name + self.assertIs(get_crypt_handler("DUMMY-1"), dummy_1) + + # lazy load of unloaded handler, using incorrect name + register_crypt_handler_path('dummy_0', __name__) + self.assertIs(get_crypt_handler("DUMMY-0"), dummy_0) + + # check system & private names aren't returned + from passlib import hash + hash.__dict__["_fake"] = "dummy" + for name in ["_fake", "__package__"]: + self.assertRaises(KeyError, get_crypt_handler, name) + self.assertIs(get_crypt_handler(name, None), None) + + def test_list_crypt_handlers(self): + """test list_crypt_handlers()""" + from passlib.registry import list_crypt_handlers + + # check system & private names aren't returned + hash.__dict__["_fake"] = "dummy" + for name in list_crypt_handlers(): + self.assertFalse(name.startswith("_"), "%r: " % name) + unload_handler_name("_fake") + + def test_handlers(self): + """verify we have tests for all builtin handlers""" + from passlib.registry import list_crypt_handlers + from passlib.tests.test_handlers import get_handler_case, conditionally_available_hashes + for name in list_crypt_handlers(): + # skip some wrappers that don't need independant testing + if name.startswith("ldap_") and name[5:] in list_crypt_handlers(): + continue + if name in ["roundup_plaintext"]: + continue + # check the remaining ones all have a handler + try: + self.assertTrue(get_handler_case(name)) + except exc.MissingBackendError: + if name in conditionally_available_hashes: # expected to fail on some setups + continue + raise + +#============================================================================= +# eof +#============================================================================= diff --git a/venv/Lib/site-packages/passlib/tests/test_totp.py b/venv/Lib/site-packages/passlib/tests/test_totp.py new file mode 100644 index 0000000..604d2e9 --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_totp.py @@ -0,0 +1,1604 @@ +"""passlib.tests -- test passlib.totp""" +#============================================================================= +# imports +#============================================================================= +# core +import datetime +from functools import partial +import logging; log = logging.getLogger(__name__) +import sys +import time as _time +# site +# pkg +from passlib import exc +from passlib.utils.compat import unicode, u +from passlib.tests.utils import TestCase, time_call +# subject +from passlib import totp as totp_module +from passlib.totp import TOTP, AppWallet, AES_SUPPORT +# local +__all__ = [ + "EngineTest", +] + +#============================================================================= +# helpers +#============================================================================= + +# XXX: python 3 changed what error base64.b16decode() throws, from TypeError to base64.Error(). +# it wasn't until 3.3 that base32decode() also got changed. +# really should normalize this in the code to a single BinaryDecodeError, +# predicting this cross-version is getting unmanagable. +Base32DecodeError = Base16DecodeError = TypeError +if sys.version_info >= (3,0): + from binascii import Error as Base16DecodeError +if sys.version_info >= (3,3): + from binascii import Error as Base32DecodeError + +PASS1 = "abcdef" +PASS2 = b"\x00\xFF" +KEY1 = '4AOGGDBBQSYHNTUZ' +KEY1_RAW = b'\xe0\x1cc\x0c!\x84\xb0v\xce\x99' +KEY2_RAW = b'\xee]\xcb9\x870\x06 D\xc8y/\xa54&\xe4\x9c\x13\xc2\x18' +KEY3 = 'S3JDVB7QD2R7JPXX' # used in docstrings +KEY4 = 'JBSWY3DPEHPK3PXP' # from google keyuri spec +KEY4_RAW = b'Hello!\xde\xad\xbe\xef' + +# NOTE: for randtime() below, +# * want at least 7 bits on fractional side, to test fractional times to at least 0.01s precision +# * want at least 32 bits on integer side, to test for 32-bit epoch issues. +# most systems *should* have 53 bit mantissa, leaving plenty of room on both ends, +# so using (1<<37) as scale, to allocate 16 bits on fractional side, but generate reasonable # of > 1<<32 times. +# sanity check that we're above 44 ensures minimum requirements (44 - 37 int = 7 frac) +assert sys.float_info.radix == 2, "unexpected float_info.radix" +assert sys.float_info.mant_dig >= 44, "double precision unexpectedly small" + +def _get_max_time_t(): + """ + helper to calc max_time_t constant (see below) + """ + value = 1 << 30 # even for 32 bit systems will handle this + year = 0 + while True: + next_value = value << 1 + try: + next_year = datetime.datetime.utcfromtimestamp(next_value-1).year + except (ValueError, OSError, OverflowError): + # utcfromtimestamp() may throw any of the following: + # + # * year out of range for datetime: + # py < 3.6 throws ValueError. + # (py 3.6.0 returns odd value instead, see workaround below) + # + # * int out of range for host's gmtime/localtime: + # py2 throws ValueError, py3 throws OSError. + # + # * int out of range for host's time_t: + # py2 throws ValueError, py3 throws OverflowError. + # + break + + # Workaround for python 3.6.0 issue -- + # Instead of throwing ValueError if year out of range for datetime, + # Python 3.6 will do some weird behavior that masks high bits + # e.g. (1<<40) -> year 36812, but (1<<41) -> year 6118. + # (Appears to be bug http://bugs.python.org/issue29100) + # This check stops at largest non-wrapping bit size. + if next_year < year: + break + + value = next_value + + # 'value-1' is maximum. + value -= 1 + + # check for crazy case where we're beyond what datetime supports + # (caused by bug 29100 again). compare to max value that datetime + # module supports -- datetime.datetime(9999, 12, 31, 23, 59, 59, 999999) + max_datetime_timestamp = 253402318800 + return min(value, max_datetime_timestamp) + +#: Rough approximation of max value acceptable by hosts's time_t. +#: This is frequently ~2**37 on 64 bit, and ~2**31 on 32 bit systems. +max_time_t = _get_max_time_t() + +def to_b32_size(raw_size): + return (raw_size * 8 + 4) // 5 + +#============================================================================= +# wallet +#============================================================================= +class AppWalletTest(TestCase): + descriptionPrefix = "passlib.totp.AppWallet" + + #============================================================================= + # constructor + #============================================================================= + + def test_secrets_types(self): + """constructor -- 'secrets' param -- input types""" + + # no secrets + wallet = AppWallet() + self.assertEqual(wallet._secrets, {}) + self.assertFalse(wallet.has_secrets) + + # dict + ref = {"1": b"aaa", "2": b"bbb"} + wallet = AppWallet(ref) + self.assertEqual(wallet._secrets, ref) + self.assertTrue(wallet.has_secrets) + + # # list + # wallet = AppWallet(list(ref.items())) + # self.assertEqual(wallet._secrets, ref) + + # # iter + # wallet = AppWallet(iter(ref.items())) + # self.assertEqual(wallet._secrets, ref) + + # "tag:value" string + wallet = AppWallet("\n 1: aaa\n# comment\n \n2: bbb ") + self.assertEqual(wallet._secrets, ref) + + # ensure ":" allowed in secret + wallet = AppWallet("1: aaa: bbb \n# comment\n \n2: bbb ") + self.assertEqual(wallet._secrets, {"1": b"aaa: bbb", "2": b"bbb"}) + + # json dict + wallet = AppWallet('{"1":"aaa","2":"bbb"}') + self.assertEqual(wallet._secrets, ref) + + # # json list + # wallet = AppWallet('[["1","aaa"],["2","bbb"]]') + # self.assertEqual(wallet._secrets, ref) + + # invalid type + self.assertRaises(TypeError, AppWallet, 123) + + # invalid json obj + self.assertRaises(TypeError, AppWallet, "[123]") + + # # invalid list items + # self.assertRaises(ValueError, AppWallet, ["1", b"aaa"]) + + # forbid empty secret + self.assertRaises(ValueError, AppWallet, {"1": "aaa", "2": ""}) + + def test_secrets_tags(self): + """constructor -- 'secrets' param -- tag/value normalization""" + + # test reference + ref = {"1": b"aaa", "02": b"bbb", "C": b"ccc"} + wallet = AppWallet(ref) + self.assertEqual(wallet._secrets, ref) + + # accept unicode + wallet = AppWallet({u("1"): b"aaa", u("02"): b"bbb", u("C"): b"ccc"}) + self.assertEqual(wallet._secrets, ref) + + # normalize int tags + wallet = AppWallet({1: b"aaa", "02": b"bbb", "C": b"ccc"}) + self.assertEqual(wallet._secrets, ref) + + # forbid non-str/int tags + self.assertRaises(TypeError, AppWallet, {(1,): "aaa"}) + + # accept valid tags + wallet = AppWallet({"1-2_3.4": b"aaa"}) + + # forbid invalid tags + self.assertRaises(ValueError, AppWallet, {"-abc": "aaa"}) + self.assertRaises(ValueError, AppWallet, {"ab*$": "aaa"}) + + # coerce value to bytes + wallet = AppWallet({"1": u("aaa"), "02": "bbb", "C": b"ccc"}) + self.assertEqual(wallet._secrets, ref) + + # forbid invalid value types + self.assertRaises(TypeError, AppWallet, {"1": 123}) + self.assertRaises(TypeError, AppWallet, {"1": None}) + self.assertRaises(TypeError, AppWallet, {"1": []}) + + # TODO: test secrets_path + + def test_default_tag(self): + """constructor -- 'default_tag' param""" + + # should sort numerically + wallet = AppWallet({"1": "one", "02": "two"}) + self.assertEqual(wallet.default_tag, "02") + self.assertEqual(wallet.get_secret(wallet.default_tag), b"two") + + # should sort alphabetically if non-digit present + wallet = AppWallet({"1": "one", "02": "two", "A": "aaa"}) + self.assertEqual(wallet.default_tag, "A") + self.assertEqual(wallet.get_secret(wallet.default_tag), b"aaa") + + # should use honor custom tag + wallet = AppWallet({"1": "one", "02": "two", "A": "aaa"}, default_tag="1") + self.assertEqual(wallet.default_tag, "1") + self.assertEqual(wallet.get_secret(wallet.default_tag), b"one") + + # throw error on unknown value + self.assertRaises(KeyError, AppWallet, {"1": "one", "02": "two", "A": "aaa"}, + default_tag="B") + + # should be empty + wallet = AppWallet() + self.assertEqual(wallet.default_tag, None) + self.assertRaises(KeyError, wallet.get_secret, None) + + # TODO: test 'cost' param + + #============================================================================= + # encrypt_key() & decrypt_key() helpers + #============================================================================= + def require_aes_support(self, canary=None): + if AES_SUPPORT: + canary and canary() + else: + canary and self.assertRaises(RuntimeError, canary) + raise self.skipTest("'cryptography' package not installed") + + def test_decrypt_key(self): + """.decrypt_key()""" + + wallet = AppWallet({"1": PASS1, "2": PASS2}) + + # check for support + CIPHER1 = dict(v=1, c=13, s='6D7N7W53O7HHS37NLUFQ', + k='MHCTEGSNPFN5CGBJ', t='1') + self.require_aes_support(canary=partial(wallet.decrypt_key, CIPHER1)) + + # reference key + self.assertEqual(wallet.decrypt_key(CIPHER1)[0], KEY1_RAW) + + # different salt used to encrypt same raw key + CIPHER2 = dict(v=1, c=13, s='SPZJ54Y6IPUD2BYA4C6A', + k='ZGDXXTVQOWYLC2AU', t='1') + self.assertEqual(wallet.decrypt_key(CIPHER2)[0], KEY1_RAW) + + # different sized key, password, and cost + CIPHER3 = dict(v=1, c=8, s='FCCTARTIJWE7CPQHUDKA', + k='D2DRS32YESGHHINWFFCELKN7Z6NAHM4M', t='2') + self.assertEqual(wallet.decrypt_key(CIPHER3)[0], KEY2_RAW) + + # wrong password should silently result in wrong key + temp = CIPHER1.copy() + temp.update(t='2') + self.assertEqual(wallet.decrypt_key(temp)[0], b'\xafD6.F7\xeb\x19\x05Q') + + # missing tag should throw error + temp = CIPHER1.copy() + temp.update(t='3') + self.assertRaises(KeyError, wallet.decrypt_key, temp) + + # unknown version should throw error + temp = CIPHER1.copy() + temp.update(v=999) + self.assertRaises(ValueError, wallet.decrypt_key, temp) + + def test_decrypt_key_needs_recrypt(self): + """.decrypt_key() -- needs_recrypt flag""" + self.require_aes_support() + + wallet = AppWallet({"1": PASS1, "2": PASS2}, encrypt_cost=13) + + # ref should be accepted + ref = dict(v=1, c=13, s='AAAA', k='AAAA', t='2') + self.assertFalse(wallet.decrypt_key(ref)[1]) + + # wrong cost + temp = ref.copy() + temp.update(c=8) + self.assertTrue(wallet.decrypt_key(temp)[1]) + + # wrong tag + temp = ref.copy() + temp.update(t="1") + self.assertTrue(wallet.decrypt_key(temp)[1]) + + # XXX: should this check salt_size? + + def assertSaneResult(self, result, wallet, key, tag="1", + needs_recrypt=False): + """check encrypt_key() result has expected format""" + + self.assertEqual(set(result), set(["v", "t", "c", "s", "k"])) + + self.assertEqual(result['v'], 1) + self.assertEqual(result['t'], tag) + self.assertEqual(result['c'], wallet.encrypt_cost) + + self.assertEqual(len(result['s']), to_b32_size(wallet.salt_size)) + self.assertEqual(len(result['k']), to_b32_size(len(key))) + + result_key, result_needs_recrypt = wallet.decrypt_key(result) + self.assertEqual(result_key, key) + self.assertEqual(result_needs_recrypt, needs_recrypt) + + def test_encrypt_key(self): + """.encrypt_key()""" + + # check for support + wallet = AppWallet({"1": PASS1}, encrypt_cost=5) + self.require_aes_support(canary=partial(wallet.encrypt_key, KEY1_RAW)) + + # basic behavior + result = wallet.encrypt_key(KEY1_RAW) + self.assertSaneResult(result, wallet, KEY1_RAW) + + # creates new salt each time + other = wallet.encrypt_key(KEY1_RAW) + self.assertSaneResult(result, wallet, KEY1_RAW) + self.assertNotEqual(other['s'], result['s']) + self.assertNotEqual(other['k'], result['k']) + + # honors custom cost + wallet2 = AppWallet({"1": PASS1}, encrypt_cost=6) + result = wallet2.encrypt_key(KEY1_RAW) + self.assertSaneResult(result, wallet2, KEY1_RAW) + + # honors default tag + wallet2 = AppWallet({"1": PASS1, "2": PASS2}) + result = wallet2.encrypt_key(KEY1_RAW) + self.assertSaneResult(result, wallet2, KEY1_RAW, tag="2") + + # honor salt size + wallet2 = AppWallet({"1": PASS1}) + wallet2.salt_size = 64 + result = wallet2.encrypt_key(KEY1_RAW) + self.assertSaneResult(result, wallet2, KEY1_RAW) + + # larger key + result = wallet.encrypt_key(KEY2_RAW) + self.assertSaneResult(result, wallet, KEY2_RAW) + + # border case: empty key + # XXX: might want to allow this, but documenting behavior for now + self.assertRaises(ValueError, wallet.encrypt_key, b"") + + def test_encrypt_cost_timing(self): + """verify cost parameter via timing""" + self.require_aes_support() + + # time default cost + wallet = AppWallet({"1": "aaa"}) + wallet.encrypt_cost -= 2 + delta, _ = time_call(partial(wallet.encrypt_key, KEY1_RAW), maxtime=0) + + # this should take (2**3=8) times as long + wallet.encrypt_cost += 3 + delta2, _ = time_call(partial(wallet.encrypt_key, KEY1_RAW), maxtime=0) + + # TODO: rework timing test here to inject mock pbkdf2_hmac() function instead; + # and test that it's being invoked w/ proper options. + self.assertAlmostEqual(delta2, delta*8, delta=(delta*8)*0.5) + + #============================================================================= + # eoc + #============================================================================= + +#============================================================================= +# common OTP code +#============================================================================= + +#: used as base value for RFC test vector keys +RFC_KEY_BYTES_20 = "12345678901234567890".encode("ascii") +RFC_KEY_BYTES_32 = (RFC_KEY_BYTES_20*2)[:32] +RFC_KEY_BYTES_64 = (RFC_KEY_BYTES_20*4)[:64] + +# TODO: this class is separate from TotpTest due to historical issue, +# when there was a base class, and a separate HOTP class. +# these test case classes should probably be combined. +class TotpTest(TestCase): + """ + common code shared by TotpTest & HotpTest + """ + #============================================================================= + # class attrs + #============================================================================= + + descriptionPrefix = "passlib.totp.TOTP" + + #============================================================================= + # setup + #============================================================================= + def setUp(self): + super(TotpTest, self).setUp() + + # clear norm_hash_name() cache so 'unknown hash' warnings get emitted each time + from passlib.crypto.digest import lookup_hash + lookup_hash.clear_cache() + + # monkeypatch module's rng to be deterministic + self.patchAttr(totp_module, "rng", self.getRandom()) + + #============================================================================= + # general helpers + #============================================================================= + def randtime(self): + """ + helper to generate random epoch time + :returns float: epoch time + """ + return self.getRandom().random() * max_time_t + + def randotp(self, cls=None, **kwds): + """ + helper which generates a random TOTP instance. + """ + rng = self.getRandom() + if "key" not in kwds: + kwds['new'] = True + kwds.setdefault("digits", rng.randint(6, 10)) + kwds.setdefault("alg", rng.choice(["sha1", "sha256", "sha512"])) + kwds.setdefault("period", rng.randint(10, 120)) + return (cls or TOTP)(**kwds) + + def test_randotp(self): + """ + internal test -- randotp() + """ + otp1 = self.randotp() + otp2 = self.randotp() + + self.assertNotEqual(otp1.key, otp2.key, "key not randomized:") + + # NOTE: has (1/5)**10 odds of failure + for _ in range(10): + if otp1.digits != otp2.digits: + break + otp2 = self.randotp() + else: + self.fail("digits not randomized") + + # NOTE: has (1/3)**10 odds of failure + for _ in range(10): + if otp1.alg != otp2.alg: + break + otp2 = self.randotp() + else: + self.fail("alg not randomized") + + #============================================================================= + # reference vector helpers + #============================================================================= + + #: default options used by test vectors (unless otherwise stated) + vector_defaults = dict(format="base32", alg="sha1", period=30, digits=8) + + #: various TOTP test vectors, + #: each element in list has format [options, (time, token <, int(expires)>), ...] + vectors = [ + + #------------------------------------------------------------------------- + # passlib test vectors + #------------------------------------------------------------------------- + + # 10 byte key, 6 digits + [dict(key="ACDEFGHJKL234567", digits=6), + # test fencepost to make sure we're rounding right + (1412873399, '221105'), # == 29 mod 30 + (1412873400, '178491'), # == 0 mod 30 + (1412873401, '178491'), # == 1 mod 30 + (1412873429, '178491'), # == 29 mod 30 + (1412873430, '915114'), # == 0 mod 30 + ], + + # 10 byte key, 8 digits + [dict(key="ACDEFGHJKL234567", digits=8), + # should be same as 6 digits (above), but w/ 2 more digits on left side of token. + (1412873399, '20221105'), # == 29 mod 30 + (1412873400, '86178491'), # == 0 mod 30 + (1412873401, '86178491'), # == 1 mod 30 + (1412873429, '86178491'), # == 29 mod 30 + (1412873430, '03915114'), # == 0 mod 30 + ], + + # sanity check on key used in docstrings + [dict(key="S3JD-VB7Q-D2R7-JPXX", digits=6), + (1419622709, '000492'), + (1419622739, '897212'), + ], + + #------------------------------------------------------------------------- + # reference vectors taken from http://tools.ietf.org/html/rfc6238, appendix B + # NOTE: while appendix B states same key used for all tests, the reference + # code in the appendix repeats the key up to the alg's block size, + # and uses *that* as the secret... so that's what we're doing here. + #------------------------------------------------------------------------- + + # sha1 test vectors + [dict(key=RFC_KEY_BYTES_20, format="raw", alg="sha1"), + (59, '94287082'), + (1111111109, '07081804'), + (1111111111, '14050471'), + (1234567890, '89005924'), + (2000000000, '69279037'), + (20000000000, '65353130'), + ], + + # sha256 test vectors + [dict(key=RFC_KEY_BYTES_32, format="raw", alg="sha256"), + (59, '46119246'), + (1111111109, '68084774'), + (1111111111, '67062674'), + (1234567890, '91819424'), + (2000000000, '90698825'), + (20000000000, '77737706'), + ], + + # sha512 test vectors + [dict(key=RFC_KEY_BYTES_64, format="raw", alg="sha512"), + (59, '90693936'), + (1111111109, '25091201'), + (1111111111, '99943326'), + (1234567890, '93441116'), + (2000000000, '38618901'), + (20000000000, '47863826'), + ], + + #------------------------------------------------------------------------- + # other test vectors + #------------------------------------------------------------------------- + + # generated at http://blog.tinisles.com/2011/10/google-authenticator-one-time-password-algorithm-in-javascript + [dict(key="JBSWY3DPEHPK3PXP", digits=6), (1409192430, '727248'), (1419890990, '122419')], + [dict(key="JBSWY3DPEHPK3PXP", digits=9, period=41), (1419891152, '662331049')], + + # found in https://github.com/eloquent/otis/blob/develop/test/suite/Totp/Value/TotpValueGeneratorTest.php, line 45 + [dict(key=RFC_KEY_BYTES_20, format="raw", period=60), (1111111111, '19360094')], + [dict(key=RFC_KEY_BYTES_32, format="raw", alg="sha256", period=60), (1111111111, '40857319')], + [dict(key=RFC_KEY_BYTES_64, format="raw", alg="sha512", period=60), (1111111111, '37023009')], + + ] + + def iter_test_vectors(self): + """ + helper to iterate over test vectors. + yields ``(totp, time, token, expires, prefix)`` tuples. + """ + from passlib.totp import TOTP + for row in self.vectors: + kwds = self.vector_defaults.copy() + kwds.update(row[0]) + for entry in row[1:]: + if len(entry) == 3: + time, token, expires = entry + else: + time, token = entry + expires = None + # NOTE: not re-using otp between calls so that stateful methods + # (like .match) don't have problems. + log.debug("test vector: %r time=%r token=%r expires=%r", kwds, time, token, expires) + otp = TOTP(**kwds) + prefix = "alg=%r time=%r token=%r: " % (otp.alg, time, token) + yield otp, time, token, expires, prefix + + #============================================================================= + # constructor tests + #============================================================================= + def test_ctor_w_new(self): + """constructor -- 'new' parameter""" + + # exactly one of 'key' or 'new' is required + self.assertRaises(TypeError, TOTP) + self.assertRaises(TypeError, TOTP, key='4aoggdbbqsyhntuz', new=True) + + # generates new key + otp = TOTP(new=True) + otp2 = TOTP(new=True) + self.assertNotEqual(otp.key, otp2.key) + + def test_ctor_w_size(self): + """constructor -- 'size' parameter""" + + # should default to digest size, per RFC + self.assertEqual(len(TOTP(new=True, alg="sha1").key), 20) + self.assertEqual(len(TOTP(new=True, alg="sha256").key), 32) + self.assertEqual(len(TOTP(new=True, alg="sha512").key), 64) + + # explicit key size + self.assertEqual(len(TOTP(new=True, size=10).key), 10) + self.assertEqual(len(TOTP(new=True, size=16).key), 16) + + # for new=True, maximum size enforced (based on alg) + self.assertRaises(ValueError, TOTP, new=True, size=21, alg="sha1") + + # for new=True, minimum size enforced + self.assertRaises(ValueError, TOTP, new=True, size=9) + + # for existing key, minimum size is only warned about + with self.assertWarningList([ + dict(category=exc.PasslibSecurityWarning, message_re=".*for security purposes, secret key must be.*") + ]): + _ = TOTP('0A'*9, 'hex') + + def test_ctor_w_key_and_format(self): + """constructor -- 'key' and 'format' parameters""" + + # handle base32 encoding (the default) + self.assertEqual(TOTP(KEY1).key, KEY1_RAW) + + # .. w/ lower case + self.assertEqual(TOTP(KEY1.lower()).key, KEY1_RAW) + + # .. w/ spaces (e.g. user-entered data) + self.assertEqual(TOTP(' 4aog gdbb qsyh ntuz ').key, KEY1_RAW) + + # .. w/ invalid char + self.assertRaises(Base32DecodeError, TOTP, 'ao!ggdbbqsyhntuz') + + # handle hex encoding + self.assertEqual(TOTP('e01c630c2184b076ce99', 'hex').key, KEY1_RAW) + + # .. w/ invalid char + self.assertRaises(Base16DecodeError, TOTP, 'X01c630c2184b076ce99', 'hex') + + # handle raw bytes + self.assertEqual(TOTP(KEY1_RAW, "raw").key, KEY1_RAW) + + def test_ctor_w_alg(self): + """constructor -- 'alg' parameter""" + + # normalize hash names + self.assertEqual(TOTP(KEY1, alg="SHA-256").alg, "sha256") + self.assertEqual(TOTP(KEY1, alg="SHA256").alg, "sha256") + + # invalid alg + self.assertRaises(ValueError, TOTP, KEY1, alg="SHA-333") + + def test_ctor_w_digits(self): + """constructor -- 'digits' parameter""" + self.assertRaises(ValueError, TOTP, KEY1, digits=5) + self.assertEqual(TOTP(KEY1, digits=6).digits, 6) # min value + self.assertEqual(TOTP(KEY1, digits=10).digits, 10) # max value + self.assertRaises(ValueError, TOTP, KEY1, digits=11) + + def test_ctor_w_period(self): + """constructor -- 'period' parameter""" + + # default + self.assertEqual(TOTP(KEY1).period, 30) + + # explicit value + self.assertEqual(TOTP(KEY1, period=63).period, 63) + + # reject wrong type + self.assertRaises(TypeError, TOTP, KEY1, period=1.5) + self.assertRaises(TypeError, TOTP, KEY1, period='abc') + + # reject non-positive values + self.assertRaises(ValueError, TOTP, KEY1, period=0) + self.assertRaises(ValueError, TOTP, KEY1, period=-1) + + def test_ctor_w_label(self): + """constructor -- 'label' parameter""" + self.assertEqual(TOTP(KEY1).label, None) + self.assertEqual(TOTP(KEY1, label="foo@bar").label, "foo@bar") + self.assertRaises(ValueError, TOTP, KEY1, label="foo:bar") + + def test_ctor_w_issuer(self): + """constructor -- 'issuer' parameter""" + self.assertEqual(TOTP(KEY1).issuer, None) + self.assertEqual(TOTP(KEY1, issuer="foo.com").issuer, "foo.com") + self.assertRaises(ValueError, TOTP, KEY1, issuer="foo.com:bar") + + #============================================================================= + # using() tests + #============================================================================= + + # TODO: test using() w/ 'digits', 'alg', 'issue', 'wallet', **wallet_kwds + + def test_using_w_period(self): + """using() -- 'period' parameter""" + + # default + self.assertEqual(TOTP(KEY1).period, 30) + + # explicit value + self.assertEqual(TOTP.using(period=63)(KEY1).period, 63) + + # reject wrong type + self.assertRaises(TypeError, TOTP.using, period=1.5) + self.assertRaises(TypeError, TOTP.using, period='abc') + + # reject non-positive values + self.assertRaises(ValueError, TOTP.using, period=0) + self.assertRaises(ValueError, TOTP.using, period=-1) + + def test_using_w_now(self): + """using -- 'now' parameter""" + + # NOTE: reading time w/ normalize_time() to make sure custom .now actually has effect. + + # default -- time.time + otp = self.randotp() + self.assertIs(otp.now, _time.time) + self.assertAlmostEqual(otp.normalize_time(None), int(_time.time())) + + # custom function + counter = [123.12] + def now(): + counter[0] += 1 + return counter[0] + otp = self.randotp(cls=TOTP.using(now=now)) + # NOTE: TOTP() constructor invokes this as part of test, using up counter values 124 & 125 + self.assertEqual(otp.normalize_time(None), 126) + self.assertEqual(otp.normalize_time(None), 127) + + # require callable + self.assertRaises(TypeError, TOTP.using, now=123) + + # require returns int/float + msg_re = r"now\(\) function must return non-negative" + self.assertRaisesRegex(AssertionError, msg_re, TOTP.using, now=lambda: 'abc') + + # require returns non-negative value + self.assertRaisesRegex(AssertionError, msg_re, TOTP.using, now=lambda: -1) + + #============================================================================= + # internal method tests + #============================================================================= + + def test_normalize_token_instance(self, otp=None): + """normalize_token() -- instance method""" + if otp is None: + otp = self.randotp(digits=7) + + # unicode & bytes + self.assertEqual(otp.normalize_token(u('1234567')), '1234567') + self.assertEqual(otp.normalize_token(b'1234567'), '1234567') + + # int + self.assertEqual(otp.normalize_token(1234567), '1234567') + + # int which needs 0 padding + self.assertEqual(otp.normalize_token(234567), '0234567') + + # reject wrong types (float, None) + self.assertRaises(TypeError, otp.normalize_token, 1234567.0) + self.assertRaises(TypeError, otp.normalize_token, None) + + # too few digits + self.assertRaises(exc.MalformedTokenError, otp.normalize_token, '123456') + + # too many digits + self.assertRaises(exc.MalformedTokenError, otp.normalize_token, '01234567') + self.assertRaises(exc.MalformedTokenError, otp.normalize_token, 12345678) + + def test_normalize_token_class(self): + """normalize_token() -- class method""" + self.test_normalize_token_instance(otp=TOTP.using(digits=7)) + + def test_normalize_time(self): + """normalize_time()""" + TotpFactory = TOTP.using() + otp = self.randotp(TotpFactory) + + for _ in range(10): + time = self.randtime() + tint = int(time) + + self.assertEqual(otp.normalize_time(time), tint) + self.assertEqual(otp.normalize_time(tint + 0.5), tint) + + self.assertEqual(otp.normalize_time(tint), tint) + + dt = datetime.datetime.utcfromtimestamp(time) + self.assertEqual(otp.normalize_time(dt), tint) + + orig = TotpFactory.now + try: + TotpFactory.now = staticmethod(lambda: time) + self.assertEqual(otp.normalize_time(None), tint) + finally: + TotpFactory.now = orig + + self.assertRaises(TypeError, otp.normalize_time, '1234') + + #============================================================================= + # key attr tests + #============================================================================= + + def test_key_attrs(self): + """pretty_key() and .key attributes""" + rng = self.getRandom() + + # test key attrs + otp = TOTP(KEY1_RAW, "raw") + self.assertEqual(otp.key, KEY1_RAW) + self.assertEqual(otp.hex_key, 'e01c630c2184b076ce99') + self.assertEqual(otp.base32_key, KEY1) + + # test pretty_key() + self.assertEqual(otp.pretty_key(), '4AOG-GDBB-QSYH-NTUZ') + self.assertEqual(otp.pretty_key(sep=" "), '4AOG GDBB QSYH NTUZ') + self.assertEqual(otp.pretty_key(sep=False), KEY1) + self.assertEqual(otp.pretty_key(format="hex"), 'e01c-630c-2184-b076-ce99') + + # quick fuzz test: make attr access works for random key & random size + otp = TOTP(new=True, size=rng.randint(10, 20)) + _ = otp.hex_key + _ = otp.base32_key + _ = otp.pretty_key() + + #============================================================================= + # generate() tests + #============================================================================= + def test_totp_token(self): + """generate() -- TotpToken() class""" + from passlib.totp import TOTP, TotpToken + + # test known set of values + otp = TOTP('s3jdvb7qd2r7jpxx') + result = otp.generate(1419622739) + self.assertIsInstance(result, TotpToken) + self.assertEqual(result.token, '897212') + self.assertEqual(result.counter, 47320757) + ##self.assertEqual(result.start_time, 1419622710) + self.assertEqual(result.expire_time, 1419622740) + self.assertEqual(result, ('897212', 1419622740)) + self.assertEqual(len(result), 2) + self.assertEqual(result[0], '897212') + self.assertEqual(result[1], 1419622740) + self.assertRaises(IndexError, result.__getitem__, -3) + self.assertRaises(IndexError, result.__getitem__, 2) + self.assertTrue(result) + + # time dependant bits... + otp.now = lambda : 1419622739.5 + self.assertEqual(result.remaining, 0.5) + self.assertTrue(result.valid) + + otp.now = lambda : 1419622741 + self.assertEqual(result.remaining, 0) + self.assertFalse(result.valid) + + # same time -- shouldn't return same object, but should be equal + result2 = otp.generate(1419622739) + self.assertIsNot(result2, result) + self.assertEqual(result2, result) + + # diff time in period -- shouldn't return same object, but should be equal + result3 = otp.generate(1419622711) + self.assertIsNot(result3, result) + self.assertEqual(result3, result) + + # shouldn't be equal + result4 = otp.generate(1419622999) + self.assertNotEqual(result4, result) + + def test_generate(self): + """generate()""" + from passlib.totp import TOTP + + # generate token + otp = TOTP(new=True) + time = self.randtime() + result = otp.generate(time) + token = result.token + self.assertIsInstance(token, unicode) + start_time = result.counter * 30 + + # should generate same token for next 29s + self.assertEqual(otp.generate(start_time + 29).token, token) + + # and new one at 30s + self.assertNotEqual(otp.generate(start_time + 30).token, token) + + # verify round-trip conversion of datetime + dt = datetime.datetime.utcfromtimestamp(time) + self.assertEqual(int(otp.normalize_time(dt)), int(time)) + + # handle datetime object + self.assertEqual(otp.generate(dt).token, token) + + # omitting value should use current time + otp2 = TOTP.using(now=lambda: time)(key=otp.base32_key) + self.assertEqual(otp2.generate().token, token) + + # reject invalid time + self.assertRaises(ValueError, otp.generate, -1) + + def test_generate_w_reference_vectors(self): + """generate() -- reference vectors""" + for otp, time, token, expires, prefix in self.iter_test_vectors(): + # should output correct token for specified time + result = otp.generate(time) + self.assertEqual(result.token, token, msg=prefix) + self.assertEqual(result.counter, time // otp.period, msg=prefix) + if expires: + self.assertEqual(result.expire_time, expires) + + #============================================================================= + # TotpMatch() tests + #============================================================================= + + def assertTotpMatch(self, match, time, skipped=0, period=30, window=30, msg=''): + from passlib.totp import TotpMatch + + # test type + self.assertIsInstance(match, TotpMatch) + + # totp sanity check + self.assertIsInstance(match.totp, TOTP) + self.assertEqual(match.totp.period, period) + + # test attrs + self.assertEqual(match.time, time, msg=msg + " matched time:") + expected = time // period + counter = expected + skipped + self.assertEqual(match.counter, counter, msg=msg + " matched counter:") + self.assertEqual(match.expected_counter, expected, msg=msg + " expected counter:") + self.assertEqual(match.skipped, skipped, msg=msg + " skipped:") + self.assertEqual(match.cache_seconds, period + window) + expire_time = (counter + 1) * period + self.assertEqual(match.expire_time, expire_time) + self.assertEqual(match.cache_time, expire_time + window) + + # test tuple + self.assertEqual(len(match), 2) + self.assertEqual(match, (counter, time)) + self.assertRaises(IndexError, match.__getitem__, -3) + self.assertEqual(match[0], counter) + self.assertEqual(match[1], time) + self.assertRaises(IndexError, match.__getitem__, 2) + + # test bool + self.assertTrue(match) + + def test_totp_match_w_valid_token(self): + """match() -- valid TotpMatch object""" + time = 141230981 + token = '781501' + otp = TOTP.using(now=lambda: time + 24 * 3600)(KEY3) + result = otp.match(token, time) + self.assertTotpMatch(result, time=time, skipped=0) + + def test_totp_match_w_older_token(self): + """match() -- valid TotpMatch object with future token""" + from passlib.totp import TotpMatch + + time = 141230981 + token = '781501' + otp = TOTP.using(now=lambda: time + 24 * 3600)(KEY3) + result = otp.match(token, time - 30) + self.assertTotpMatch(result, time=time - 30, skipped=1) + + def test_totp_match_w_new_token(self): + """match() -- valid TotpMatch object with past token""" + time = 141230981 + token = '781501' + otp = TOTP.using(now=lambda: time + 24 * 3600)(KEY3) + result = otp.match(token, time + 30) + self.assertTotpMatch(result, time=time + 30, skipped=-1) + + def test_totp_match_w_invalid_token(self): + """match() -- invalid TotpMatch object""" + time = 141230981 + token = '781501' + otp = TOTP.using(now=lambda: time + 24 * 3600)(KEY3) + self.assertRaises(exc.InvalidTokenError, otp.match, token, time + 60) + + #============================================================================= + # match() tests + #============================================================================= + + def assertVerifyMatches(self, expect_skipped, token, time, # * + otp, gen_time=None, **kwds): + """helper to test otp.match() output is correct""" + # NOTE: TotpMatch return type tested more throughly above ^^^ + msg = "key=%r alg=%r period=%r token=%r gen_time=%r time=%r:" % \ + (otp.base32_key, otp.alg, otp.period, token, gen_time, time) + result = otp.match(token, time, **kwds) + self.assertTotpMatch(result, + time=otp.normalize_time(time), + period=otp.period, + window=kwds.get("window", 30), + skipped=expect_skipped, + msg=msg) + + def assertVerifyRaises(self, exc_class, token, time, # * + otp, gen_time=None, + **kwds): + """helper to test otp.match() throws correct error""" + # NOTE: TotpMatch return type tested more throughly above ^^^ + msg = "key=%r alg=%r period=%r token=%r gen_time=%r time=%r:" % \ + (otp.base32_key, otp.alg, otp.period, token, gen_time, time) + return self.assertRaises(exc_class, otp.match, token, time, + __msg__=msg, **kwds) + + def test_match_w_window(self): + """match() -- 'time' and 'window' parameters""" + + # init generator & helper + otp = self.randotp() + period = otp.period + time = self.randtime() + token = otp.generate(time).token + common = dict(otp=otp, gen_time=time) + assertMatches = partial(self.assertVerifyMatches, **common) + assertRaises = partial(self.assertVerifyRaises, **common) + + #------------------------------- + # basic validation, and 'window' parameter + #------------------------------- + + # validate against previous counter (passes if window >= period) + assertRaises(exc.InvalidTokenError, token, time - period, window=0) + assertMatches(+1, token, time - period, window=period) + assertMatches(+1, token, time - period, window=2 * period) + + # validate against current counter + assertMatches(0, token, time, window=0) + + # validate against next counter (passes if window >= period) + assertRaises(exc.InvalidTokenError, token, time + period, window=0) + assertMatches(-1, token, time + period, window=period) + assertMatches(-1, token, time + period, window=2 * period) + + # validate against two time steps later (should never pass) + assertRaises(exc.InvalidTokenError, token, time + 2 * period, window=0) + assertRaises(exc.InvalidTokenError, token, time + 2 * period, window=period) + assertMatches(-2, token, time + 2 * period, window=2 * period) + + # TODO: test window values that aren't multiples of period + # (esp ensure counter rounding works correctly) + + #------------------------------- + # time normalization + #------------------------------- + + # handle datetimes + dt = datetime.datetime.utcfromtimestamp(time) + assertMatches(0, token, dt, window=0) + + # reject invalid time + assertRaises(ValueError, token, -1) + + def test_match_w_skew(self): + """match() -- 'skew' parameters""" + # init generator & helper + otp = self.randotp() + period = otp.period + time = self.randtime() + common = dict(otp=otp, gen_time=time) + assertMatches = partial(self.assertVerifyMatches, **common) + assertRaises = partial(self.assertVerifyRaises, **common) + + # assume client is running far behind server / has excessive transmission delay + skew = 3 * period + behind_token = otp.generate(time - skew).token + assertRaises(exc.InvalidTokenError, behind_token, time, window=0) + assertMatches(-3, behind_token, time, window=0, skew=-skew) + + # assume client is running far ahead of server + ahead_token = otp.generate(time + skew).token + assertRaises(exc.InvalidTokenError, ahead_token, time, window=0) + assertMatches(+3, ahead_token, time, window=0, skew=skew) + + # TODO: test skew + larger window + + def test_match_w_reuse(self): + """match() -- 'reuse' and 'last_counter' parameters""" + + # init generator & helper + otp = self.randotp() + period = otp.period + time = self.randtime() + tdata = otp.generate(time) + token = tdata.token + counter = tdata.counter + expire_time = tdata.expire_time + common = dict(otp=otp, gen_time=time) + assertMatches = partial(self.assertVerifyMatches, **common) + assertRaises = partial(self.assertVerifyRaises, **common) + + # last counter unset -- + # previous period's token should count as valid + assertMatches(-1, token, time + period, window=period) + + # last counter set 2 periods ago -- + # previous period's token should count as valid + assertMatches(-1, token, time + period, last_counter=counter-1, + window=period) + + # last counter set 2 periods ago -- + # 2 periods ago's token should NOT count as valid + assertRaises(exc.InvalidTokenError, token, time + 2 * period, + last_counter=counter, window=period) + + # last counter set 1 period ago -- + # previous period's token should now be rejected as 'used' + err = assertRaises(exc.UsedTokenError, token, time + period, + last_counter=counter, window=period) + self.assertEqual(err.expire_time, expire_time) + + # last counter set to current period -- + # current period's token should be rejected + err = assertRaises(exc.UsedTokenError, token, time, + last_counter=counter, window=0) + self.assertEqual(err.expire_time, expire_time) + + def test_match_w_token_normalization(self): + """match() -- token normalization""" + # setup test helper + otp = TOTP('otxl2f5cctbprpzx') + match = otp.match + time = 1412889861 + + # separators / spaces should be stripped (orig token '332136') + self.assertTrue(match(' 3 32-136 ', time)) + + # ascii bytes + self.assertTrue(match(b'332136', time)) + + # too few digits + self.assertRaises(exc.MalformedTokenError, match, '12345', time) + + # invalid char + self.assertRaises(exc.MalformedTokenError, match, '12345X', time) + + # leading zeros count towards size + self.assertRaises(exc.MalformedTokenError, match, '0123456', time) + + def test_match_w_reference_vectors(self): + """match() -- reference vectors""" + for otp, time, token, expires, msg in self.iter_test_vectors(): + # create wrapper + match = otp.match + + # token should match against time + result = match(token, time) + self.assertTrue(result) + self.assertEqual(result.counter, time // otp.period, msg=msg) + + # should NOT match against another time + self.assertRaises(exc.InvalidTokenError, match, token, time + 100, window=0) + + #============================================================================= + # verify() tests + #============================================================================= + def test_verify(self): + """verify()""" + # NOTE: since this is thin wrapper around .from_source() and .match(), + # just testing basic behavior here. + + from passlib.totp import TOTP + + time = 1412889861 + TotpFactory = TOTP.using(now=lambda: time) + + # successful match + source1 = dict(v=1, type="totp", key='otxl2f5cctbprpzx') + match = TotpFactory.verify('332136', source1) + self.assertTotpMatch(match, time=time) + + # failed match + source1 = dict(v=1, type="totp", key='otxl2f5cctbprpzx') + self.assertRaises(exc.InvalidTokenError, TotpFactory.verify, '332155', source1) + + # bad source + source1 = dict(v=1, type="totp") + self.assertRaises(ValueError, TotpFactory.verify, '332155', source1) + + # successful match -- json source + source1json = '{"v": 1, "type": "totp", "key": "otxl2f5cctbprpzx"}' + match = TotpFactory.verify('332136', source1json) + self.assertTotpMatch(match, time=time) + + # successful match -- URI + source1uri = 'otpauth://totp/Label?secret=otxl2f5cctbprpzx' + match = TotpFactory.verify('332136', source1uri) + self.assertTotpMatch(match, time=time) + + #============================================================================= + # serialization frontend tests + #============================================================================= + def test_from_source(self): + """from_source()""" + from passlib.totp import TOTP + from_source = TOTP.from_source + + # uri (unicode) + otp = from_source(u("otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&" + "issuer=Example")) + self.assertEqual(otp.key, KEY4_RAW) + + # uri (bytes) + otp = from_source(b"otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&" + b"issuer=Example") + self.assertEqual(otp.key, KEY4_RAW) + + # dict + otp = from_source(dict(v=1, type="totp", key=KEY4)) + self.assertEqual(otp.key, KEY4_RAW) + + # json (unicode) + otp = from_source(u('{"v": 1, "type": "totp", "key": "JBSWY3DPEHPK3PXP"}')) + self.assertEqual(otp.key, KEY4_RAW) + + # json (bytes) + otp = from_source(b'{"v": 1, "type": "totp", "key": "JBSWY3DPEHPK3PXP"}') + self.assertEqual(otp.key, KEY4_RAW) + + # TOTP object -- return unchanged + self.assertIs(from_source(otp), otp) + + # TOTP object w/ different wallet -- return new one. + wallet1 = AppWallet() + otp1 = TOTP.using(wallet=wallet1).from_source(otp) + self.assertIsNot(otp1, otp) + self.assertEqual(otp1.to_dict(), otp.to_dict()) + + # TOTP object w/ same wallet -- return original + otp2 = TOTP.using(wallet=wallet1).from_source(otp1) + self.assertIs(otp2, otp1) + + # random string + self.assertRaises(ValueError, from_source, u("foo")) + self.assertRaises(ValueError, from_source, b"foo") + + #============================================================================= + # uri serialization tests + #============================================================================= + def test_from_uri(self): + """from_uri()""" + from passlib.totp import TOTP + from_uri = TOTP.from_uri + + # URIs from https://code.google.com/p/google-authenticator/wiki/KeyUriFormat + + #-------------------------------------------------------------------------------- + # canonical uri + #-------------------------------------------------------------------------------- + otp = from_uri("otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&" + "issuer=Example") + self.assertIsInstance(otp, TOTP) + self.assertEqual(otp.key, KEY4_RAW) + self.assertEqual(otp.label, "alice@google.com") + self.assertEqual(otp.issuer, "Example") + self.assertEqual(otp.alg, "sha1") # default + self.assertEqual(otp.period, 30) # default + self.assertEqual(otp.digits, 6) # default + + #-------------------------------------------------------------------------------- + # secret param + #-------------------------------------------------------------------------------- + + # secret case insensitive + otp = from_uri("otpauth://totp/Example:alice@google.com?secret=jbswy3dpehpk3pxp&" + "issuer=Example") + self.assertEqual(otp.key, KEY4_RAW) + + # missing secret + self.assertRaises(ValueError, from_uri, "otpauth://totp/Example:alice@google.com?digits=6") + + # undecodable secret + self.assertRaises(Base32DecodeError, from_uri, "otpauth://totp/Example:alice@google.com?" + "secret=JBSWY3DPEHP@3PXP") + + #-------------------------------------------------------------------------------- + # label param + #-------------------------------------------------------------------------------- + + # w/ encoded space + otp = from_uri("otpauth://totp/Provider1:Alice%20Smith?secret=JBSWY3DPEHPK3PXP&" + "issuer=Provider1") + self.assertEqual(otp.label, "Alice Smith") + self.assertEqual(otp.issuer, "Provider1") + + # w/ encoded space and colon + # (note url has leading space before 'alice') -- taken from KeyURI spec + otp = from_uri("otpauth://totp/Big%20Corporation%3A%20alice@bigco.com?" + "secret=JBSWY3DPEHPK3PXP") + self.assertEqual(otp.label, "alice@bigco.com") + self.assertEqual(otp.issuer, "Big Corporation") + + #-------------------------------------------------------------------------------- + # issuer param / prefix + #-------------------------------------------------------------------------------- + + # 'new style' issuer only + otp = from_uri("otpauth://totp/alice@bigco.com?secret=JBSWY3DPEHPK3PXP&issuer=Big%20Corporation") + self.assertEqual(otp.label, "alice@bigco.com") + self.assertEqual(otp.issuer, "Big Corporation") + + # new-vs-old issuer mismatch + self.assertRaises(ValueError, TOTP.from_uri, + "otpauth://totp/Provider1:alice?secret=JBSWY3DPEHPK3PXP&issuer=Provider2") + + #-------------------------------------------------------------------------------- + # algorithm param + #-------------------------------------------------------------------------------- + + # custom alg + otp = from_uri("otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&algorithm=SHA256") + self.assertEqual(otp.alg, "sha256") + + # unknown alg + self.assertRaises(ValueError, from_uri, "otpauth://totp/Example:alice@google.com?" + "secret=JBSWY3DPEHPK3PXP&algorithm=SHA333") + + #-------------------------------------------------------------------------------- + # digit param + #-------------------------------------------------------------------------------- + + # custom digits + otp = from_uri("otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&digits=8") + self.assertEqual(otp.digits, 8) + + # digits out of range / invalid + self.assertRaises(ValueError, from_uri, "otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&digits=A") + self.assertRaises(ValueError, from_uri, "otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&digits=%20") + self.assertRaises(ValueError, from_uri, "otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&digits=15") + + #-------------------------------------------------------------------------------- + # period param + #-------------------------------------------------------------------------------- + + # custom period + otp = from_uri("otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&period=63") + self.assertEqual(otp.period, 63) + + # reject period < 1 + self.assertRaises(ValueError, from_uri, "otpauth://totp/Example:alice@google.com?" + "secret=JBSWY3DPEHPK3PXP&period=0") + + self.assertRaises(ValueError, from_uri, "otpauth://totp/Example:alice@google.com?" + "secret=JBSWY3DPEHPK3PXP&period=-1") + + #-------------------------------------------------------------------------------- + # unrecognized param + #-------------------------------------------------------------------------------- + + # should issue warning, but otherwise ignore extra param + with self.assertWarningList([ + dict(category=exc.PasslibRuntimeWarning, message_re="unexpected parameters encountered") + ]): + otp = from_uri("otpauth://totp/Example:alice@google.com?secret=JBSWY3DPEHPK3PXP&" + "foo=bar&period=63") + self.assertEqual(otp.base32_key, KEY4) + self.assertEqual(otp.period, 63) + + def test_to_uri(self): + """to_uri()""" + + #------------------------------------------------------------------------- + # label & issuer parameters + #------------------------------------------------------------------------- + + # with label & issuer + otp = TOTP(KEY4, alg="sha1", digits=6, period=30) + self.assertEqual(otp.to_uri("alice@google.com", "Example Org"), + "otpauth://totp/Example%20Org:alice@google.com?secret=JBSWY3DPEHPK3PXP&" + "issuer=Example%20Org") + + # label is required + self.assertRaises(ValueError, otp.to_uri, None, "Example Org") + + # with label only + self.assertEqual(otp.to_uri("alice@google.com"), + "otpauth://totp/alice@google.com?secret=JBSWY3DPEHPK3PXP") + + # with default label from constructor + otp.label = "alice@google.com" + self.assertEqual(otp.to_uri(), + "otpauth://totp/alice@google.com?secret=JBSWY3DPEHPK3PXP") + + # with default label & default issuer from constructor + otp.issuer = "Example Org" + self.assertEqual(otp.to_uri(), + "otpauth://totp/Example%20Org:alice@google.com?secret=JBSWY3DPEHPK3PXP" + "&issuer=Example%20Org") + + # reject invalid label + self.assertRaises(ValueError, otp.to_uri, "label:with:semicolons") + + # reject invalid issuer + self.assertRaises(ValueError, otp.to_uri, "alice@google.com", "issuer:with:semicolons") + + #------------------------------------------------------------------------- + # algorithm parameter + #------------------------------------------------------------------------- + self.assertEqual(TOTP(KEY4, alg="sha256").to_uri("alice@google.com"), + "otpauth://totp/alice@google.com?secret=JBSWY3DPEHPK3PXP&" + "algorithm=SHA256") + + #------------------------------------------------------------------------- + # digits parameter + #------------------------------------------------------------------------- + self.assertEqual(TOTP(KEY4, digits=8).to_uri("alice@google.com"), + "otpauth://totp/alice@google.com?secret=JBSWY3DPEHPK3PXP&" + "digits=8") + + #------------------------------------------------------------------------- + # period parameter + #------------------------------------------------------------------------- + self.assertEqual(TOTP(KEY4, period=63).to_uri("alice@google.com"), + "otpauth://totp/alice@google.com?secret=JBSWY3DPEHPK3PXP&" + "period=63") + + #============================================================================= + # dict serialization tests + #============================================================================= + def test_from_dict(self): + """from_dict()""" + from passlib.totp import TOTP + from_dict = TOTP.from_dict + + #-------------------------------------------------------------------------------- + # canonical simple example + #-------------------------------------------------------------------------------- + otp = from_dict(dict(v=1, type="totp", key=KEY4, label="alice@google.com", issuer="Example")) + self.assertIsInstance(otp, TOTP) + self.assertEqual(otp.key, KEY4_RAW) + self.assertEqual(otp.label, "alice@google.com") + self.assertEqual(otp.issuer, "Example") + self.assertEqual(otp.alg, "sha1") # default + self.assertEqual(otp.period, 30) # default + self.assertEqual(otp.digits, 6) # default + + #-------------------------------------------------------------------------------- + # metadata + #-------------------------------------------------------------------------------- + + # missing version + self.assertRaises(ValueError, from_dict, dict(type="totp", key=KEY4)) + + # invalid version + self.assertRaises(ValueError, from_dict, dict(v=0, type="totp", key=KEY4)) + self.assertRaises(ValueError, from_dict, dict(v=999, type="totp", key=KEY4)) + + # missing type + self.assertRaises(ValueError, from_dict, dict(v=1, key=KEY4)) + + #-------------------------------------------------------------------------------- + # secret param + #-------------------------------------------------------------------------------- + + # secret case insensitive + otp = from_dict(dict(v=1, type="totp", key=KEY4.lower(), label="alice@google.com", issuer="Example")) + self.assertEqual(otp.key, KEY4_RAW) + + # missing secret + self.assertRaises(ValueError, from_dict, dict(v=1, type="totp")) + + # undecodable secret + self.assertRaises(Base32DecodeError, from_dict, + dict(v=1, type="totp", key="JBSWY3DPEHP@3PXP")) + + #-------------------------------------------------------------------------------- + # label & issuer params + #-------------------------------------------------------------------------------- + + otp = from_dict(dict(v=1, type="totp", key=KEY4, label="Alice Smith", issuer="Provider1")) + self.assertEqual(otp.label, "Alice Smith") + self.assertEqual(otp.issuer, "Provider1") + + #-------------------------------------------------------------------------------- + # algorithm param + #-------------------------------------------------------------------------------- + + # custom alg + otp = from_dict(dict(v=1, type="totp", key=KEY4, alg="sha256")) + self.assertEqual(otp.alg, "sha256") + + # unknown alg + self.assertRaises(ValueError, from_dict, dict(v=1, type="totp", key=KEY4, alg="sha333")) + + #-------------------------------------------------------------------------------- + # digit param + #-------------------------------------------------------------------------------- + + # custom digits + otp = from_dict(dict(v=1, type="totp", key=KEY4, digits=8)) + self.assertEqual(otp.digits, 8) + + # digits out of range / invalid + self.assertRaises(TypeError, from_dict, dict(v=1, type="totp", key=KEY4, digits="A")) + self.assertRaises(ValueError, from_dict, dict(v=1, type="totp", key=KEY4, digits=15)) + + #-------------------------------------------------------------------------------- + # period param + #-------------------------------------------------------------------------------- + + # custom period + otp = from_dict(dict(v=1, type="totp", key=KEY4, period=63)) + self.assertEqual(otp.period, 63) + + # reject period < 1 + self.assertRaises(ValueError, from_dict, dict(v=1, type="totp", key=KEY4, period=0)) + self.assertRaises(ValueError, from_dict, dict(v=1, type="totp", key=KEY4, period=-1)) + + #-------------------------------------------------------------------------------- + # unrecognized param + #-------------------------------------------------------------------------------- + self.assertRaises(TypeError, from_dict, dict(v=1, type="totp", key=KEY4, INVALID=123)) + + def test_to_dict(self): + """to_dict()""" + + #------------------------------------------------------------------------- + # label & issuer parameters + #------------------------------------------------------------------------- + + # without label or issuer + otp = TOTP(KEY4, alg="sha1", digits=6, period=30) + self.assertEqual(otp.to_dict(), dict(v=1, type="totp", key=KEY4)) + + # with label & issuer from constructor + otp = TOTP(KEY4, alg="sha1", digits=6, period=30, + label="alice@google.com", issuer="Example Org") + self.assertEqual(otp.to_dict(), + dict(v=1, type="totp", key=KEY4, + label="alice@google.com", issuer="Example Org")) + + # with label only + otp = TOTP(KEY4, alg="sha1", digits=6, period=30, + label="alice@google.com") + self.assertEqual(otp.to_dict(), + dict(v=1, type="totp", key=KEY4, + label="alice@google.com")) + + # with issuer only + otp = TOTP(KEY4, alg="sha1", digits=6, period=30, + issuer="Example Org") + self.assertEqual(otp.to_dict(), + dict(v=1, type="totp", key=KEY4, + issuer="Example Org")) + + # don't serialize default issuer + TotpFactory = TOTP.using(issuer="Example Org") + otp = TotpFactory(KEY4) + self.assertEqual(otp.to_dict(), dict(v=1, type="totp", key=KEY4)) + + # don't serialize default issuer *even if explicitly set* + otp = TotpFactory(KEY4, issuer="Example Org") + self.assertEqual(otp.to_dict(), dict(v=1, type="totp", key=KEY4)) + + #------------------------------------------------------------------------- + # algorithm parameter + #------------------------------------------------------------------------- + self.assertEqual(TOTP(KEY4, alg="sha256").to_dict(), + dict(v=1, type="totp", key=KEY4, alg="sha256")) + + #------------------------------------------------------------------------- + # digits parameter + #------------------------------------------------------------------------- + self.assertEqual(TOTP(KEY4, digits=8).to_dict(), + dict(v=1, type="totp", key=KEY4, digits=8)) + + #------------------------------------------------------------------------- + # period parameter + #------------------------------------------------------------------------- + self.assertEqual(TOTP(KEY4, period=63).to_dict(), + dict(v=1, type="totp", key=KEY4, period=63)) + + # TODO: to_dict() + # with encrypt=False + # with encrypt="auto" + wallet + secrets + # with encrypt="auto" + wallet + no secrets + # with encrypt="auto" + no wallet + # with encrypt=True + wallet + secrets + # with encrypt=True + wallet + no secrets + # with encrypt=True + no wallet + # that 'changed' is set for old versions, and old encryption tags. + + #============================================================================= + # json serialization tests + #============================================================================= + + # TODO: from_json() / to_json(). + # (skipped for right now cause just wrapper for from_dict/to_dict) + + #============================================================================= + # eoc + #============================================================================= + +#============================================================================= +# eof +#============================================================================= diff --git a/venv/Lib/site-packages/passlib/tests/test_utils.py b/venv/Lib/site-packages/passlib/tests/test_utils.py new file mode 100644 index 0000000..59ba160 --- /dev/null +++ b/venv/Lib/site-packages/passlib/tests/test_utils.py @@ -0,0 +1,1171 @@ +"""tests for passlib.util""" +#============================================================================= +# imports +#============================================================================= +from __future__ import with_statement +# core +from functools import partial +import warnings +# site +# pkg +# module +from passlib.utils import is_ascii_safe, to_bytes +from passlib.utils.compat import irange, PY2, PY3, u, unicode, join_bytes, PYPY +from passlib.tests.utils import TestCase, hb, run_with_fixed_seeds + +#============================================================================= +# byte funcs +#============================================================================= +class MiscTest(TestCase): + """tests various parts of utils module""" + + # NOTE: could test xor_bytes(), but it's exercised well enough by pbkdf2 test + + def test_compat(self): + """test compat's lazymodule""" + from passlib.utils import compat + # "" + self.assertRegex(repr(compat), + r"^$") + + # test synthentic dir() + dir(compat) + self.assertTrue('UnicodeIO' in dir(compat)) + self.assertTrue('irange' in dir(compat)) + + def test_classproperty(self): + from passlib.utils.decor import classproperty + + class test(object): + xvar = 1 + @classproperty + def xprop(cls): + return cls.xvar + + self.assertEqual(test.xprop, 1) + prop = test.__dict__['xprop'] + self.assertIs(prop.im_func, prop.__func__) + + def test_deprecated_function(self): + from passlib.utils.decor import deprecated_function + # NOTE: not comprehensive, just tests the basic behavior + + @deprecated_function(deprecated="1.6", removed="1.8") + def test_func(*args): + """test docstring""" + return args + + self.assertTrue(".. deprecated::" in test_func.__doc__) + + with self.assertWarningList(dict(category=DeprecationWarning, + message="the function passlib.tests.test_utils.test_func() " + "is deprecated as of Passlib 1.6, and will be " + "removed in Passlib 1.8." + )): + self.assertEqual(test_func(1,2), (1,2)) + + def test_memoized_property(self): + from passlib.utils.decor import memoized_property + + class dummy(object): + counter = 0 + + @memoized_property + def value(self): + value = self.counter + self.counter = value+1 + return value + + d = dummy() + self.assertEqual(d.value, 0) + self.assertEqual(d.value, 0) + self.assertEqual(d.counter, 1) + + prop = dummy.value + if not PY3: + self.assertIs(prop.im_func, prop.__func__) + + def test_getrandbytes(self): + """getrandbytes()""" + from passlib.utils import getrandbytes + wrapper = partial(getrandbytes, self.getRandom()) + self.assertEqual(len(wrapper(0)), 0) + a = wrapper(10) + b = wrapper(10) + self.assertIsInstance(a, bytes) + self.assertEqual(len(a), 10) + self.assertEqual(len(b), 10) + self.assertNotEqual(a, b) + + @run_with_fixed_seeds(count=1024) + def test_getrandstr(self, seed): + """getrandstr()""" + from passlib.utils import getrandstr + + wrapper = partial(getrandstr, self.getRandom(seed=seed)) + + # count 0 + self.assertEqual(wrapper('abc',0), '') + + # count <0 + self.assertRaises(ValueError, wrapper, 'abc', -1) + + # letters 0 + self.assertRaises(ValueError, wrapper, '', 0) + + # letters 1 + self.assertEqual(wrapper('a', 5), 'aaaaa') + + # NOTE: the following parts are non-deterministic, + # with a small chance of failure (outside chance it may pick + # a string w/o one char, even more remote chance of picking + # same string). to combat this, we run it against multiple + # fixed seeds (using run_with_fixed_seeds decorator), + # and hope that they're sufficient to test the range of behavior. + + # letters + x = wrapper(u('abc'), 32) + y = wrapper(u('abc'), 32) + self.assertIsInstance(x, unicode) + self.assertNotEqual(x,y) + self.assertEqual(sorted(set(x)), [u('a'),u('b'),u('c')]) + + # bytes + x = wrapper(b'abc', 32) + y = wrapper(b'abc', 32) + self.assertIsInstance(x, bytes) + self.assertNotEqual(x,y) + # NOTE: decoding this due to py3 bytes + self.assertEqual(sorted(set(x.decode("ascii"))), [u('a'),u('b'),u('c')]) + + def test_generate_password(self): + """generate_password()""" + from passlib.utils import generate_password + warnings.filterwarnings("ignore", "The function.*generate_password\(\) is deprecated") + self.assertEqual(len(generate_password(15)), 15) + + def test_is_crypt_context(self): + """test is_crypt_context()""" + from passlib.utils import is_crypt_context + from passlib.context import CryptContext + cc = CryptContext(["des_crypt"]) + self.assertTrue(is_crypt_context(cc)) + self.assertFalse(not is_crypt_context(cc)) + + def test_genseed(self): + """test genseed()""" + import random + from passlib.utils import genseed + rng = random.Random(genseed()) + a = rng.randint(0, 10**10) + + rng = random.Random(genseed()) + b = rng.randint(0, 10**10) + + self.assertNotEqual(a,b) + + rng.seed(genseed(rng)) + + def test_crypt(self): + """test crypt.crypt() wrappers""" + from passlib.utils import has_crypt, safe_crypt, test_crypt + from passlib.registry import get_supported_os_crypt_schemes, get_crypt_handler + + # test everything is disabled + supported = get_supported_os_crypt_schemes() + if not has_crypt: + self.assertEqual(supported, ()) + self.assertEqual(safe_crypt("test", "aa"), None) + self.assertFalse(test_crypt("test", "aaqPiZY5xR5l.")) # des_crypt() hash of "test" + raise self.skipTest("crypt.crypt() not available") + + # expect there to be something supported, if crypt() is present + if not supported: + # NOTE: failures here should be investigated. usually means one of: + # 1) at least one of passlib's os_crypt detection routines is giving false negative + # 2) crypt() ONLY supports some hash alg which passlib doesn't know about + # 3) crypt() is present but completely disabled (never encountered this yet) + raise self.fail("crypt() present, but no supported schemes found!") + + # pick cheap alg if possible, with minimum rounds, to speed up this test. + # NOTE: trusting hasher class works properly (should have been verified using it's own UTs) + for scheme in ("md5_crypt", "sha256_crypt"): + if scheme in supported: + break + else: + scheme = supported[-1] + hasher = get_crypt_handler(scheme) + if getattr(hasher, "min_rounds", None): + hasher = hasher.using(rounds=hasher.min_rounds) + + # helpers to generate hashes & config strings to work with + def get_hash(secret): + assert isinstance(secret, unicode) + hash = hasher.hash(secret) + if isinstance(hash, bytes): # py2 + hash = hash.decode("utf-8") + assert isinstance(hash, unicode) + return hash + + # test ascii password & return type + s1 = u("test") + h1 = get_hash(s1) + result = safe_crypt(s1, h1) + self.assertIsInstance(result, unicode) + self.assertEqual(result, h1) + self.assertEqual(safe_crypt(to_bytes(s1), to_bytes(h1)), h1) + + # make sure crypt doesn't just blindly return h1 for whatever we pass in + h1x = h1[:-2] + 'xx' + self.assertEqual(safe_crypt(s1, h1x), h1) + + # test utf-8 / unicode password + s2 = u('test\u1234') + h2 = get_hash(s2) + self.assertEqual(safe_crypt(s2, h2), h2) + self.assertEqual(safe_crypt(to_bytes(s2), to_bytes(h2)), h2) + + # test rejects null chars in password + self.assertRaises(ValueError, safe_crypt, '\x00', h1) + + # check test_crypt() + self.assertTrue(test_crypt("test", h1)) + self.assertFalse(test_crypt("test", h1x)) + + # check crypt returning variant error indicators + # some platforms return None on errors, others empty string, + # The BSDs in some cases return ":" + import passlib.utils as mod + orig = mod._crypt + try: + retval = None + mod._crypt = lambda secret, hash: retval + + for retval in [None, "", ":", ":0", "*0"]: + self.assertEqual(safe_crypt("test", h1), None) + self.assertFalse(test_crypt("test", h1)) + + retval = 'xxx' + self.assertEqual(safe_crypt("test", h1), "xxx") + self.assertFalse(test_crypt("test", h1)) + + finally: + mod._crypt = orig + + def test_consteq(self): + """test consteq()""" + # NOTE: this test is kind of over the top, but that's only because + # this is used for the critical task of comparing hashes for equality. + from passlib.utils import consteq, str_consteq + + # ensure error raises for wrong types + self.assertRaises(TypeError, consteq, u(''), b'') + self.assertRaises(TypeError, consteq, u(''), 1) + self.assertRaises(TypeError, consteq, u(''), None) + + self.assertRaises(TypeError, consteq, b'', u('')) + self.assertRaises(TypeError, consteq, b'', 1) + self.assertRaises(TypeError, consteq, b'', None) + + self.assertRaises(TypeError, consteq, None, u('')) + self.assertRaises(TypeError, consteq, None, b'') + self.assertRaises(TypeError, consteq, 1, u('')) + self.assertRaises(TypeError, consteq, 1, b'') + + def consteq_supports_string(value): + # under PY2, it supports all unicode strings (when present at all), + # under PY3, compare_digest() only supports ascii unicode strings. + # confirmed for: cpython 2.7.9, cpython 3.4, pypy, pypy3, pyston + return (consteq is str_consteq or PY2 or is_ascii_safe(value)) + + # check equal inputs compare correctly + for value in [ + u("a"), + u("abc"), + u("\xff\xa2\x12\x00")*10, + ]: + if consteq_supports_string(value): + self.assertTrue(consteq(value, value), "value %r:" % (value,)) + else: + self.assertRaises(TypeError, consteq, value, value) + self.assertTrue(str_consteq(value, value), "value %r:" % (value,)) + + value = value.encode("latin-1") + self.assertTrue(consteq(value, value), "value %r:" % (value,)) + + # check non-equal inputs compare correctly + for l,r in [ + # check same-size comparisons with differing contents fail. + (u("a"), u("c")), + (u("abcabc"), u("zbaabc")), + (u("abcabc"), u("abzabc")), + (u("abcabc"), u("abcabz")), + ((u("\xff\xa2\x12\x00")*10)[:-1] + u("\x01"), + u("\xff\xa2\x12\x00")*10), + + # check different-size comparisons fail. + (u(""), u("a")), + (u("abc"), u("abcdef")), + (u("abc"), u("defabc")), + (u("qwertyuiopasdfghjklzxcvbnm"), u("abc")), + ]: + if consteq_supports_string(l) and consteq_supports_string(r): + self.assertFalse(consteq(l, r), "values %r %r:" % (l,r)) + self.assertFalse(consteq(r, l), "values %r %r:" % (r,l)) + else: + self.assertRaises(TypeError, consteq, l, r) + self.assertRaises(TypeError, consteq, r, l) + self.assertFalse(str_consteq(l, r), "values %r %r:" % (l,r)) + self.assertFalse(str_consteq(r, l), "values %r %r:" % (r,l)) + + l = l.encode("latin-1") + r = r.encode("latin-1") + self.assertFalse(consteq(l, r), "values %r %r:" % (l,r)) + self.assertFalse(consteq(r, l), "values %r %r:" % (r,l)) + + # TODO: add some tests to ensure we take THETA(strlen) time. + # this might be hard to do reproducably. + # NOTE: below code was used to generate stats for analysis + ##from math import log as logb + ##import timeit + ##multipliers = [ 1< encode() -> decode() -> raw + # + + # generate some random bytes + size = rng.randint(1 if saw_zero else 0, 12) + if not size: + saw_zero = True + enc_size = (4*size+2)//3 + raw = getrandbytes(rng, size) + + # encode them, check invariants + encoded = engine.encode_bytes(raw) + self.assertEqual(len(encoded), enc_size) + + # make sure decode returns original + result = engine.decode_bytes(encoded) + self.assertEqual(result, raw) + + # + # test encoded -> decode() -> encode() -> encoded + # + + # generate some random encoded data + if size % 4 == 1: + size += rng.choice([-1,1,2]) + raw_size = 3*size//4 + encoded = getrandstr(rng, engine.bytemap, size) + + # decode them, check invariants + raw = engine.decode_bytes(encoded) + self.assertEqual(len(raw), raw_size, "encoded %d:" % size) + + # make sure encode returns original (barring padding bits) + result = engine.encode_bytes(raw) + if size % 4: + self.assertEqual(result[:-1], encoded[:-1]) + else: + self.assertEqual(result, encoded) + + def test_repair_unused(self): + """test repair_unused()""" + # NOTE: this test relies on encode_bytes() always returning clear + # padding bits - which should be ensured by test vectors. + from passlib.utils import getrandstr + rng = self.getRandom() + engine = self.engine + check_repair_unused = self.engine.check_repair_unused + i = 0 + while i < 300: + size = rng.randint(0,23) + cdata = getrandstr(rng, engine.charmap, size).encode("ascii") + if size & 3 == 1: + # should throw error + self.assertRaises(ValueError, check_repair_unused, cdata) + continue + rdata = engine.encode_bytes(engine.decode_bytes(cdata)) + if rng.random() < .5: + cdata = cdata.decode("ascii") + rdata = rdata.decode("ascii") + if cdata == rdata: + # should leave unchanged + ok, result = check_repair_unused(cdata) + self.assertFalse(ok) + self.assertEqual(result, rdata) + else: + # should repair bits + self.assertNotEqual(size % 4, 0) + ok, result = check_repair_unused(cdata) + self.assertTrue(ok) + self.assertEqual(result, rdata) + i += 1 + + #=================================================================== + # test transposed encode/decode - encoding independant + #=================================================================== + # NOTE: these tests assume normal encode/decode has been tested elsewhere. + + transposed = [ + # orig, result, transpose map + (b"\x33\x22\x11", b"\x11\x22\x33",[2,1,0]), + (b"\x22\x33\x11", b"\x11\x22\x33",[1,2,0]), + ] + + transposed_dups = [ + # orig, result, transpose projection + (b"\x11\x11\x22", b"\x11\x22\x33",[0,0,1]), + ] + + def test_encode_transposed_bytes(self): + """test encode_transposed_bytes()""" + engine = self.engine + for result, input, offsets in self.transposed + self.transposed_dups: + tmp = engine.encode_transposed_bytes(input, offsets) + out = engine.decode_bytes(tmp) + self.assertEqual(out, result) + + self.assertRaises(TypeError, engine.encode_transposed_bytes, u("a"), []) + + def test_decode_transposed_bytes(self): + """test decode_transposed_bytes()""" + engine = self.engine + for input, result, offsets in self.transposed: + tmp = engine.encode_bytes(input) + out = engine.decode_transposed_bytes(tmp, offsets) + self.assertEqual(out, result) + + def test_decode_transposed_bytes_bad(self): + """test decode_transposed_bytes() fails if map is a one-way""" + engine = self.engine + for input, _, offsets in self.transposed_dups: + tmp = engine.encode_bytes(input) + self.assertRaises(TypeError, engine.decode_transposed_bytes, tmp, + offsets) + + #=================================================================== + # test 6bit handling + #=================================================================== + def check_int_pair(self, bits, encoded_pairs): + """helper to check encode_intXX & decode_intXX functions""" + rng = self.getRandom() + engine = self.engine + encode = getattr(engine, "encode_int%s" % bits) + decode = getattr(engine, "decode_int%s" % bits) + pad = -bits % 6 + chars = (bits+pad)//6 + upper = 1<