#!/usr/bin/env python2.5
"""
This module contains an C{AuthDigestMiddleware} for authenticating HTTP
requests using the method described in RFC2617. The credentials are to be
provided using an C{AuthTokenGenerator} or a compatible instance. Furthermore
digest authentication has to preserve some state across requests, more
specifically nonces. There are three different C{NonceStoreBase}
implementations for different needs. While the C{StatelessNonceStore} has
minimal requirements it only prevents replay attacks in a limited way. If the
WSGI server uses threading or a single process the C{MemoryNonceStore} can be
used. If that is not possible the nonces can be stored in a DBAPI2 compatible
database using C{DBAPI2NonceStore}.
"""
__all__ = []
import random
try:
from hashlib import md5
except ImportError:
from md5 import md5
import binascii
import base64
import time
import os
sysrand = random.SystemRandom()
def gen_rand_str(bytes=33):
"""
Generates a string of random base64 characters.
@param bytes: is the number of random 8bit values to be used
>>> gen_rand_str() != gen_rand_str()
True
"""
randnum = sysrand.getrandbits(bytes*8)
randstr = ("%%0%dX" % (2*bytes)) % randnum
randstr = binascii.unhexlify(randstr)
randstr = base64.encodestring(randstr).strip()
return randstr
def parse_digest_response(data, ret=None):
"""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')]
>>> try:
... parse_digest_response('spam')
... except ValueError:
... print("ValueError")
ValueError
>>> try:
... parse_digest_response('spam="egg"error')
... except ValueError:
... print("ValueError")
ValueError
"""
assert isinstance(data, str)
if ret is None:
ret = {}
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):
"""
Internal Exception class that is thrown inside L{AuthDigestMiddleware}, but
not visible to other code.
"""
__all__.append("AbstractTokenGenerator")
class AbstractTokenGenerator:
"""Interface class for generating authentication tokens for
L{AuthDigestMiddleware}.
@ivar realm: is a string according to RFC2617.
@type realm: str
"""
def __init__(self, realm):
"""
@type realm: str
"""
assert isinstance(realm, str)
self.realm = realm
def __call__(self, username, algo="md5"):
"""Generates an authentication token from a username.
@type username: str
@type algo: str
@param algo: currently the only value supported by
AuthDigestMiddleware is "md5"
@rtype: str or None
@returns: a valid token or None to signal that authentication should
fail
"""
raise NotImplementedError
def check_password(self, username, password, environ=None):
"""
This function implements the interface for verifying passwords
used by L{BasicAuthMiddleware}. It works by computing a token
from the user and comparing it to the token returned by the
__call__ method.
@type username: str
@type password: str
@param environ: ignored
@rtype: bool
"""
assert isinstance(username, str)
assert isinstance(password, str)
token = md5("%s:%s:%s" % (username, self.realm, password)).hexdigest()
return token == self(username)
__all__.append("AuthTokenGenerator")
class AuthTokenGenerator(AbstractTokenGenerator):
"""Generates authentication tokens for L{AuthDigestMiddleware}. The
interface consists of beeing callable with a username and having a
realm attribute being a string."""
def __init__(self, realm, getpass):
"""
@type realm: str
@param realm: is a string according to RFC2617.
@type getpass: str -> (str or None)
@param getpass: this function is called with a username and password is
expected as result. C{None} may be used as an invalid password.
An example for getpass would be C{{username: password}.get}.
"""
AbstractTokenGenerator.__init__(self, realm)
self.getpass = getpass
def __call__(self, username, algo="md5"):
assert isinstance(username, 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("HtdigestTokenGenerator")
class HtdigestTokenGenerator(AbstractTokenGenerator):
"""Reads authentication tokens for L{AuthDigestMiddleware} from an
apache htdigest file.
"""
def __init__(self, realm, htdigestfile, ignoreparseerrors=False):
"""
@type realm: str
@type htdigestfile: str
@type ignoreparseerrors: bool
@param ignoreparseerrors: passed to readhtdigest
@raises IOError:
@raises ValueError:
"""
AbstractTokenGenerator.__init__(self, realm)
self.users = {}
self.readhtdigest(htdigestfile, ignoreparseerrors)
def readhtdigest(self, htdigestfile, ignoreparseerrors=False):
"""
@type htdigestfile: str
@type ignoreparseerrors: bool
@param ignoreparseerrors: do not raise ValueErrors for bad files
@raises IOError:
@raises ValueError:
"""
assert isinstance(htdigestfile, str)
self.users = {}
for line in file(htdigestfile):
parts = line.rstrip("\n").split(":")
if len(parts) != 3:
if ignoreparseerrors:
continue
raise ValueError("invalid number of colons in htdigest file")
user, realm, token = parts
if realm != self.realm:
continue
if user in self.users and not ignoreparseerrors:
raise ValueError("duplicate user in htdigest file")
self.users[user] = token
def __call__(self, user, algo="md5"):
assert algo.lower() in ["md5", "md5-sess"]
return self.users.get(user)
__all__.append("UpdatingHtdigestTokenGenerator")
class UpdatingHtdigestTokenGenerator(HtdigestTokenGenerator):
"""Behaves like HtdigestTokenGenerator, checks the htdigest file
for changes on each invocation.
"""
def __init__(self, realm, htdigestfile, ignoreparseerrors=False):
assert isinstance(htdigestfile, str)
try:
self.statcache = os.stat(htdigestfile)
except OSError, err:
raise IOError(str(err))
HtdigestTokenGenerator.__init__(self, realm, htdigestfile,
ignoreparseerrors)
self.htdigestfile = htdigestfile
self.ignoreparseerrors = ignoreparseerrors
def __call__(self, user, algo="md5"):
try:
statcache = os.stat(self.htdigestfile)
except OSError:
return None
if self.statcache != statcache:
try:
self.readhtdigest(self.htdigestfile, self.ignoreparseerrors)
except (IOError, ValueError):
return None
return HtdigestTokenGenerator.__call__(self, user, algo)
__all__.append("NonceStoreBase")
class NonceStoreBase:
"""Nonce storage interface."""
def __init__(self):
pass
def newnonce(self, ident=None):
"""
This method is to be overriden and should return new nonces.
@type ident: str
@param ident: is an identifier to be associated with this nonce
@rtype: str
"""
raise NotImplementedError
def checknonce(self, nonce, count=1, ident=None):
"""
This method is to be overridden and should do a check for whether the
given nonce is valid as being used count times.
@type nonce: str
@type count: int
@param count: indicates how often the nonce has been used (including
this check)
@type ident: str
@param ident: it is also checked that the nonce was associated to this
identifier when given
@rtype: bool
"""
raise NotImplementedError
def format_time(seconds):
"""
internal method formatting a unix time to a fixed-length string
@type seconds: float
@rtype: str
"""
# the overflow will happen about 2112
return "%013X" % int(seconds * 1000000)
__all__.append("StatelessNonceStore")
class StatelessNonceStore(NonceStoreBase):
"""
This is a stateless nonce storage that cannot check the usage count for
a nonce and thus cannot protect against replay attacks. It however can make
it difficult by posing a timeout on nonces and making it difficult to forge
nonces.
This nonce store is usable with L{scgi.forkpool}.
>>> s = StatelessNonceStore()
>>> n = s.newnonce()
>>> s.checknonce("spam")
False
>>> s.checknonce(n)
True
>>> s.checknonce(n)
True
>>> s.checknonce(n.rsplit(':', 1)[0] + "bad hash")
False
"""
def __init__(self, maxage=300, secret=None):
"""
@type maxage: int
@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.
@type secret: str
@param secret: if not given, a secret is generated and is therefore
shared after forks. Knowing this secret permits creating nonces.
"""
NonceStoreBase.__init__(self)
self.maxage = maxage
if secret:
self.server_secret = secret
else:
self.server_secret = gen_rand_str()
def newnonce(self, ident=None):
"""
Generates a new nonce string.
@type ident: None or str
@rtype: str
"""
nonce_time = format_time(time.time())
nonce_value = gen_rand_str()
token = "%s:%s:%s" % (nonce_time, nonce_value, self.server_secret)
if ident is not None:
token = "%s:%s" % (token, ident)
token = md5(token).hexdigest()
return "%s:%s:%s" % (nonce_time, nonce_value, token)
def checknonce(self, nonce, count=1, ident=None):
"""
Do a check for whether the provided string is a nonce and increase usage
count on returning True.
@type nonce: str
@type count: int
@type ident: None or str
@rtype: bool
"""
if count != 1:
return False
try:
nonce_time, nonce_value, nonce_hash = nonce.split(':')
except ValueError:
return False
token = "%s:%s:%s" % (nonce_time, nonce_value, self.server_secret)
if ident is not None:
token = "%s:%s" % (token, ident)
token = md5(token).hexdigest()
if token != nonce_hash:
return False
if nonce_time < format_time(time.time() - self.maxage):
return False
return True
__all__.append("MemoryNonceStore")
class MemoryNonceStore(NonceStoreBase):
"""
Simple in-memory mechanism to store nonces.
>>> s = MemoryNonceStore(maxuses=1)
>>> n = s.newnonce()
>>> s.checknonce("spam")
False
>>> s.checknonce(n)
True
>>> s.checknonce(n)
False
>>> n = s.newnonce()
>>> s.checknonce(n.rsplit(':', 1)[0] + "bad hash")
False
"""
def __init__(self, maxage=300, maxuses=5):
"""
@type maxage: int
@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.
@type maxuses: int
@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.
"""
NonceStoreBase.__init__(self)
self.maxage = maxage
self.maxuses = maxuses
self.nonces = [] # [(creation_time, nonce_value, useage_count)]
# as [(float, str, int)]
self.server_secret = gen_rand_str()
def _cleanup(self):
"""internal methods cleaning list of valid nonces"""
old = format_time(time.time() - self.maxage)
while self.nonces and self.nonces[0][0] < old:
self.nonces.pop(0)
def newnonce(self, ident=None):
"""
Generates a new nonce string.
@type ident: None or str
@rtype: str
"""
self._cleanup() # avoid growing self.nonces
nonce_time = format_time(time.time())
nonce_value = gen_rand_str()
self.nonces.append((nonce_time, nonce_value, 1))
token = "%s:%s:%s" % (nonce_time, nonce_value, self.server_secret)
if ident is not None:
token = "%s:%s" % (token, ident)
token = md5(token).hexdigest()
return "%s:%s:%s" % (nonce_time, nonce_value, token)
def checknonce(self, nonce, count=1, ident=None):
"""
Do a check for whether the provided string is a nonce and increase usage
count on returning True.
@type nonce: str
@type count: int
@type ident: None or str
@rtype: bool
"""
try:
nonce_time, nonce_value, nonce_hash = nonce.split(':')
except ValueError:
return False
token = "%s:%s:%s" % (nonce_time, nonce_value, self.server_secret)
if ident is not None:
token = "%s:%s" % (token, ident)
token = md5(token).hexdigest()
if token != nonce_hash:
return False
self._cleanup() # avoid stale nonces
# 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
if len(self.nonces) <= lower:
return False
(nt, nv, uses) = self.nonces[lower]
if nt != nonce_time or nv != nonce_value:
return False
if count != 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
__all__.append("LazyDBAPI2Opener")
class LazyDBAPI2Opener:
"""
Connects to database on first request. Otherwise it behaves like a dbapi2
connection. This may be usefull in combination with scgi.forkpool, because
this way each worker child opens a new database connection when the first
request is to be answered.
"""
def __init__(self, function, *args, **kwargs):
"""
The database will be connected on the first method call. This is done
by calling the given function with the remaining parameters.
@param function: is the function that connects to the database
"""
self._function = function
self._args = args
self._kwargs = kwargs
self._dbhandle = None
def _getdbhandle(self):
"""Returns an open database connection. Open if necessary."""
if self._dbhandle is None:
self._dbhandle = self._function(*self._args, **self._kwargs)
self._function = self._args = self._kwargs = None
return self._dbhandle
def cursor(self):
"""dbapi2"""
return self._getdbhandle().cursor()
def commit(self):
"""dbapi2"""
return self._getdbhandle().commit()
def rollback(self):
"""dbapi2"""
return self._getdbhandle().rollback()
def close(self):
"""dbapi2"""
return self._getdbhandle().close()
__all__.append("DBAPI2NonceStore")
class DBAPI2NonceStore(NonceStoreBase):
"""
A dbapi2-backed nonce store implementation suitable for usage with forking
wsgi servers such as scgi.forkpool.
>>> import sqlite3
>>> db = sqlite3.connect(":memory:")
>>> db.cursor().execute("CREATE TABLE nonces (key, value);") and None
>>> db.commit() and None
>>> s = DBAPI2NonceStore(db, maxuses=1)
>>> n = s.newnonce()
>>> s.checknonce("spam")
False
>>> s.checknonce(n)
True
>>> s.checknonce(n)
False
>>> n = s.newnonce()
>>> s.checknonce(n.rsplit(':', 1)[0] + "bad hash")
False
"""
def __init__(self, dbhandle, maxage=300, maxuses=5, table="nonces"):
"""
@param dbhandle: is a dbapi2 connection
@type maxage: int
@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.
@type maxuses: int
@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.
"""
NonceStoreBase.__init__(self)
self.dbhandle = dbhandle
self.maxage = maxage
self.maxuses = maxuses
self.table = table
self.server_secret = gen_rand_str()
def _cleanup(self, cur):
"""internal methods cleaning list of valid nonces"""
old = format_time(time.time() - self.maxage)
cur.execute("DELETE FROM %s WHERE key < '%s:';" % (self.table, old))
def newnonce(self, ident=None):
"""
Generates a new nonce string.
@rtype: str
"""
nonce_time = format_time(time.time())
nonce_value = gen_rand_str()
dbkey = "%s:%s" % (nonce_time, nonce_value)
cur = self.dbhandle.cursor()
self._cleanup(cur) # avoid growing database
cur.execute("INSERT INTO %s VALUES ('%s', '1');" % (self.table, dbkey))
self.dbhandle.commit()
token = "%s:%s" % (dbkey, self.server_secret)
if ident is not None:
token = "%s:%s" % (token, ident)
token = md5(token).hexdigest()
return "%s:%s:%s" % (nonce_time, nonce_value, token)
def checknonce(self, nonce, count=1, ident=None):
"""
Do a check for whether the provided string is a nonce and increase usage
count on returning True.
@type nonce: str
@type count: int
@rtype: bool
"""
try:
nonce_time, nonce_value, nonce_hash = nonce.split(':')
except ValueError:
return False
if not nonce_time.isalnum() or not nonce_value.replace("+", ""). \
replace("/", "").replace("=", "").isalnum():
return False
token = "%s:%s:%s" % (nonce_time, nonce_value, self.server_secret)
if ident is not None:
token = "%s:%s" % (token, ident)
token = md5(token).hexdigest()
if token != nonce_hash:
return False
if nonce_time < format_time(time.time() - self.maxage):
return False
cur = self.dbhandle.cursor()
#self._cleanup(cur) # avoid growing database
dbkey = "%s:%s" % (nonce_time, nonce_value)
cur.execute("SELECT value FROM %s WHERE key = '%s';" %
(self.table, dbkey))
uses = cur.fetchone()
if uses is None:
self.dbhandle.commit()
return False
uses = int(uses[0])
if count != uses:
cur.execute("DELETE FROM %s WHERE key = '%s';" %
(self.table, dbkey))
self.dbhandle.commit()
return False
if uses >= self.maxuses:
cur.execute("DELETE FROM %s WHERE key = '%s';" %
(self.table, dbkey))
else:
cur.execute("UPDATE %s SET value = '%d' WHERE key = '%s';" %
(self.table, uses + 1, dbkey))
self.dbhandle.commit()
return True
__all__.append("AuthDigestMiddleware")
class AuthDigestMiddleware:
"""Middleware partly implementing RFC2617. (md5-sess was omited)
Upon successful authentication the environ dict will be extended
by a REMOTE_USER key before being passed to the wrapped
application."""
algorithms = {"md5": lambda data: md5(data).hexdigest()}
def __init__(self, app, gentoken, maxage=300, maxuses=5, store=None):
"""
@param app: is the wsgi application to be served with authentification.
@type gentoken: str -> (str or None)
@param gentoken: has to have the same functionality and interface as the
L{AuthTokenGenerator} class.
@type maxage: int
@param maxage: deprecated, see L{MemoryNonceStore} or
L{StatelessNonceStore}
@type maxuses: int
@param maxuses: deprecated, see L{MemoryNonceStore}
@type store: L{NonceStoreBase}
@param store: a nonce storage implementation object. Usage of this
parameter will override maxage and maxuses.
"""
self.app = app
self.gentoken = gentoken
if store is None:
self.noncestore = MemoryNonceStore(maxage, maxuses)
else:
assert hasattr(store, "newnonce")
assert hasattr(store, "checknonce")
self.noncestore = store
def __call__(self, environ, start_response):
"""wsgi interface"""
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
noncecount = 1
if credentials.get("qop") is not None:
# raises ValueError
noncecount = int(credentials["nc"], 16)
if not self.noncestore.checknonce(credentials["nonce"], noncecount):
return self.authorization_required(environ, start_response,
stale=True) # stale nonce!
# raises KeyError, ValueError
response = self.auth_response(credentials,
environ["REQUEST_METHOD"])
if response != credentials["response"]:
raise AuthenticationRequired
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.noncestore.newnonce())
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
@raises KeyError:
@raises ValueError:
"""
username = credentials["username"]
algo = credentials["algorithm"]
uri = credentials["uri"]
nonce = credentials["nonce"]
a1h = self.gentoken(username, algo)
if a1h is None:
raise ValueError
a2h = self.algorithms[algo]("%s:%s" % (reqmethod, uri))
qop = credentials.get("qop", None)
if qop is None:
dig = ":".join((a1h, nonce, a2h))
else:
if qop != "auth":
raise ValueError
# raises KeyError
dig = ":".join((a1h, nonce, credentials["nc"],
credentials["cnonce"], qop, a2h))
return self.algorithms[algo](dig)
def authorization_required(self, environ, start_response, stale=False):
"""internal method implementing wsgi interface, serving 401 page"""
digest = dict(nonce=self.noncestore.newnonce(),
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 = "401 Not authorized"
data += "401 Not authorized
"
headers.append(("Content-length", str(len(data))))
start_response(status, headers)
if environ["REQUEST_METHOD"] == "HEAD":
return []
return [data]