path: root/wsgitools/digest.py

#!/usr/bin/env python2.5

__all__ = []

import random
try:
    from hashlib import md5
except ImportError:
    from md5 import md5

import time

sysrand = random.SystemRandom()

def parse_digest_response(data, ret=dict()):
    """internal
    @raises ValueError:

    >>> parse_digest_response('foo=bar')
    {'foo': 'bar'}
    >>> parse_digest_response('foo="bar"')
    {'foo': 'bar'}
    >>> sorted(parse_digest_response('foo="bar=qux",spam=egg').items())
    [('foo', 'bar=qux'), ('spam', 'egg')]
    """
    data = data.strip()
    key, rest = data.split('=', 1) # raises ValueError
    if rest.startswith('"'):
        rest = rest[1:]
        value, rest = rest.split('"', 1) # raises ValueError
        if not rest:
            ret[key] = value
            return ret
        if rest[0] != ',':
            raise ValueError, "invalid digest response"
        rest = rest[1:]
    else:
        if ',' not in rest:
            ret[key] = rest
            return ret
        value, rest = rest.split(',' , 1)
    ret[key] = value
    return parse_digest_response(rest, ret)

class AuthenticationRequired(Exception):
    pass

__all__.append("AuthTokenGenerator")
class AuthTokenGenerator:
    """Generates authentification tokens for AuthDigestMiddleware. The
    interface consists of beeing callable with a username and having a
    realm attribute being a string."""
    def __init__(self, realm, getpass):
        """
        @param realm: is a string according to RFC2617.
        @param getpass: this function is called with a username and password is
                expected as result. None may be used as an invalid password."""
        self.realm = realm
        self.getpass = getpass
    def __call__(self, username, algo="md5"):
        """Generates an authentification token from a username.
        @type username: str
        @rtype: str
        """
        assert algo.lower() in ["md5", "md5-sess"]
        password = self.getpass(username)
        if password is None:
            return None
        a1 = "%s:%s:%s" % (username, self.realm, password)
        return md5(a1).hexdigest()

__all__.append("AuthDigestMiddleware")
class AuthDigestMiddleware:
    """Middleware partly implementing RFC2617. (md5-sess was omited)"""
    algorithms = {"md5": lambda data: md5(data).hexdigest()}
    def __init__(self, app, gentoken, maxage=300, maxuses=5):
        """
        @param app: is the wsgi application to be served with authentification.
        @param gentoken: has to have the same functionality and interface as the
                 AuthTokenGenerator class.
        @param maxage: is the number of seconds a nonce may be valid. Choosing a
               large value may result in more memory usage whereas a smaller
               value results in more requests. Defaults to 5 minutes.
        @param maxuses: is the number of times a nonce may be used (with
                different nc values). A value of 1 makes nonces usable exactly
                once resulting in more requests. Defaults to 5.
        """
        self.app = app
        self.gentoken = gentoken
        self.maxage = maxage
        self.maxuses = maxuses
        self.server_secret = ("%066X" % sysrand.getrandbits(33*8)
                              ).decode("hex").encode("base64").strip()
        self.nonces = [] # [(creation_time, nonce_value, useage_count)]
                         # as [(float, str, int)]

    def __call__(self, environ, start_response):
        """wsgi interface"""
        self.cleanup_nonces()

        try:
            auth = environ["HTTP_AUTHORIZATION"] # raises KeyError
            method, rest = auth.split(' ', 1) # raises ValueError

            if method.lower() != "digest":
                raise AuthenticationRequired
            credentials = parse_digest_response(rest) # raises ValueError

            ### Check algorithm field
            credentials["algorithm"] = credentials.get("algorithm",
                                                       "md5").lower()
            if not credentials["algorithm"] in self.algorithms:
                raise AuthenticationRequired

            ### Check uri field
            # Doing this by stripping known parts from the passed uri field
            # until something trivial remains, as the uri cannot be
            # reconstructed from the environment exactly.
            uri = credentials["uri"] # raises KeyError
            if "QUERY_STRING" in environ and environ["QUERY_STRING"]:
                if not uri.endswith(environ["QUERY_STRING"]):
                    raise AuthenticationRequired
                uri = uri[:-len(environ["QUERY_STRING"])]
            if "SCRIPT_NAME" in environ:
                if not uri.startswith(environ["SCRIPT_NAME"]):
                    raise AuthenticationRequired
                uri = uri[len(environ["SCRIPT_NAME"]):]
            if "PATH_INFO" in environ:
                if not uri.startswith(environ["PATH_INFO"]):
                    raise AuthenticationRequired
                uri = uri[len(environ["PATH_INFO"]):]
            if uri not in ('', '?'):
                raise AuthenticationRequired
            del uri

            if ("username" not in credentials or
                    "nonce" not in credentials or
                    "response" not in credentials or
                    "qop" in credentials and (
                        credentials["qop"] != "auth" or 
                        "nc" not in credentials or
                        credentials["nc"].lower().strip("0123456789abcdef") or
                        "cnonce" not in credentials)):
                raise AuthenticationRequired

            if not self.is_nonce(credentials): # riases KeyError, ValueError
                raise AuthenticationRequired

            # raises KeyError, ValueError
            response = self.auth_response(credentials,
                                          environ["REQUEST_METHOD"])
            if response is None or response != credentials["response"]:
                raise AuthenticationRequired

            if not self.check_nonce(credentials): # raises KeyError, ValueError
                return self.authorization_required(environ, start_response,
                                                   stale=True) # stale nonce!

        except (KeyError, ValueError, AuthenticationRequired):
            return self.authorization_required(environ, start_response)
        else:
            environ["REMOTE_USER"] = credentials["username"]
            def modified_start_response(status, headers, exc_info=None):
                digest = dict(nextnonce=self.new_nonce())
                if "qop" in credentials:
                    digest["qop"] = "auth"
                    digest["cnonce"] = credentials["cnonce"] # no KeyError
                    digest["rspauth"] = self.auth_response(credentials, "")
                challenge = ", ".join(map('%s="%s"'.__mod__, digest.items()))
                headers.append(("Authentication-Info", challenge))
                return start_response(status, headers, exc_info)
            return self.app(environ, modified_start_response)

    def auth_response(self, credentials, reqmethod):
        """internal method generating authentication tokens
        @raise KeyError:
        @raise ValueError:
        """
        username = credentials["username"]
        algo = credentials["algorithm"]
        uri = credentials["uri"]
        nonce = credentials["nonce"]
        a1h = self.gentoken(username, algo)
        if a1h is None:
            raise ValueError
        a2 = "%s:%s" % (reqmethod, uri)
        a2h = self.algorithms[algo](a2)
        qop = credentials.get("qop", None)
        if qop is None:
            dig = ":".join((a1h, nonce, a2h))
        else:
            nc = credentials["nc"] # raises KeyError
            cnonce = credentials["cnonce"] # raises KeyError
            if qop != "auth":
                return ValueError
            dig =  ":".join((a1h, nonce, nc, cnonce, qop, a2h))
        digh = self.algorithms[algo](dig)
        return digh

    def cleanup_nonces(self):
        """internal methods cleaning list of valid nonces"""
        # see new_nonce
        old = "%13X" % long((time.time() - self.maxage) * 1000000)
        while self.nonces and self.nonces[0][0] < old:
            self.nonces.pop(0)

    def is_nonce(self, credentials):
        """internal method checking whether a nonce might be from this server
        @raise KeyError:
        @raise ValueError:
        """
        nonce = credentials["nonce"] # raises KeyError
        # raises ValueError
        nonce_time, nonce_value, nonce_hash = nonce.split(':')
        token = "%s:%s:%s" % (nonce_time, nonce_value, self.server_secret)
        token = md5(token).hexdigest()
        return nonce_hash == token

    def check_nonce(self, credentials):
        """internal method checking nonce validity
        @raise KeyError:
        @raise ValueError:
        """
        nonce = credentials["nonce"]
        # raises ValueError
        nonce_time, nonce_value, nonce_hash = nonce.split(':')
        token = "%s:%s:%s" % (nonce_time, nonce_value, self.server_secret)
        token = md5(token).hexdigest()
        if token != nonce_hash:
            return False
        qop = credentials.get("qop", None)
        if qop is None:
            nc = 1
        else:
            nc = long(credentials["nc"], 16) # raises KeyError, ValueError
        # searching nonce_time
        lower, upper = 0, len(self.nonces) - 1
        while lower < upper:
            mid = (lower + upper) // 2
            if nonce_time <= self.nonces[mid][0]:
                upper = mid
            else:
                lower = mid + 1

        (nt, nv, uses) = self.nonces[lower]
        if nt != nonce_time or nv != nonce_value:
            return False
        if nc != uses:
            del self.nonces[lower]
            return False
        if uses >= self.maxuses:
            del self.nonces[lower]
        else:
            self.nonces[lower] = (nt, nv, uses+1)
        return True

    def new_nonce(self):
        """internal method generating a new nonce"""
        # 13 = (32 bit + 20 bit) / 4
        nonce_time = "%13X" % long(time.time() * 1000000)
        randval = sysrand.getrandbits(33*8)
        nonce_value = ("%066X" % randval).decode("hex").encode("base64").strip()
        self.nonces.append((nonce_time, nonce_value, 1))
        token = "%s:%s:%s" % (nonce_time, nonce_value, self.server_secret)
        token = md5(token).hexdigest()
        return "%s:%s:%s" % (nonce_time, nonce_value, token)

    def authorization_required(self, environ, start_response, stale=False):
        """internal method implementing wsgi interface, serving 401 page"""
        nonce = self.new_nonce()
        digest = dict(nonce=nonce,
                      realm=self.gentoken.realm,
                      algorithm="md5",
                      qop="auth")
        if stale:
            digest["stale"] = "TRUE"
        challenge = ", ".join(map('%s="%s"'.__mod__, digest.items()))
        status = "401 Not authorized"
        headers = [("Content-type", "text/html"),
                   ("WWW-Authenticate", "Digest %s" % challenge)]
        data = "<html><head><title>401 Not authorized</title></head><body><h1>"
        data += "401 Not authorized</h1></body></html>"
        headers.append(("Content-length", str(len(data))))
        start_response(status, headers)
        if environ["REQUEST_METHOD"] == "HEAD":
            return []
        return [data]