summaryrefslogtreecommitdiff
path: root/wsgitools/digest.py
blob: f6792b9b384436d2ae045fef8b406fa31d2e1f3d (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
#!/usr/bin/env python2.5

__all__ = []

import random
try:
    from hashlib import md5
except ImportError:
    from md5 import md5
import binascii
import base64
import time

sysrand = random.SystemRandom()

def gen_rand_str(bytes=33):
    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=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')]
    >>> try:
    ...     parse_digest_response('spam')
    ... except ValueError:
    ...     print("ValueError")
    ValueError
    """
    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 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
        @param getpass: this function is called with a username and password is
                expected as result. C{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("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
    """
    return "%13X" % 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
    """
    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.
        @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
        @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
    """
    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.
        @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
        @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("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, 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
                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
        @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
        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 = "<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]