a sketch of HTTP Digest Mixin for Tornado

gistfile1.py

# -*- coding: utf-8 -*-

"""
Incomplete RFC 2617 implementation for Tornado web server [1], originally
implemeted as `curtain` by Brian K. Jones [2].

[1] http://tornadoweb.org
[2] http://github.com/bkjones/curtain
"""

import os
import sys
import time

if sys.version_info >= (2, 5):
    import hashlib
    md5 = hashlib.md5
else:
    import md5 as _md5
    md5 = md5.new

from werkzeug.datastructures import WWWAuthenticate
from werkzeug.http import parse_authorization_header


class HTTPDigestMixin(object):
    """Mixin implementing HTTP Digest authentication.

    Not: subclasses may override get_current_user() method, to fetch
    additional user related data.
    """
    def response(self, credentials, password):
        # No matter what "qop" value, H(A1) and H(A2) are allways
        # there.
        data = [HA1(credentials, password)]

        if credentials.get("qop") in ("auth", "auth-int"):
            data.extend(credentials.get(k)
                        for k in ("nonce", "nc", "cnonce", "qop"))
        else:
            data.extend(credentials.nonce, HA2(credentials, self.request))

        data.append(HA2(credentials, self.request))

        # Note: we don't have an explicit KD() function, but just so
        # you feel comfortable, here's it's signature from the RFC :)
        # KD(secret, data) = H(concat(secret, ":", data))
        return H(":".join(data))

    def challenge(self):
        # Generating server nonce, format propsed by RFC 2069 is
        # H(client-IP:time-stamp:private-key).
        nonce = md5("%s:%d:%s" % (self.request.remote_ip,
                                  time.time(),
                                  os.urandom(10))).hexdigest()

        # A string of data, specified by the server, which should be
        # returned by the client unchanged.
        opaque = md5(os.urandom(10)).hexdigest()

        auth = WWWAuthenticate("digest")
        auth.set_digest(self.application.settings["auth_realm"],
                        nonce, opaque=opaque)

        self.set_status(401)
        self.set_header("WWW-Authenticate", auth.to_header())

    def get_authenticated_user(self, callback):
        self.require_setting("auth_realm")

        credentials = parse_authorization_header(
            self.request.headers.get("Authorization"))

        if credentials:
            password = callback(credentials.get("username"))
            if (password and
                self.response(credentials, password) == credentials.get("response")):
                return {"username": credentials.get("username")}

        self.challenge()


# Santa's little helpers.

def H(data):
    return md5(data).hexdigest()

def HA1(credentials, password):
    # Note: currenly "MD5" is the only supported algorithm, someday
    # "MD5-sess" will be there too :)
    return H("%s:%s:%s" % (credentials.username,
                           credentials.realm,
                           password))

def HA2(credentials, request):
    if not credentials.qop or credentials.get("qop") == "auth":
        return H("%s:%s" % (request.method, request.uri))
    elif credentials.get("qop") == "auth-int":
        return H("%s:%s:%s" %(request.method,
                              request.uri,
                              H(request.body)))

    # Now, where exactly is that qop value mentioned in RFC 2617?
    raise ValueError