Add custom logger, update unit tests.

Author: Ben Weaver

sasl/__init__.py

@@ -1,4 +1,5 @@
 from __future__ import absolute_import
+from .prelude import *
 from .mechanism import *
 from .auth import *
 from .plain import *

sasl/digest_md5.py

@@ -8,8 +8,7 @@
 
 from __future__ import absolute_import
 import re, functools, random, struct, base64, hashlib, binascii
-import logging as log
-from . import mechanism as mech, rfc, auth
+from . import mechanism as mech, rfc, auth, log
 
 __all__ = ('DigestMD5', 'DigestMD5Password')
 

sasl/mechanism.py

@@ -72,6 +72,9 @@ def failure(self):
     def confirm(self):
         return self.k is None
 
+    def finished(self):
+        return not(callable(self.k))
+
 
 ### Registry
 

sasl/prelude.py

@@ -0,0 +1,12 @@
+## Copyright (c) 2010, Coptix, Inc.  All rights reserved.
+## See the LICENSE file for license terms and warranty disclaimer.
+
+"""prelude -- extra builtins"""
+
+from __future__ import absolute_import
+import os, logging
+
+__all__ = ('log', )
+
+log = logging.getLogger(os.path.basename(os.path.dirname(__file__)))
+log.addHandler(logging.StreamHandler())

sasl/rfc.py

@@ -275,8 +275,12 @@ def __init__(self, kind):
         self.write = kind.write
 
     def read(self, data, pos=0):
-        if data[pos] != '"':
-            return self.kind.read(data, pos)
+        try:
+            if data[pos] != '"':
+                return self.kind.read(data, pos)
+        except IndexError:
+            raise BadToken(self, data, pos)
+
         try:
             (quoted, new_pos) = require(quoted_string, data, pos)
             (value, val_pos) = require(self.kind, quoted, 0)

sasl/tests.py

@@ -76,7 +76,9 @@ class TestMech(object):
     MECH = None
 
     def setUp(self):
+        import logging
         self.mech = self.MECH(AUTH)
+        log.setLevel(logging.CRITICAL)
 
     def test_success(self):
         (sk, ck) = self.negotiate('foo@bar.com', 'baz')
@@ -88,11 +90,7 @@ def test_failure(self):
         self.assertFalse(self.coalesce(sk, ck))
 
     def coalesce(self, value, default):
-        """A value of None indicates the decision about whether
-        authentication succeeded is left up to the other end of the
-        exchange."""
-
-        return default if value is None else value
+        return default.success() if value.confirm() else value.success()
 
     def negotiate(self, user, passwd):
         """Normally this negotiation would take place over a network.
@@ -107,26 +105,18 @@ def negotiate(self, user, passwd):
         of the exchange."""
 
         ## Server issues a challenge.
-        (sk, sdata) = self.mech.challenge()
+        sk = self.mech.challenge()
 
         ## Client responds.
         with fluid.let((USER, user), (PASS, passwd)):
-            (ck, cdata) = self.mech.respond(sdata)
-
-        ## Server and client continue the exchange until they're
-        ## satisfied.
-        while callable(sk) or callable(ck):
-            if callable(sk):
-                self.assert_(cdata is not None)
-                (sk, sdata) = sk(cdata)
-            else:
-                sdata = None
-
-            if callable(ck):
-                self.assert_(sdata is not None)
-                (ck, cdata) = ck(sdata)
-            else:
-                cdata = None
+            ck = self.mech.respond(sk.data)
+
+        while not (sk.finished() and ck.finished()):
+            if not sk.finished():
+                sk = sk(ck.data)
+
+            if not ck.finished():
+                ck = ck(sk.data)
 
         return (sk, ck)
 
@@ -139,8 +129,8 @@ class TestDigestMD5(TestMech, unittest.TestCase):
     def test_challenge(self):
         """Test basic expectations about a challenge."""
 
-        (_, data) = self.mech.challenge()
-        challenge = dict(rfc.data(self.mech.CHALLENGE, data))
+        state = self.mech.challenge()
+        challenge = dict(rfc.data(self.mech.CHALLENGE, state.data))
 
         ## Nonce is random, so it can't be compared for equality.
         self.assert_(challenge.pop('nonce'))
@@ -154,10 +144,10 @@ def test_challenge(self):
     def test_response(self):
         """Test basic expectations about a response."""
 
-        (_, data) = self.mech.challenge()
+        state = self.mech.challenge()
         with fluid.let((USER, 'user@example.net'), (PASS, 'secret')):
-            (_, data) = self.mech.respond(data)
-            resp = dict(rfc.data(self.mech.RESPOND, data))
+            rstate = self.mech.respond(state.data)
+            resp = dict(rfc.data(self.mech.RESPOND, rstate.data))
 
         self.assertNotEqual(resp['nonce'], resp['cnonce'])