res_levis_API/.venv/lib/python3.10/site-packages/passlib/handlers/scram.py

"""passlib.handlers.scram - hash for SCRAM credential storage"""
#=============================================================================
# imports
#=============================================================================
# core
import logging; log = logging.getLogger(__name__)
# site
# pkg
from passlib.utils import consteq, saslprep, to_native_str, splitcomma
from passlib.utils.binary import ab64_decode, ab64_encode
from passlib.utils.compat import bascii_to_str, iteritems, u, native_string_types
from passlib.crypto.digest import pbkdf2_hmac, norm_hash_name
import passlib.utils.handlers as uh
# local
__all__ = [
    "scram",
]

#=============================================================================
# scram credentials hash
#=============================================================================
class scram(uh.HasRounds, uh.HasRawSalt, uh.HasRawChecksum, uh.GenericHandler):
    """This class provides a format for storing SCRAM passwords, and follows
    the :ref:`password-hash-api`.

    It supports a variable-length salt, and a variable number of rounds.

    The :meth:`~passlib.ifc.PasswordHash.using` method accepts the following optional keywords:

    :type salt: bytes
    :param salt:
        Optional salt bytes.
        If specified, the length must be between 0-1024 bytes.
        If not specified, a 12 byte salt will be autogenerated
        (this is recommended).

    :type salt_size: int
    :param salt_size:
        Optional number of bytes to use when autogenerating new salts.
        Defaults to 12 bytes, but can be any value between 0 and 1024.

    :type rounds: int
    :param rounds:
        Optional number of rounds to use.
        Defaults to 100000, but must be within ``range(1,1<<32)``.

    :type algs: list of strings
    :param algs:
        Specify list of digest algorithms to use.

        By default each scram hash will contain digests for SHA-1,
        SHA-256, and SHA-512. This can be overridden by specify either be a
        list such as ``["sha-1", "sha-256"]``, or a comma-separated string
        such as ``"sha-1, sha-256"``. Names are case insensitive, and may
        use :mod:`!hashlib` or `IANA <http://www.iana.org/assignments/hash-function-text-names>`_
        hash names.

    :type relaxed: bool
    :param relaxed:
        By default, providing an invalid value for one of the other
        keywords will result in a :exc:`ValueError`. If ``relaxed=True``,
        and the error can be corrected, a :exc:`~passlib.exc.PasslibHashWarning`
        will be issued instead. Correctable errors include ``rounds``
        that are too small or too large, and ``salt`` strings that are too long.

        .. versionadded:: 1.6

    In addition to the standard :ref:`password-hash-api` methods,
    this class also provides the following methods for manipulating Passlib
    scram hashes in ways useful for pluging into a SCRAM protocol stack:

    .. automethod:: extract_digest_info
    .. automethod:: extract_digest_algs
    .. automethod:: derive_digest
    """
    #===================================================================
    # class attrs
    #===================================================================

    # NOTE: unlike most GenericHandler classes, the 'checksum' attr of
    # ScramHandler is actually a map from digest_name -> digest, so
    # many of the standard methods have been overridden.

    # NOTE: max_salt_size and max_rounds are arbitrarily chosen to provide
    # a sanity check; the underlying pbkdf2 specifies no bounds for either.

    #--GenericHandler--
    name = "scram"
    setting_kwds = ("salt", "salt_size", "rounds", "algs")
    ident = u("$scram$")

    #--HasSalt--
    default_salt_size = 12
    max_salt_size = 1024

    #--HasRounds--
    default_rounds = 100000
    min_rounds = 1
    max_rounds = 2**32-1
    rounds_cost = "linear"

    #--custom--

    # default algorithms when creating new hashes.
    default_algs = ["sha-1", "sha-256", "sha-512"]

    # list of algs verify prefers to use, in order.
    _verify_algs = ["sha-256", "sha-512", "sha-224", "sha-384", "sha-1"]

    #===================================================================
    # instance attrs
    #===================================================================

    # 'checksum' is different from most GenericHandler subclasses,
    # in that it contains a dict mapping from alg -> digest,
    # or None if no checksum present.

    # list of algorithms to create/compare digests for.
    algs = None

    #===================================================================
    # scram frontend helpers
    #===================================================================
    @classmethod
    def extract_digest_info(cls, hash, alg):
        """return (salt, rounds, digest) for specific hash algorithm.

        :type hash: str
        :arg hash:
            :class:`!scram` hash stored for desired user

        :type alg: str
        :arg alg:
            Name of digest algorithm (e.g. ``"sha-1"``) requested by client.

            This value is run through :func:`~passlib.crypto.digest.norm_hash_name`,
            so it is case-insensitive, and can be the raw SCRAM
            mechanism name (e.g. ``"SCRAM-SHA-1"``), the IANA name,
            or the hashlib name.

        :raises KeyError:
            If the hash does not contain an entry for the requested digest
            algorithm.

        :returns:
            A tuple containing ``(salt, rounds, digest)``,
            where *digest* matches the raw bytes returned by
            SCRAM's :func:`Hi` function for the stored password,
            the provided *salt*, and the iteration count (*rounds*).
            *salt* and *digest* are both raw (unencoded) bytes.
        """
        # XXX: this could be sped up by writing custom parsing routine
        # that just picks out relevant digest, and doesn't bother
        # with full structure validation each time it's called.
        alg = norm_hash_name(alg, 'iana')
        self = cls.from_string(hash)
        chkmap = self.checksum
        if not chkmap:
            raise ValueError("scram hash contains no digests")
        return self.salt, self.rounds, chkmap[alg]

    @classmethod
    def extract_digest_algs(cls, hash, format="iana"):
        """Return names of all algorithms stored in a given hash.

        :type hash: str
        :arg hash:
            The :class:`!scram` hash to parse

        :type format: str
        :param format:
            This changes the naming convention used by the
            returned algorithm names. By default the names
            are IANA-compatible; possible values are ``"iana"`` or ``"hashlib"``.

        :returns:
            Returns a list of digest algorithms; e.g. ``["sha-1"]``
        """
        # XXX: this could be sped up by writing custom parsing routine
        # that just picks out relevant names, and doesn't bother
        # with full structure validation each time it's called.
        algs = cls.from_string(hash).algs
        if format == "iana":
            return algs
        else:
            return [norm_hash_name(alg, format) for alg in algs]

    @classmethod
    def derive_digest(cls, password, salt, rounds, alg):
        """helper to create SaltedPassword digest for SCRAM.

        This performs the step in the SCRAM protocol described as::

            SaltedPassword  := Hi(Normalize(password), salt, i)

        :type password: unicode or utf-8 bytes
        :arg password: password to run through digest

        :type salt: bytes
        :arg salt: raw salt data

        :type rounds: int
        :arg rounds: number of iterations.

        :type alg: str
        :arg alg: name of digest to use (e.g. ``"sha-1"``).

        :returns:
            raw bytes of ``SaltedPassword``
        """
        if isinstance(password, bytes):
            password = password.decode("utf-8")
        # NOTE: pbkdf2_hmac() will encode secret & salt using utf-8,
        #       and handle normalizing alg name.
        return pbkdf2_hmac(alg, saslprep(password), salt, rounds)

    #===================================================================
    # serialization
    #===================================================================

    @classmethod
    def from_string(cls, hash):
        hash = to_native_str(hash, "ascii", "hash")
        if not hash.startswith("$scram$"):
            raise uh.exc.InvalidHashError(cls)
        parts = hash[7:].split("$")
        if len(parts) != 3:
            raise uh.exc.MalformedHashError(cls)
        rounds_str, salt_str, chk_str = parts

        # decode rounds
        rounds = int(rounds_str)
        if rounds_str != str(rounds): # forbid zero padding, etc.
            raise uh.exc.MalformedHashError(cls)

        # decode salt
        try:
            salt = ab64_decode(salt_str.encode("ascii"))
        except TypeError:
            raise uh.exc.MalformedHashError(cls)

        # decode algs/digest list
        if not chk_str:
            # scram hashes MUST have something here.
            raise uh.exc.MalformedHashError(cls)
        elif "=" in chk_str:
            # comma-separated list of 'alg=digest' pairs
            algs = None
            chkmap = {}
            for pair in chk_str.split(","):
                alg, digest = pair.split("=")
                try:
                    chkmap[alg] = ab64_decode(digest.encode("ascii"))
                except TypeError:
                    raise uh.exc.MalformedHashError(cls)
        else:
            # comma-separated list of alg names, no digests
            algs = chk_str
            chkmap = None

        # return new object
        return cls(
            rounds=rounds,
            salt=salt,
            checksum=chkmap,
            algs=algs,
        )

    def to_string(self):
        salt = bascii_to_str(ab64_encode(self.salt))
        chkmap = self.checksum
        chk_str = ",".join(
            "%s=%s" % (alg, bascii_to_str(ab64_encode(chkmap[alg])))
            for alg in self.algs
        )
        return '$scram$%d$%s$%s' % (self.rounds, salt, chk_str)

    #===================================================================
    # variant constructor
    #===================================================================
    @classmethod
    def using(cls, default_algs=None, algs=None, **kwds):
        # parse aliases
        if algs is not None:
            assert default_algs is None
            default_algs = algs

        # create subclass
        subcls = super(scram, cls).using(**kwds)

        # fill in algs
        if default_algs is not None:
            subcls.default_algs = cls._norm_algs(default_algs)
        return subcls

    #===================================================================
    # init
    #===================================================================
    def __init__(self, algs=None, **kwds):
        super(scram, self).__init__(**kwds)

        # init algs
        digest_map = self.checksum
        if algs is not None:
            if digest_map is not None:
                raise RuntimeError("checksum & algs kwds are mutually exclusive")
            algs = self._norm_algs(algs)
        elif digest_map is not None:
            # derive algs list from digest map (if present).
            algs = self._norm_algs(digest_map.keys())
        elif self.use_defaults:
            algs = list(self.default_algs)
            assert self._norm_algs(algs) == algs, "invalid default algs: %r" % (algs,)
        else:
            raise TypeError("no algs list specified")
        self.algs = algs

    def _norm_checksum(self, checksum, relaxed=False):
        if not isinstance(checksum, dict):
            raise uh.exc.ExpectedTypeError(checksum, "dict", "checksum")
        for alg, digest in iteritems(checksum):
            if alg != norm_hash_name(alg, 'iana'):
                raise ValueError("malformed algorithm name in scram hash: %r" %
                                 (alg,))
            if len(alg) > 9:
                raise ValueError("SCRAM limits algorithm names to "
                                 "9 characters: %r" % (alg,))
            if not isinstance(digest, bytes):
                raise uh.exc.ExpectedTypeError(digest, "raw bytes", "digests")
            # TODO: verify digest size (if digest is known)
        if 'sha-1' not in checksum:
            # NOTE: required because of SCRAM spec.
            raise ValueError("sha-1 must be in algorithm list of scram hash")
        return checksum

    @classmethod
    def _norm_algs(cls, algs):
        """normalize algs parameter"""
        if isinstance(algs, native_string_types):
            algs = splitcomma(algs)
        algs = sorted(norm_hash_name(alg, 'iana') for alg in algs)
        if any(len(alg)>9 for alg in algs):
            raise ValueError("SCRAM limits alg names to max of 9 characters")
        if 'sha-1' not in algs:
            # NOTE: required because of SCRAM spec (rfc 5802)
            raise ValueError("sha-1 must be in algorithm list of scram hash")
        return algs

    #===================================================================
    # migration
    #===================================================================
    def _calc_needs_update(self, **kwds):
        # marks hashes as deprecated if they don't include at least all default_algs.
        # XXX: should we deprecate if they aren't exactly the same,
        #      to permit removing legacy hashes?
        if not set(self.algs).issuperset(self.default_algs):
            return True

        # hand off to base implementation
        return super(scram, self)._calc_needs_update(**kwds)

    #===================================================================
    # digest methods
    #===================================================================
    def _calc_checksum(self, secret, alg=None):
        rounds = self.rounds
        salt = self.salt
        hash = self.derive_digest
        if alg:
            # if requested, generate digest for specific alg
            return hash(secret, salt, rounds, alg)
        else:
            # by default, return dict containing digests for all algs
            return dict(
                (alg, hash(secret, salt, rounds, alg))
                for alg in self.algs
            )

    @classmethod
    def verify(cls, secret, hash, full=False):
        uh.validate_secret(secret)
        self = cls.from_string(hash)
        chkmap = self.checksum
        if not chkmap:
            raise ValueError("expected %s hash, got %s config string instead" %
                             (cls.name, cls.name))

        # NOTE: to make the verify method efficient, we just calculate hash
        # of shortest digest by default. apps can pass in "full=True" to
        # check entire hash for consistency.
        if full:
            correct = failed = False
            for alg, digest in iteritems(chkmap):
                other = self._calc_checksum(secret, alg)
                # NOTE: could do this length check in norm_algs(),
                # but don't need to be that strict, and want to be able
                # to parse hashes containing algs not supported by platform.
                # it's fine if we fail here though.
                if len(digest) != len(other):
                    raise ValueError("mis-sized %s digest in scram hash: %r != %r"
                                     % (alg, len(digest), len(other)))
                if consteq(other, digest):
                    correct = True
                else:
                    failed = True
            if correct and failed:
                raise ValueError("scram hash verified inconsistently, "
                                 "may be corrupted")
            else:
                return correct
        else:
            # XXX: should this just always use sha1 hash? would be faster.
            # otherwise only verify against one hash, pick one w/ best security.
            for alg in self._verify_algs:
                if alg in chkmap:
                    other = self._calc_checksum(secret, alg)
                    return consteq(other, chkmap[alg])
            # there should always be sha-1 at the very least,
            # or something went wrong inside _norm_algs()
            raise AssertionError("sha-1 digest not found!")

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

#=============================================================================
# code used for testing scram against protocol examples during development.
#=============================================================================
##def _test_reference_scram():
##    "quick hack testing scram reference vectors"
##    # NOTE: "n,," is GS2 header - see https://tools.ietf.org/html/rfc5801
##    from passlib.utils.compat import print_
##
##    engine = _scram_engine(
##        alg="sha-1",
##        salt='QSXCR+Q6sek8bf92'.decode("base64"),
##        rounds=4096,
##        password=u("pencil"),
##    )
##    print_(engine.digest.encode("base64").rstrip())
##
##    msg = engine.format_auth_msg(
##        username="user",
##        client_nonce = "fyko+d2lbbFgONRv9qkxdawL",
##        server_nonce = "3rfcNHYJY1ZVvWVs7j",
##        header='c=biws',
##    )
##
##    cp = engine.get_encoded_client_proof(msg)
##    assert cp == "v0X8v3Bz2T0CJGbJQyF0X+HI4Ts=", cp
##
##    ss = engine.get_encoded_server_sig(msg)
##    assert ss == "rmF9pqV8S7suAoZWja4dJRkFsKQ=", ss
##
##class _scram_engine(object):
##    """helper class for verifying scram hash behavior
##    against SCRAM protocol examples. not officially part of Passlib.
##
##    takes in alg, salt, rounds, and a digest or password.
##
##    can calculate the various keys & messages of the scram protocol.
##
##    """
##    #=========================================================
##    # init
##    #=========================================================
##
##    @classmethod
##    def from_string(cls, hash, alg):
##        "create record from scram hash, for given alg"
##        return cls(alg, *scram.extract_digest_info(hash, alg))
##
##    def __init__(self, alg, salt, rounds, digest=None, password=None):
##        self.alg = norm_hash_name(alg)
##        self.salt = salt
##        self.rounds = rounds
##        self.password = password
##        if password:
##            data = scram.derive_digest(password, salt, rounds, alg)
##            if digest and data != digest:
##                raise ValueError("password doesn't match digest")
##            else:
##                digest = data
##        elif not digest:
##            raise TypeError("must provide password or digest")
##        self.digest = digest
##
##    #=========================================================
##    # frontend methods
##    #=========================================================
##    def get_hash(self, data):
##        "return hash of raw data"
##        return hashlib.new(iana_to_hashlib(self.alg), data).digest()
##
##    def get_client_proof(self, msg):
##        "return client proof of specified auth msg text"
##        return xor_bytes(self.client_key, self.get_client_sig(msg))
##
##    def get_encoded_client_proof(self, msg):
##        return self.get_client_proof(msg).encode("base64").rstrip()
##
##    def get_client_sig(self, msg):
##        "return client signature of specified auth msg text"
##        return self.get_hmac(self.stored_key, msg)
##
##    def get_server_sig(self, msg):
##        "return server signature of specified auth msg text"
##        return self.get_hmac(self.server_key, msg)
##
##    def get_encoded_server_sig(self, msg):
##        return self.get_server_sig(msg).encode("base64").rstrip()
##
##    def format_server_response(self, client_nonce, server_nonce):
##        return 'r={client_nonce}{server_nonce},s={salt},i={rounds}'.format(
##            client_nonce=client_nonce,
##            server_nonce=server_nonce,
##            rounds=self.rounds,
##            salt=self.encoded_salt,
##            )
##
##    def format_auth_msg(self, username, client_nonce, server_nonce,
##                        header='c=biws'):
##        return (
##            'n={username},r={client_nonce}'
##                ','
##            'r={client_nonce}{server_nonce},s={salt},i={rounds}'
##                ','
##            '{header},r={client_nonce}{server_nonce}'
##            ).format(
##                username=username,
##                client_nonce=client_nonce,
##                server_nonce=server_nonce,
##                salt=self.encoded_salt,
##                rounds=self.rounds,
##                header=header,
##                )
##
##    #=========================================================
##    # helpers to calculate & cache constant data
##    #=========================================================
##    def _calc_get_hmac(self):
##        return get_prf("hmac-" + iana_to_hashlib(self.alg))[0]
##
##    def _calc_client_key(self):
##        return self.get_hmac(self.digest, b("Client Key"))
##
##    def _calc_stored_key(self):
##        return self.get_hash(self.client_key)
##
##    def _calc_server_key(self):
##        return self.get_hmac(self.digest, b("Server Key"))
##
##    def _calc_encoded_salt(self):
##        return self.salt.encode("base64").rstrip()
##
##    #=========================================================
##    # hacks for calculated attributes
##    #=========================================================
##
##    def __getattr__(self, attr):
##        if not attr.startswith("_"):
##            f = getattr(self, "_calc_" + attr, None)
##            if f:
##                value = f()
##                setattr(self, attr, value)
##                return value
##        raise AttributeError("attribute not found")
##
##    def __dir__(self):
##        cdir = dir(self.__class__)
##        attrs = set(cdir)
##        attrs.update(self.__dict__)
##        attrs.update(attr[6:] for attr in cdir
##                     if attr.startswith("_calc_"))
##        return sorted(attrs)
##    #=========================================================
##    # eoc
##    #=========================================================

#=============================================================================
# eof
#=============================================================================