path: root/paste/auth/digest.py

# (c) 2005 Clark C. Evans
# This module is part of the Python Paste Project and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
# This code was written with funding by http://prometheusresearch.com
"""
Digest HTTP/1.1 Authentication

This module implements ``Digest`` authentication as described by
RFC 2617 [1]_ .

Basically, you just put this module before your application, and it
takes care of requesting and handling authentication requests.  This
module has been tested with several common browsers "out-in-the-wild".

>>> from paste.wsgilib import dump_environ
>>> from paste.httpserver import serve
>>> # from paste.auth.digest import digest_password, AuthDigestHandler
>>> realm = 'Test Realm'
>>> def authfunc(environ, realm, username):
...     return digest_password(realm, username, username)
>>> serve(AuthDigestHandler(dump_environ, realm, authfunc))
serving on...

This code has not been audited by a security expert, please use with
caution (or better yet, report security holes). At this time, this
implementation does not provide for further challenges, nor does it
support Authentication-Info header.  It also uses md5, and an option
to use sha would be a good thing.

.. [1] http://www.faqs.org/rfcs/rfc2617.html
"""
from paste.httpexceptions import HTTPUnauthorized
from paste.httpheaders import *
try:
    from hashlib import md5
except ImportError:
    from md5 import md5
import time, random
from six.moves.urllib.parse import quote as url_quote
import six

def _split_auth_string(auth_string):
    """ split a digest auth string into individual key=value strings """
    prev = None
    for item in auth_string.split(","):
        try:
            if prev.count('"') == 1:
                prev = "%s,%s" % (prev, item)
                continue
        except AttributeError:
            if prev == None:
                prev = item
                continue
            else:
                raise StopIteration
        yield prev.strip()
        prev = item

    yield prev.strip()
    return

def _auth_to_kv_pairs(auth_string):
    """ split a digest auth string into key, value pairs """
    for item in _split_auth_string(auth_string):
        (k, v) = item.split("=", 1)
        if v.startswith('"') and len(v) > 1 and v.endswith('"'):
            v = v[1:-1]
        yield (k, v)

def digest_password(realm, username, password):
    """ construct the appropriate hashcode needed for HTTP digest """
    content = "%s:%s:%s" % (username, realm, password)
    if six.PY3:
        content = content.encode('utf8')
    return md5(content).hexdigest()

class AuthDigestAuthenticator(object):
    """ implementation of RFC 2617 - HTTP Digest Authentication """
    def __init__(self, realm, authfunc):
        self.nonce    = {} # list to prevent replay attacks
        self.authfunc = authfunc
        self.realm    = realm

    def build_authentication(self, stale = ''):
        """ builds the authentication error """
        content = "%s:%s" % (time.time(), random.random())
        if six.PY3:
            content = content.encode('utf-8')
        nonce  = md5(content).hexdigest()

        content = "%s:%s" % (time.time(), random.random())
        if six.PY3:
            content = content.encode('utf-8')
        opaque = md5(content).hexdigest()

        self.nonce[nonce] = None
        parts = {'realm': self.realm, 'qop': 'auth',
                 'nonce': nonce, 'opaque': opaque }
        if stale:
            parts['stale'] = 'true'
        head = ", ".join(['%s="%s"' % (k, v) for (k, v) in parts.items()])
        head = [("WWW-Authenticate", 'Digest %s' % head)]
        return HTTPUnauthorized(headers=head)

    def compute(self, ha1, username, response, method,
                      path, nonce, nc, cnonce, qop):
        """ computes the authentication, raises error if unsuccessful """
        if not ha1:
            return self.build_authentication()
        content = '%s:%s' % (method, path)
        if six.PY3:
            content = content.encode('utf8')
        ha2 = md5(content).hexdigest()
        if qop:
            chk = "%s:%s:%s:%s:%s:%s" % (ha1, nonce, nc, cnonce, qop, ha2)
        else:
            chk = "%s:%s:%s" % (ha1, nonce, ha2)
        if six.PY3:
            chk = chk.encode('utf8')
        if response != md5(chk).hexdigest():
            if nonce in self.nonce:
                del self.nonce[nonce]
            return self.build_authentication()
        pnc = self.nonce.get(nonce,'00000000')
        if pnc is not None and nc <= pnc:
            if nonce in self.nonce:
                del self.nonce[nonce]
            return self.build_authentication(stale = True)
        self.nonce[nonce] = nc
        return username

    def authenticate(self, environ):
        """ This function takes a WSGI environment and authenticates
            the request returning authenticated user or error.
        """
        method = REQUEST_METHOD(environ)
        fullpath = url_quote(SCRIPT_NAME(environ)) + url_quote(PATH_INFO(environ))
        authorization = AUTHORIZATION(environ)
        if not authorization:
            return self.build_authentication()
        (authmeth, auth) = authorization.split(" ", 1)
        if 'digest' != authmeth.lower():
            return self.build_authentication()
        amap = dict(_auth_to_kv_pairs(auth))
        try:
            username = amap['username']
            authpath = amap['uri']
            nonce    = amap['nonce']
            realm    = amap['realm']
            response = amap['response']
            assert authpath.split("?", 1)[0] in fullpath
            assert realm == self.realm
            qop      = amap.get('qop', '')
            cnonce   = amap.get('cnonce', '')
            nc       = amap.get('nc', '00000000')
            if qop:
                assert 'auth' == qop
                assert nonce and nc
        except:
            return self.build_authentication()
        ha1 = self.authfunc(environ, realm, username)
        return self.compute(ha1, username, response, method, authpath,
                            nonce, nc, cnonce, qop)

    __call__ = authenticate

class AuthDigestHandler(object):
    """
    middleware for HTTP Digest authentication (RFC 2617)

    This component follows the procedure below:

        0. If the REMOTE_USER environment variable is already populated;
           then this middleware is a no-op, and the request is passed
           along to the application.

        1. If the HTTP_AUTHORIZATION header was not provided or specifies
           an algorithem other than ``digest``, then a HTTPUnauthorized
           response is generated with the challenge.

        2. If the response is malformed or or if the user's credientials
           do not pass muster, another HTTPUnauthorized is raised.

        3. If all goes well, and the user's credintials pass; then
           REMOTE_USER environment variable is filled in and the
           AUTH_TYPE is listed as 'digest'.

    Parameters:

        ``application``

            The application object is called only upon successful
            authentication, and can assume ``environ['REMOTE_USER']``
            is set.  If the ``REMOTE_USER`` is already set, this
            middleware is simply pass-through.

        ``realm``

            This is a identifier for the authority that is requesting
            authorization.  It is shown to the user and should be unique
            within the domain it is being used.

        ``authfunc``

            This is a callback function which performs the actual
            authentication; the signature of this callback is:

              authfunc(environ, realm, username) -> hashcode

            This module provides a 'digest_password' helper function
            which can help construct the hashcode; it is recommended
            that the hashcode is stored in a database, not the user's
            actual password (since you only need the hashcode).
    """
    def __init__(self, application, realm, authfunc):
        self.authenticate = AuthDigestAuthenticator(realm, authfunc)
        self.application = application

    def __call__(self, environ, start_response):
        username = REMOTE_USER(environ)
        if not username:
            result = self.authenticate(environ)
            if isinstance(result, str):
                AUTH_TYPE.update(environ,'digest')
                REMOTE_USER.update(environ, result)
            else:
                return result.wsgi_application(environ, start_response)
        return self.application(environ, start_response)

middleware = AuthDigestHandler

__all__ = ['digest_password', 'AuthDigestHandler' ]

def make_digest(app, global_conf, realm, authfunc, **kw):
    """
    Grant access via digest authentication

    Config looks like this::

      [filter:grant]
      use = egg:Paste#auth_digest
      realm=myrealm
      authfunc=somepackage.somemodule:somefunction

    """
    from paste.util.import_string import eval_import
    import types
    authfunc = eval_import(authfunc)
    assert isinstance(authfunc, types.FunctionType), "authfunc must resolve to a function"
    return AuthDigestHandler(app, realm, authfunc)

if "__main__" == __name__:
    import doctest
    doctest.testmod(optionflags=doctest.ELLIPSIS)