From 3a37a9166b6e6f8f2bad86961ed263572ee470b9 Mon Sep 17 00:00:00 2001
From: Helmut Grohne <helmut@subdivi.de>
Date: Sun, 21 Sep 2008 17:42:37 +0200
Subject: digest: create a generic nonce storage interface
---
wsgitools/digest.py | 333 ++++++++++++++++++++++++++++++++++++++--------------
1 file changed, 247 insertions(+), 86 deletions(-)
diff --git a/wsgitools/digest.py b/wsgitools/digest.py
index 65560f6..8c5bc0f 100755
--- a/wsgitools/digest.py
+++ b/wsgitools/digest.py
@@ -52,6 +52,7 @@ class AuthTokenGenerator:
realm attribute being a string."""
def __init__(self, realm, getpass):
"""
+ @type realm: str
@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."""
@@ -69,34 +70,260 @@ class AuthTokenGenerator:
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):
+__all__.append("NonceStoreBase")
+class NonceStoreBase:
+ """Nonce storage interface."""
+ def __init__(self):
+ pass
+ def newnonce(self):
"""
- @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.
+ This method is to be overriden and should return new nonces.
+ @rtype: str
+ """
+ raise NotImplementedError
+ def isnonce(self, nonce):
+ """
+ This method is to be overridden and should do a quick check for whether
+ the given nonce has a chance to be a valid one. This function must not
+ return false for a stale nonce.
+ @type nonce: str
+ @rtype: bool
+ """
+ raise NotImplementedError
+ def checknonce(self, nonce, qop, nc):
+ """
+ This method is to be overridden and should do a thorough check for
+ whether the given nonce is a valid one taking qop and nc into account.
+ @type nonce: str
+ @type qop: str or None
+ @type nc: str or None
+ @rtype: bool
+ """
+ raise NotImplementedError
+
+def format_time(seconds):
+ """
+ internal method formatting a unix time to a fixed-length string
+ @type seconds: float
+ @rtype: str
+ """
+ return "%13X" % long(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 scgi.forkpool.
+ """
+ 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 = ("%066X" % sysrand.getrandbits(33*8)
+ ).decode("hex").encode("base64").strip()
+
+ def newnonce(self):
+ """
+ Generates a new nonce string.
+ @rtype: str
+ """
+ nonce_time = format_time(time.time())
+ nonce_value = ("%066X" % sysrand.getrandbits(33*8)
+ ).decode("hex").encode("base64").strip()
+ 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 isnonce(self, nonce):
+ """
+ Do a quick a stateless check for whether the provides string might
+ be a nonce.
+ @type nonce: 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)
+ token = md5(token).hexdigest()
+ return nonce_hash == token
+
+ def checknonce(self, nonce, qop, nc):
+ """
+ Do a thorough check for whether the provided string is a nonce and
+ increase usage count on returning True.
+ @type nonce: str
+ @type qop: str or None
+ @type nc: str or None
+ @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)
+ 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."""
+ 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.
"""
- self.app = app
- self.gentoken = gentoken
+ NonceStoreBase.__init__(self)
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)]
+ self.server_secret = ("%066X" % sysrand.getrandbits(33*8)
+ ).decode("hex").encode("base64").strip()
+
+ 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):
+ """
+ Generates a new nonce string.
+ @rtype: str
+ """
+ self._cleanup() # avoid growing self.nonces
+ nonce_time = format_time(time.time())
+ nonce_value = ("%066X" % sysrand.getrandbits(33*8)
+ ).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 isnonce(self, nonce):
+ """
+ Do a quick a stateless check for whether the provides string might
+ be a nonce.
+ @type nonce: 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)
+ token = md5(token).hexdigest()
+ return nonce_hash == token
+
+ def checknonce(self, nonce, qop, nc):
+ """
+ Do a thorough check for whether the provided string is a nonce and
+ increase usage count on returning True.
+ @type nonce: str
+ @type qop: str or None
+ @type nc: str or None
+ @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)
+ token = md5(token).hexdigest()
+ if token != nonce_hash:
+ return False
+ if qop is None:
+ nc = 1
+ else:
+ try:
+ nc = long(nc, 16)
+ except (KeyError, ValueError):
+ 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
+
+ (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
+
+__all__.append("AuthDigestMiddleware")
+class AuthDigestMiddleware:
+ """Middleware partly implementing RFC2617. (md5-sess was omited)
+ This middleware does not work with cgi servers or fork servers like
+ scgi.forkpool.
+ """
+ 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
+ @param gentoken: has to have the same functionality and interface as the
+ AuthTokenGenerator class.
+ @type maxage: int
+ @param maxage: deprecated, see MemoryNonceStore or StatelessNonceStore
+ @type maxuses: int
+ @param maxuses: deprecated, see MemoryNonceStore
+ @type store: 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, "isnonce")
+ assert hasattr(store, "checknonce")
+ self.noncestore = store
def __call__(self, environ, start_response):
"""wsgi interface"""
- self.cleanup_nonces()
try:
auth = environ["HTTP_AUTHORIZATION"] # raises KeyError
@@ -143,16 +370,19 @@ class AuthDigestMiddleware:
"cnonce" not in credentials)):
raise AuthenticationRequired
- if not self.is_nonce(credentials): # riases KeyError, ValueError
+ if not self.noncestore.isnonce(credentials["nonce"]):
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
+ if not self.noncestore.checknonce(credentials["nonce"],
+ credentials.get("qop"),
+ credentials.get("nc")):
return self.authorization_required(environ, start_response,
stale=True) # stale nonce!
@@ -161,7 +391,7 @@ class AuthDigestMiddleware:
else:
environ["REMOTE_USER"] = credentials["username"]
def modified_start_response(status, headers, exc_info=None):
- digest = dict(nextnonce=self.new_nonce())
+ digest = dict(nextnonce=self.noncestore.newnonce())
if "qop" in credentials:
digest["qop"] = "auth"
digest["cnonce"] = credentials["cnonce"] # no KeyError
@@ -197,78 +427,9 @@ class AuthDigestMiddleware:
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,
+ digest = dict(nonce=self.noncestore.newnonce(),
realm=self.gentoken.realm,
algorithm="md5",
qop="auth")
--
cgit v1.2.3