codekingpro/portable-devtools
114k
1"""passlib.totp -- TOTP / RFC6238 / Google Authenticator utilities."""2#=============================================================================3# imports4#=============================================================================5from __future__ import absolute_import, division, print_function6from passlib.utils.compat import PY37# core8import base649import calendar10import json11import logging; log = logging.getLogger(__name__)12import math13import struct14import sys15import time as _time16import re17if PY3:18 from urllib.parse import urlparse, parse_qsl, quote, unquote19else:20 from urllib import quote, unquote21 from urlparse import urlparse, parse_qsl22from warnings import warn23# site24try:25 # TOTP encrypted keys only supported if cryptography (https://cryptography.io) is installed26 from cryptography.hazmat.backends import default_backend as _cg_default_backend27 import cryptography.hazmat.primitives.ciphers.algorithms28 import cryptography.hazmat.primitives.ciphers.modes29 from cryptography.hazmat.primitives import ciphers as _cg_ciphers30 del cryptography31except ImportError:32 log.debug("can't import 'cryptography' package, totp encryption disabled")33 _cg_ciphers = _cg_default_backend = None34# pkg35from passlib import exc36from passlib.exc import TokenError, MalformedTokenError, InvalidTokenError, UsedTokenError37from passlib.utils import (to_unicode, to_bytes, consteq,38 getrandbytes, rng, SequenceMixin, xor_bytes, getrandstr)39from passlib.utils.binary import BASE64_CHARS, b32encode, b32decode40from passlib.utils.compat import (u, unicode, native_string_types, bascii_to_str, int_types, num_types,41 irange, byte_elem_value, UnicodeIO, suppress_cause)42from passlib.utils.decor import hybrid_method, memoized_property43from passlib.crypto.digest import lookup_hash, compile_hmac, pbkdf2_hmac44from passlib.hash import pbkdf2_sha25645# local46__all__ = [47 # frontend classes48 "AppWallet",49 "TOTP",50 51 # errors (defined in passlib.exc, but exposed here for convenience)52 "TokenError",53 "MalformedTokenError",54 "InvalidTokenError",55 "UsedTokenError",56 57 # internal helper classes58 "TotpToken",59 "TotpMatch",60]61 62#=============================================================================63# HACK: python < 2.7.4's urlparse() won't parse query strings unless the url scheme64# is one of the schemes in the urlparse.uses_query list. 2.7 abandoned65# this, and parses query if present, regardless of the scheme.66# as a workaround for older versions, we add "otpauth" to the known list.67# this was fixed by https://bugs.python.org/issue9374, in 2.7.4 release.68#=============================================================================69if sys.version_info < (2,7,4):70 from urlparse import uses_query71 if "otpauth" not in uses_query:72 uses_query.append("otpauth")73 log.debug("registered 'otpauth' scheme with urlparse.uses_query")74 del uses_query75 76#=============================================================================77# internal helpers78#=============================================================================79 80#-----------------------------------------------------------------------------81# token parsing / rendering helpers82#-----------------------------------------------------------------------------83 84#: regex used to clean whitespace from tokens & keys85_clean_re = re.compile(u(r"\s|[-=]"), re.U)86 87_chunk_sizes = [4,6,5]88 89def _get_group_size(klen):90 """91 helper for group_string() --92 calculates optimal size of group for given string size.93 """94 # look for exact divisor95 for size in _chunk_sizes:96 if not klen % size:97 return size98 # fallback to divisor with largest remainder99 # (so chunks are as close to even as possible)100 best = _chunk_sizes[0]101 rem = 0102 for size in _chunk_sizes:103 if klen % size > rem:104 best = size105 rem = klen % size106 return best107 108def group_string(value, sep="-"):109 """110 reformat string into (roughly) evenly-sized groups, separated by **sep**.111 useful for making tokens & keys easier to read by humans.112 """113 klen = len(value)114 size = _get_group_size(klen)115 return sep.join(value[o:o+size] for o in irange(0, klen, size))116 117#-----------------------------------------------------------------------------118# encoding helpers119#-----------------------------------------------------------------------------120 121def _decode_bytes(key, format):122 """123 internal TOTP() helper --124 decodes key according to specified format.125 """126 if format == "raw":127 if not isinstance(key, bytes):128 raise exc.ExpectedTypeError(key, "bytes", "key")129 return key130 # for encoded data, key must be either unicode or ascii-encoded bytes,131 # and must contain a hex or base32 string.132 key = to_unicode(key, param="key")133 key = _clean_re.sub("", key).encode("utf-8") # strip whitespace & hypens134 if format == "hex" or format == "base16":135 return base64.b16decode(key.upper())136 elif format == "base32":137 return b32decode(key)138 # XXX: add base64 support?139 else:140 raise ValueError("unknown byte-encoding format: %r" % (format,))141 142#=============================================================================143# OTP management144#=============================================================================145 146#: flag for detecting if encrypted totp support is present147AES_SUPPORT = bool(_cg_ciphers)148 149#: regex for validating secret tags150_tag_re = re.compile("(?i)^[a-z0-9][a-z0-9_.-]*$")151 152class AppWallet(object):153 """154 This class stores application-wide secrets that can be used155 to encrypt & decrypt TOTP keys for storage.156 It's mostly an internal detail, applications usually just need157 to pass ``secrets`` or ``secrets_path`` to :meth:`TOTP.using`.158 159 .. seealso::160 161 :ref:`totp-storing-instances` for more details on this workflow.162 163 Arguments164 =========165 :param secrets:166 Dict of application secrets to use when encrypting/decrypting167 stored TOTP keys. This should include a secret to use when encrypting168 new keys, but may contain additional older secrets to decrypt169 existing stored keys.170 171 The dict should map tags -> secrets, so that each secret is identified172 by a unique tag. This tag will be stored along with the encrypted173 key in order to determine which secret should be used for decryption.174 Tag should be string that starts with regex range ``[a-z0-9]``,175 and the remaining characters must be in ``[a-z0-9_.-]``.176 177 It is recommended to use something like a incremental counter178 ("1", "2", ...), an ISO date ("2016-01-01", "2016-05-16", ...), 179 or a timestamp ("19803495", "19813495", ...) when assigning tags.180 181 This mapping be provided in three formats:182 183 * A python dict mapping tag -> secret184 * A JSON-formatted string containing the dict185 * A multiline string with the format ``"tag: value\\ntag: value\\n..."``186 187 (This last format is mainly useful when loading from a text file via **secrets_path**)188 189 .. seealso:: :func:`generate_secret` to create a secret with sufficient entropy190 191 :param secrets_path:192 Alternately, callers can specify a separate file where the193 application-wide secrets are stored, using either of the string194 formats described in **secrets**.195 196 :param default_tag:197 Specifies which tag in **secrets** should be used as the default198 for encrypting new keys. If omitted, the tags will be sorted,199 and the largest tag used as the default.200 201 if all tags are numeric, they will be sorted numerically;202 otherwise they will be sorted alphabetically.203 this permits tags to be assigned numerically,204 or e.g. using ``YYYY-MM-DD`` dates.205 206 :param encrypt_cost:207 Optional time-cost factor for key encryption.208 This value corresponds to log2() of the number of PBKDF2209 rounds used.210 211 .. warning::212 213 The application secret(s) should be stored in a secure location by214 your application, and each secret should contain a large amount215 of entropy (to prevent brute-force attacks if the encrypted keys216 are leaked).217 218 :func:`generate_secret` is provided as a convenience helper219 to generate a new application secret of suitable size.220 221 Best practice is to load these values from a file via **secrets_path**,222 and then have your application give up permission to read this file223 once it's running.224 225 Public Methods226 ==============227 .. autoattribute:: has_secrets228 .. autoattribute:: default_tag229 230 Semi-Private Methods231 ====================232 The following methods are used internally by the :class:`TOTP`233 class in order to encrypt & decrypt keys using the provided application234 secrets. They will generally not be publically useful, and may have their235 API changed periodically.236 237 .. automethod:: get_secret238 .. automethod:: encrypt_key239 .. automethod:: decrypt_key240 """241 #========================================================================242 # instance attrs243 #========================================================================244 245 #: default salt size for encrypt_key() output246 salt_size = 12247 248 #: default cost (log2 of pbkdf2 rounds) for encrypt_key() output249 #: NOTE: this is relatively low, since the majority of the security250 #: relies on a high entropy secret to pass to AES.251 encrypt_cost = 14252 253 #: map of secret tag -> secret bytes254 _secrets = None255 256 #: tag for default secret257 default_tag = None258 259 #========================================================================260 # init261 #========================================================================262 def __init__(self, secrets=None, default_tag=None, encrypt_cost=None,263 secrets_path=None):264 265 # TODO: allow a lot more things to be customized from here,266 # e.g. setting default TOTP constructor options.267 268 #269 # init cost270 #271 if encrypt_cost is not None:272 if isinstance(encrypt_cost, native_string_types):273 encrypt_cost = int(encrypt_cost)274 assert encrypt_cost >= 0275 self.encrypt_cost = encrypt_cost276 277 #278 # init secrets map279 #280 281 # load secrets from file (if needed)282 if secrets_path is not None:283 if secrets is not None:284 raise TypeError("'secrets' and 'secrets_path' are mutually exclusive")285 secrets = open(secrets_path, "rt").read()286 287 # parse & store secrets288 secrets = self._secrets = self._parse_secrets(secrets)289 290 #291 # init default tag/secret292 #293 if secrets:294 if default_tag is not None:295 # verify that tag is present in map296 self.get_secret(default_tag)297 elif all(tag.isdigit() for tag in secrets):298 default_tag = max(secrets, key=int)299 else:300 default_tag = max(secrets)301 self.default_tag = default_tag302 303 def _parse_secrets(self, source):304 """305 parse 'secrets' parameter306 307 :returns:308 Dict[tag:str, secret:bytes]309 """310 # parse string formats311 # to make this easy to pass in configuration from a separate file,312 # 'secrets' can be string using two formats -- json & "tag:value\n"313 check_type = True314 if isinstance(source, native_string_types):315 if source.lstrip().startswith(("[", "{")):316 # json list / dict317 source = json.loads(source)318 elif "\n" in source and ":" in source:319 # multiline string containing series of "tag: value\n" rows;320 # empty and "#\n" rows are ignored321 def iter_pairs(source):322 for line in source.splitlines():323 line = line.strip()324 if line and not line.startswith("#"):325 tag, secret = line.split(":", 1)326 yield tag.strip(), secret.strip()327 source = iter_pairs(source)328 check_type = False329 else:330 raise ValueError("unrecognized secrets string format")331 332 # ensure we have iterable of (tag, value) pairs333 if source is None:334 return {}335 elif isinstance(source, dict):336 source = source.items()337 # XXX: could support iterable of (tag,value) pairs, but not yet needed...338 # elif check_type and (isinstance(source, str) or not isinstance(source, Iterable)):339 elif check_type:340 raise TypeError("'secrets' must be mapping, or list of items")341 342 # parse into final dict, normalizing contents343 return dict(self._parse_secret_pair(tag, value)344 for tag, value in source)345 346 def _parse_secret_pair(self, tag, value):347 if isinstance(tag, native_string_types):348 pass349 elif isinstance(tag, int):350 tag = str(tag)351 else:352 raise TypeError("tag must be unicode/string: %r" % (tag,))353 if not _tag_re.match(tag):354 raise ValueError("tag contains invalid characters: %r" % (tag,))355 if not isinstance(value, bytes):356 value = to_bytes(value, param="secret %r" % (tag,))357 if not value:358 raise ValueError("tag contains empty secret: %r" % (tag,))359 return tag, value360 361 #========================================================================362 # accessing secrets363 #========================================================================364 365 @property366 def has_secrets(self):367 """whether at least one application secret is present"""368 return self.default_tag is not None369 370 def get_secret(self, tag):371 """372 resolve a secret tag to the secret (as bytes).373 throws a KeyError if not found.374 """375 secrets = self._secrets376 if not secrets:377 raise KeyError("no application secrets configured")378 try:379 return secrets[tag]380 except KeyError:381 raise suppress_cause(KeyError("unknown secret tag: %r" % (tag,)))382 383 #========================================================================384 # encrypted key helpers -- used internally by TOTP385 #========================================================================386 387 @staticmethod388 def _cipher_aes_key(value, secret, salt, cost, decrypt=False):389 """390 Internal helper for :meth:`encrypt_key` --391 handles lowlevel encryption/decryption.392 393 Algorithm details:394 395 This function uses PBKDF2-HMAC-SHA256 to generate a 32-byte AES key396 and a 16-byte IV from the application secret & random salt.397 It then uses AES-256-CTR to encrypt/decrypt the TOTP key.398 399 CTR mode was chosen over CBC because the main attack scenario here400 is that the attacker has stolen the database, and is trying to decrypt a TOTP key401 (the plaintext value here). To make it hard for them, we want every password402 to decrypt to a potentially valid key -- thus need to avoid any authentication403 or padding oracle attacks. While some random padding construction could be devised404 to make this work for CBC mode, a stream cipher mode is just plain simpler.405 OFB/CFB modes would also work here, but seeing as they have malleability406 and cyclic issues (though remote and barely relevant here),407 CTR was picked as the best overall choice.408 """409 # make sure backend AES support is available410 if _cg_ciphers is None:411 raise RuntimeError("TOTP encryption requires 'cryptography' package "412 "(https://cryptography.io)")413 414 # use pbkdf2 to derive both key (32 bytes) & iv (16 bytes)415 # NOTE: this requires 2 sha256 blocks to be calculated.416 keyiv = pbkdf2_hmac("sha256", secret, salt=salt, rounds=(1 << cost), keylen=48)417 418 # use AES-256-CTR to encrypt/decrypt input value419 cipher = _cg_ciphers.Cipher(_cg_ciphers.algorithms.AES(keyiv[:32]),420 _cg_ciphers.modes.CTR(keyiv[32:]),421 _cg_default_backend())422 ctx = cipher.decryptor() if decrypt else cipher.encryptor()423 return ctx.update(value) + ctx.finalize()424 425 def encrypt_key(self, key):426 """427 Helper used to encrypt TOTP keys for storage.428 429 :param key:430 TOTP key to encrypt, as raw bytes.431 432 :returns:433 dict containing encrypted TOTP key & configuration parameters.434 this format should be treated as opaque, and potentially subject435 to change, though it is designed to be easily serialized/deserialized436 (e.g. via JSON).437 438 .. note::439 440 This function requires installation of the external441 `cryptography <https://cryptography.io>`_ package.442 443 To give some algorithm details: This function uses AES-256-CTR to encrypt444 the provided data. It takes the application secret and randomly generated salt,445 and uses PBKDF2-HMAC-SHA256 to combine them and generate the AES key & IV.446 """447 if not key:448 raise ValueError("no key provided")449 salt = getrandbytes(rng, self.salt_size)450 cost = self.encrypt_cost451 tag = self.default_tag452 if not tag:453 raise TypeError("no application secrets configured, can't encrypt OTP key")454 ckey = self._cipher_aes_key(key, self.get_secret(tag), salt, cost)455 # XXX: switch to base64?456 return dict(v=1, c=cost, t=tag, s=b32encode(salt), k=b32encode(ckey))457 458 def decrypt_key(self, enckey):459 """460 Helper used to decrypt TOTP keys from storage format.461 Consults configured secrets to decrypt key.462 463 :param source:464 source object, as returned by :meth:`encrypt_key`.465 466 :returns:467 ``(key, needs_recrypt)`` --468 469 **key** will be the decrypted key, as bytes.470 471 **needs_recrypt** will be a boolean flag indicating472 whether encryption cost or default tag is too old,473 and henace that key needs re-encrypting before storing.474 475 .. note::476 477 This function requires installation of the external478 `cryptography <https://cryptography.io>`_ package.479 """480 if not isinstance(enckey, dict):481 raise TypeError("'enckey' must be dictionary")482 version = enckey.get("v", None)483 needs_recrypt = False484 if version == 1:485 _cipher_key = self._cipher_aes_key486 else:487 raise ValueError("missing / unrecognized 'enckey' version: %r" % (version,))488 tag = enckey['t']489 cost = enckey['c']490 key = _cipher_key(491 value=b32decode(enckey['k']),492 secret=self.get_secret(tag),493 salt=b32decode(enckey['s']),494 cost=cost,495 )496 if cost != self.encrypt_cost or tag != self.default_tag:497 needs_recrypt = True498 return key, needs_recrypt499 500 #=============================================================================501 # eoc502 #=============================================================================503 504#=============================================================================505# TOTP class506#=============================================================================507 508#: helper to convert HOTP counter to bytes509_pack_uint64 = struct.Struct(">Q").pack510 511#: helper to extract value from HOTP digest512_unpack_uint32 = struct.Struct(">I").unpack513 514#: dummy bytes used as temp key for .using() method515_DUMMY_KEY = b"\x00" * 16516 517class TOTP(object):518 """519 Helper for generating and verifying TOTP codes.520 521 Given a secret key and set of configuration options, this object522 offers methods for token generation, token validation, and serialization.523 It can also be used to track important persistent TOTP state,524 such as the last counter used.525 526 This class accepts the following options527 (only **key** and **format** may be specified as positional arguments).528 529 :arg str key:530 The secret key to use. By default, should be encoded as531 a base32 string (see **format** for other encodings).532 533 Exactly one of **key** or ``new=True`` must be specified.534 535 :arg str format:536 The encoding used by the **key** parameter. May be one of:537 ``"base32"`` (base32-encoded string),538 ``"hex"`` (hexadecimal string), or ``"raw"`` (raw bytes).539 Defaults to ``"base32"``.540 541 :param bool new:542 If ``True``, a new key will be generated using :class:`random.SystemRandom`.543 544 Exactly one ``new=True`` or **key** must be specified.545 546 :param str label:547 Label to associate with this token when generating a URI.548 Displayed to user by most OTP client applications (e.g. Google Authenticator),549 and typically has format such as ``"John Smith"`` or ``"jsmith@webservice.example.org"``.550 Defaults to ``None``.551 See :meth:`to_uri` for details.552 553 :param str issuer:554 String identifying the token issuer (e.g. the domain name of your service).555 Used internally by some OTP client applications (e.g. Google Authenticator) to distinguish entries556 which otherwise have the same label.557 Optional but strongly recommended if you're rendering to a URI.558 Defaults to ``None``.559 See :meth:`to_uri` for details.560 561 :param int size:562 Number of bytes when generating new keys. Defaults to size of hash algorithm (e.g. 20 for SHA1).563 564 .. warning::565 566 Overriding the default values for ``digits``, ``period``, or ``alg`` may567 cause problems with some OTP client programs (such as Google Authenticator),568 which may have these defaults hardcoded.569 570 :param int digits:571 The number of digits in the generated / accepted tokens. Defaults to ``6``.572 Must be in range [6 .. 10].573 574 .. rst-class:: inline-title575 .. caution::576 Due to a limitation of the HOTP algorithm, the 10th digit can only take on values 0 .. 2,577 and thus offers very little extra security.578 579 :param str alg:580 Name of hash algorithm to use. Defaults to ``"sha1"``.581 ``"sha256"`` and ``"sha512"`` are also accepted, per :rfc:`6238`.582 583 :param int period:584 The time-step period to use, in integer seconds. Defaults to ``30``.585 586 ..587 See the passlib documentation for a full list of attributes & methods.588 """589 #=============================================================================590 # class attrs591 #=============================================================================592 593 #: minimum number of bytes to allow in key, enforced by passlib.594 # XXX: see if spec says anything relevant to this.595 _min_key_size = 10596 597 #: minimum & current serialization version (may be set independently by subclasses)598 min_json_version = json_version = 1599 600 #: AppWallet that this class will use for encrypting/decrypting keys.601 #: (can be overwritten via the :meth:`TOTP.using()` constructor)602 wallet = None603 604 #: function to get system time in seconds, as needed by :meth:`generate` and :meth:`verify`.605 #: defaults to :func:`time.time`, but can be overridden on a per-instance basis.606 now = _time.time607 608 #=============================================================================609 # instance attrs610 #=============================================================================611 612 #---------------------------------------------------------------------------613 # configuration attrs614 #---------------------------------------------------------------------------615 616 #: [private] secret key as raw :class:`!bytes`617 #: see .key property for public access.618 _key = None619 620 #: [private] cached copy of encrypted secret,621 #: so .to_json() doesn't have to re-encrypt on each call.622 _encrypted_key = None623 624 #: [private] cached copy of keyed HMAC function,625 #: so ._generate() doesn't have to rebuild this each time626 #: ._find_match() invokes it.627 _keyed_hmac = None628 629 #: number of digits in the generated tokens.630 digits = 6631 632 #: name of hash algorithm in use (e.g. ``"sha1"``)633 alg = "sha1"634 635 #: default label for :meth:`to_uri`636 label = None637 638 #: default issuer for :meth:`to_uri`639 issuer = None640 641 #: number of seconds per counter step.642 #: *(TOTP uses an internal time-derived counter which643 #: increments by 1 every* :attr:`!period` *seconds)*.644 period = 30645 646 #---------------------------------------------------------------------------647 # state attrs648 #---------------------------------------------------------------------------649 650 #: Flag set by deserialization methods to indicate the object needs to be re-serialized.651 #: This can be for a number of reasons -- encoded using deprecated format,652 #: or encrypted using a deprecated key or too few rounds.653 changed = False654 655 #=============================================================================656 # prototype construction657 #=============================================================================658 @classmethod659 def using(cls, digits=None, alg=None, period=None,660 issuer=None, wallet=None, now=None, **kwds):661 """662 Dynamically create subtype of :class:`!TOTP` class663 which has the specified defaults set.664 665 :parameters: **digits, alg, period, issuer**:666 667 All these options are the same as in the :class:`TOTP` constructor,668 and the resulting class will use any values you specify here669 as the default for all TOTP instances it creates.670 671 :param wallet:672 Optional :class:`AppWallet` that will be used for encrypting/decrypting keys.673 674 :param secrets, secrets_path, encrypt_cost:675 676 If specified, these options will be passed to the :class:`AppWallet` constructor,677 allowing you to directly specify the secret keys that should be used678 to encrypt & decrypt stored keys.679 680 :returns:681 subclass of :class:`!TOTP`.682 683 This method is useful for creating a TOTP class configured684 to use your application's secrets for encrypting & decrypting685 keys, as well as create new keys using it's desired configuration defaults.686 687 As an example::688 689 >>> # your application can create a custom class when it initializes690 >>> from passlib.totp import TOTP, generate_secret691 >>> TotpFactory = TOTP.using(secrets={"1": generate_secret()})692 693 >>> # subsequent TOTP objects created from this factory694 >>> # will use the specified secrets to encrypt their keys...695 >>> totp = TotpFactory.new()696 >>> totp.to_dict()697 {'enckey': {'c': 14,698 'k': 'H77SYXWORDPGVOQTFRR2HFUB3C45XXI7',699 's': 'G5DOQPIHIBUM2OOHHADQ',700 't': '1',701 'v': 1},702 'type': 'totp',703 'v': 1}704 705 .. seealso:: :ref:`totp-creation` and :ref:`totp-storing-instances` tutorials for a usage example706 """707 # XXX: could add support for setting default match 'window' and 'reuse' policy708 709 # :param now:710 # Optional callable that should return current time for generator to use.711 # Default to :func:`time.time`. This optional is generally not needed,712 # and is mainly present for examples & unit-testing.713 714 subcls = type("TOTP", (cls,), {})715 716 def norm_param(attr, value):717 """718 helper which uses constructor to validate parameter value.719 it returns corresponding attribute, so we use normalized value.720 """721 # NOTE: this creates *subclass* instance,722 # so normalization takes into account any custom params723 # already stored.724 kwds = dict(key=_DUMMY_KEY, format="raw")725 kwds[attr] = value726 obj = subcls(**kwds)727 return getattr(obj, attr)728 729 if digits is not None:730 subcls.digits = norm_param("digits", digits)731 732 if alg is not None:733 subcls.alg = norm_param("alg", alg)734 735 if period is not None:736 subcls.period = norm_param("period", period)737 738 # XXX: add default size as configurable parameter?739 740 if issuer is not None:741 subcls.issuer = norm_param("issuer", issuer)742 743 if kwds:744 subcls.wallet = AppWallet(**kwds)745 if wallet:746 raise TypeError("'wallet' and 'secrets' keywords are mutually exclusive")747 elif wallet is not None:748 if not isinstance(wallet, AppWallet):749 raise exc.ExpectedTypeError(wallet, AppWallet, "wallet")750 subcls.wallet = wallet751 752 if now is not None:753 assert isinstance(now(), num_types) and now() >= 0, \754 "now() function must return non-negative int/float"755 subcls.now = staticmethod(now)756 757 return subcls758 759 #=============================================================================760 # init761 #=============================================================================762 763 @classmethod764 def new(cls, **kwds):765 """766 convenience alias for creating new TOTP key, same as ``TOTP(new=True)``767 """768 return cls(new=True, **kwds)769 770 def __init__(self, key=None, format="base32",771 # keyword only...772 new=False, digits=None, alg=None, size=None, period=None,773 label=None, issuer=None, changed=False,774 **kwds):775 super(TOTP, self).__init__(**kwds)776 if changed:777 self.changed = changed778 779 # validate & normalize alg780 info = lookup_hash(alg or self.alg)781 self.alg = info.name782 digest_size = info.digest_size783 if digest_size < 4:784 raise RuntimeError("%r hash digest too small" % alg)785 786 # parse or generate new key787 if new:788 # generate new key789 if key:790 raise TypeError("'key' and 'new=True' are mutually exclusive")791 if size is None:792 # default to digest size, per RFC 6238 Section 5.1793 size = digest_size794 elif size > digest_size:795 # not forbidden by spec, but would just be wasted bytes.796 # maybe just warn about this?797 raise ValueError("'size' should be less than digest size "798 "(%d)" % digest_size)799 self.key = getrandbytes(rng, size)800 elif not key:801 raise TypeError("must specify either an existing 'key', or 'new=True'")802 elif format == "encrypted":803 # NOTE: this handles decrypting & setting '.key'804 self.encrypted_key = key805 elif key:806 # use existing key, encoded using specified <format>807 self.key = _decode_bytes(key, format)808 809 # enforce min key size810 if len(self.key) < self._min_key_size:811 # only making this fatal for new=True,812 # so that existing (but ridiculously small) keys can still be used.813 msg = "for security purposes, secret key must be >= %d bytes" % self._min_key_size814 if new:815 raise ValueError(msg)816 else:817 warn(msg, exc.PasslibSecurityWarning, stacklevel=1)818 819 # validate digits820 if digits is None:821 digits = self.digits822 if not isinstance(digits, int_types):823 raise TypeError("digits must be an integer, not a %r" % type(digits))824 if digits < 6 or digits > 10:825 raise ValueError("digits must in range(6,11)")826 self.digits = digits827 828 # validate label829 if label:830 self._check_label(label)831 self.label = label832 833 # validate issuer834 if issuer:835 self._check_issuer(issuer)836 self.issuer = issuer837 838 # init period839 if period is not None:840 self._check_serial(period, "period", minval=1)841 self.period = period842 843 #=============================================================================844 # helpers to verify value types & ranges845 #=============================================================================846 847 @staticmethod848 def _check_serial(value, param, minval=0):849 """850 check that serial value (e.g. 'counter') is non-negative integer851 """852 if not isinstance(value, int_types):853 raise exc.ExpectedTypeError(value, "int", param)854 if value < minval:855 raise ValueError("%s must be >= %d" % (param, minval))856 857 @staticmethod858 def _check_label(label):859 """860 check that label doesn't contain chars forbidden by KeyURI spec861 """862 if label and ":" in label:863 raise ValueError("label may not contain ':'")864 865 @staticmethod866 def _check_issuer(issuer):867 """868 check that issuer doesn't contain chars forbidden by KeyURI spec869 """870 if issuer and ":" in issuer:871 raise ValueError("issuer may not contain ':'")872 873 #=============================================================================874 # key attributes875 #=============================================================================876 877 #------------------------------------------------------------------878 # raw key879 #------------------------------------------------------------------880 @property881 def key(self):882 """883 secret key as raw bytes884 """885 return self._key886 887 @key.setter888 def key(self, value):889 # set key890 if not isinstance(value, bytes):891 raise exc.ExpectedTypeError(value, bytes, "key")892 self._key = value893 894 # clear cached properties derived from key895 self._encrypted_key = self._keyed_hmac = None896 897 #------------------------------------------------------------------898 # encrypted key899 #------------------------------------------------------------------900 @property901 def encrypted_key(self):902 """903 secret key, encrypted using application secret.904 this match the output of :meth:`AppWallet.encrypt_key`,905 and should be treated as an opaque json serializable object.906 """907 enckey = self._encrypted_key908 if enckey is None:909 wallet = self.wallet910 if not wallet:911 raise TypeError("no application secrets present, can't encrypt TOTP key")912 enckey = self._encrypted_key = wallet.encrypt_key(self.key)913 return enckey914 915 @encrypted_key.setter916 def encrypted_key(self, value):917 wallet = self.wallet918 if not wallet:919 raise TypeError("no application secrets present, can't decrypt TOTP key")920 self.key, needs_recrypt = wallet.decrypt_key(value)921 if needs_recrypt:922 # mark as changed so it gets re-encrypted & written to db923 self.changed = True924 else:925 # cache encrypted key for re-use926 self._encrypted_key = value927 928 #------------------------------------------------------------------929 # pretty-printed / encoded key helpers930 #------------------------------------------------------------------931 932 @property933 def hex_key(self):934 """935 secret key encoded as hexadecimal string936 """937 return bascii_to_str(base64.b16encode(self.key)).lower()938 939 @property940 def base32_key(self):941 """942 secret key encoded as base32 string943 """944 return b32encode(self.key)945 946 def pretty_key(self, format="base32", sep="-"):947 """948 pretty-print the secret key.949 950 This is mainly useful for situations where the user cannot get the qrcode to work,951 and must enter the key manually into their TOTP client. It tries to format952 the key in a manner that is easier for humans to read.953 954 :param format:955 format to output secret key. ``"hex"`` and ``"base32"`` are both accepted.956 957 :param sep:958 separator to insert to break up key visually.959 can be any of ``"-"`` (the default), ``" "``, or ``False`` (no separator).960 961 :return:962 key as native string.963 964 Usage example::965 966 >>> t = TOTP('s3jdvb7qd2r7jpxx')967 >>> t.pretty_key()968 'S3JD-VB7Q-D2R7-JPXX'969 """970 if format == "hex" or format == "base16":971 key = self.hex_key972 elif format == "base32":973 key = self.base32_key974 else:975 raise ValueError("unknown byte-encoding format: %r" % (format,))976 if sep:977 key = group_string(key, sep)978 return key979 980 #=============================================================================981 # time & token parsing982 #=============================================================================983 984 @classmethod985 def normalize_time(cls, time):986 """987 Normalize time value to unix epoch seconds.988 989 :arg time:990 Can be ``None``, :class:`!datetime`,991 or unix epoch timestamp as :class:`!float` or :class:`!int`.992 If ``None``, uses current system time.993 Naive datetimes are treated as UTC.994 995 :returns:996 unix epoch timestamp as :class:`int`.997 """998 if isinstance(time, int_types):999 return time1000 elif isinstance(time, float):1001 return int(time)1002 elif time is None:1003 return int(cls.now())1004 elif hasattr(time, "utctimetuple"):1005 # coerce datetime to UTC timestamp1006 # NOTE: utctimetuple() assumes naive datetimes are in UTC1007 # NOTE: we explicitly *don't* want microseconds.1008 return calendar.timegm(time.utctimetuple())1009 else:1010 raise exc.ExpectedTypeError(time, "int, float, or datetime", "time")1011 1012 def _time_to_counter(self, time):1013 """1014 convert timestamp to HOTP counter using :attr:`period`.1015 """1016 return time // self.period1017 1018 def _counter_to_time(self, counter):1019 """1020 convert HOTP counter to timestamp using :attr:`period`.1021 """1022 return counter * self.period1023 1024 @hybrid_method1025 def normalize_token(self_or_cls, token):1026 """1027 Normalize OTP token representation:1028 strips whitespace, converts integers to a zero-padded string,1029 validates token content & number of digits.1030 1031 This is a hybrid method -- it can be called at the class level,1032 as ``TOTP.normalize_token()``, or the instance level as ``TOTP().normalize_token()``.1033 It will normalize to the instance-specific number of :attr:`~TOTP.digits`,1034 or use the class default.1035 1036 :arg token:1037 token as ascii bytes, unicode, or an integer.1038 1039 :raises ValueError:1040 if token has wrong number of digits, or contains non-numeric characters.1041 1042 :returns:1043 token as :class:`!unicode` string, containing only digits 0-9.1044 """1045 digits = self_or_cls.digits1046 if isinstance(token, int_types):1047 token = u("%0*d") % (digits, token)1048 else:1049 token = to_unicode(token, param="token")1050 token = _clean_re.sub(u(""), token)1051 if not token.isdigit():1052 raise MalformedTokenError("Token must contain only the digits 0-9")1053 if len(token) != digits:1054 raise MalformedTokenError("Token must have exactly %d digits" % digits)1055 return token1056 1057 #=============================================================================1058 # token generation1059 #=============================================================================1060 1061# # debug helper1062# def generate_range(self, size, time=None):1063# counter = self._time_to_counter(time) - (size + 1) // 21064# end = counter + size1065# while counter <= end:1066# token = self._generate(counter)1067# yield TotpToken(self, token, counter)1068# counter += 11069 1070 def generate(self, time=None):1071 """1072 Generate token for specified time1073 (uses current time if none specified).1074 1075 :arg time:1076 Can be ``None``, a :class:`!datetime`,1077 or class:`!float` / :class:`!int` unix epoch timestamp.1078 If ``None`` (the default), uses current system time.1079 Naive datetimes are treated as UTC.1080 1081 :returns:1082 1083 A :class:`TotpToken` instance, which can be treated1084 as a sequence of ``(token, expire_time)`` -- see that class1085 for more details.1086 1087 Usage example::1088 1089 >>> # generate a new token, wrapped in a TotpToken instance...1090 >>> otp = TOTP('s3jdvb7qd2r7jpxx')1091 >>> otp.generate(1419622739)1092 <TotpToken token='897212' expire_time=1419622740>1093 1094 >>> # when you just need the token...1095 >>> otp.generate(1419622739).token1096 '897212'1097 """1098 time = self.normalize_time(time)1099 counter = self._time_to_counter(time)1100 if counter < 0:1101 raise ValueError("timestamp must be >= 0")1102 token = self._generate(counter)1103 return TotpToken(self, token, counter)1104 1105 def _generate(self, counter):1106 """1107 base implementation of HOTP token generation algorithm.1108 1109 :arg counter: HOTP counter, as non-negative integer1110 :returns: token as unicode string1111 """1112 # generate digest1113 assert isinstance(counter, int_types), "counter must be integer"1114 assert counter >= 0, "counter must be non-negative"1115 keyed_hmac = self._keyed_hmac1116 if keyed_hmac is None:1117 keyed_hmac = self._keyed_hmac = compile_hmac(self.alg, self.key)1118 digest = keyed_hmac(_pack_uint64(counter))1119 digest_size = keyed_hmac.digest_info.digest_size1120 assert len(digest) == digest_size, "digest_size: sanity check failed"1121 1122 # derive 31-bit token value1123 assert digest_size >= 20, "digest_size: sanity check 2 failed" # otherwise 0xF+4 will run off end of hash.1124 offset = byte_elem_value(digest[-1]) & 0xF1125 value = _unpack_uint32(digest[offset:offset+4])[0] & 0x7fffffff1126 1127 # render to decimal string, return last <digits> chars1128 # NOTE: the 10'th digit is not as secure, as it can only take on values 0-2, not 0-9,1129 # due to 31-bit mask on int ">I". But some servers / clients use it :|1130 # if 31-bit mask removed (which breaks spec), would only get values 0-4.1131 digits = self.digits1132 assert 0 < digits < 11, "digits: sanity check failed"1133 return (u("%0*d") % (digits, value))[-digits:]1134 1135 #=============================================================================1136 # token verification1137 #=============================================================================1138 1139 @classmethod1140 def verify(cls, token, source, **kwds):1141 r"""1142 Convenience wrapper around :meth:`TOTP.from_source` and :meth:`TOTP.match`.1143 1144 This parses a TOTP key & configuration from the specified source,1145 and tries and match the token.1146 It's designed to parallel the :meth:`passlib.ifc.PasswordHash.verify` method.1147 1148 :param token:1149 Token string to match.1150 1151 :param source:1152 Serialized TOTP key.1153 Can be anything accepted by :meth:`TOTP.from_source`.1154 1155 :param \\*\\*kwds:1156 All additional keywords passed to :meth:`TOTP.match`.1157 1158 :return:1159 A :class:`TotpMatch` instance, or raises a :exc:`TokenError`.1160 """1161 return cls.from_source(source).match(token, **kwds)1162 1163 def match(self, token, time=None, window=30, skew=0, last_counter=None):1164 """1165 Match TOTP token against specified timestamp.1166 Searches within a window before & after the provided time,1167 in order to account for transmission delay and small amounts of skew in the client's clock.1168 1169 :arg token:1170 Token to validate.1171 may be integer or string (whitespace and hyphens are ignored).1172 1173 :param time:1174 Unix epoch timestamp, can be any of :class:`!float`, :class:`!int`, or :class:`!datetime`.1175 if ``None`` (the default), uses current system time.1176 *this should correspond to the time the token was received from the client*.1177 1178 :param int window:1179 How far backward and forward in time to search for a match.1180 Measured in seconds. Defaults to ``30``. Typically only useful if set1181 to multiples of :attr:`period`.1182 1183 :param int skew:1184 Adjust timestamp by specified value, to account for excessive1185 client clock skew. Measured in seconds. Defaults to ``0``.1186 1187 Negative skew (the common case) indicates transmission delay,1188 and/or that the client clock is running behind the server.1189 1190 Positive skew indicates the client clock is running ahead of the server1191 (and by enough that it cancels out any negative skew added by1192 the transmission delay).1193 1194 You should ensure the server clock uses a reliable time source such as NTP,1195 so that only the client clock's inaccuracy needs to be accounted for.1196 1197 This is an advanced parameter that should usually be left at ``0``;1198 The **window** parameter is usually enough to account1199 for any observed transmission delay.1200