Initial implementation; SimpleAuth and PLAIN

Author: Ben Weaver

.gitignore

@@ -0,0 +1 @@
+*.py?

sasl/__init__.py

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

sasl/auth.py

@@ -0,0 +1,54 @@
+from __future__ import absolute_import
+import abc
+
+__all__ = ('Authenticator', 'SimpleAuth')
+
+class Authenticator(object):
+    """A basic authentication interface used by SASL mechanisms."""
+
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractmethod
+    def authentication_id(self):
+        """Identify the entity being authenticated (e.g. username)."""
+
+    @abc.abstractmethod
+    def password(self):
+        """The password associated with the authentication_id."""
+
+    @abc.abstractmethod
+    def verify_password(self, authzid, authcid, passwd):
+        return False
+
+    def authorization_id(self):
+        """Identify the effective entity if authentication
+        succeeds."""
+        return u''
+
+class SimpleAuth(Authenticator):
+    """Authenticate from a Mapping."""
+
+    def __init__(self, entities, authcid, passwd, authzid=None):
+        self.entities = entities
+        self.authcid = authcid
+        self.passwd = passwd
+        self.authzid = authzid
+
+    def authentication_id(self):
+        value = self.authcid()
+        if not value:
+            raise RuntimeError('Undefined authentication entity.')
+        return value
+
+    def password(self):
+        return self.passwd()
+
+    def authorization_id(self):
+        return self.authzid and self.authzid()
+
+    def verify_password(self, authzid, authcid, passwd):
+        try:
+            return self.entities[authcid] == passwd
+        except KeyError:
+            return False
+

sasl/mechanism.py

@@ -0,0 +1,51 @@
+"""mechanism.py -- SASL mechanism registry
+
+<http://tools.ietf.org/html/rfc2222>
+<http://www.iana.org/assignments/sasl-mechanisms>
+"""
+from __future__ import absolute_import
+import abc
+
+__all__ = ('define', 'Mechanism')
+
+MECHANISMS = {}
+
+def define(name=None):
+    """A class decorator that registers a SASL mechanism."""
+
+    def decorator(cls):
+        return register(name or cls.__name__, cls)
+
+    return decorator
+
+def register(name, cls):
+    """Register a SASL mechanism."""
+
+    MECHANISMS[name.upper()] = cls
+    return cls
+
+class MechanismType(abc.ABCMeta):
+    """This metaclass registers a SASL mechanism when it's defined."""
+
+    def __new__(mcls, name, bases, attr):
+        cls = abc.ABCMeta.__new__(mcls, name, bases, attr)
+        return register(name, cls)
+
+class Mechanism(object):
+    """The SASL mechanism interface."""
+
+    __metaclass__ = MechanismType
+    __slots__ = ()
+
+    @abc.abstractmethod
+    def challenge(self):
+        """Issue a challenge."""
+
+    @abc.abstractmethod
+    def respond(self, challenge):
+        """Respond to a challenge."""
+
+    @abc.abstractmethod
+    def verify_challenge(self, response):
+        """Verify a challenge.  Return True if the response was
+        verified.  Return False if the challenge failed."""

sasl/plain.py

@@ -0,0 +1,42 @@
+"""plain -- simple, unencrypted user/password authentication
+
+<http://www.ietf.org/rfc/rfc4616.txt>
+"""
+from __future__ import absolute_import
+from . import mechanism as mech
+
+__all__ = ('Plain', )
+
+class Plain(mech.Mechanism):
+    """The plain mechanism simply submits the optional authorization
+    id, the authentication id, and password separated by null
+    bytes."""
+
+    NULL = u'0x00'
+
+    def __init__(self, auth):
+        self.auth = auth
+
+    def challenge(self):
+        return ''
+
+    def respond(self, data):
+        assert data == ''
+
+        auth = self.auth
+        zid = auth.authorization_id()
+        cid = auth.authentication_id()
+
+        return self.NULL.join((
+            u'' if (not zid or zid == cid) else zid,
+            (cid or u''),
+            (auth.password() or u'')
+        )).encode('utf-8')
+
+    def verify_challenge(self, response):
+        try:
+            (zid, cid, passwd) = response.decode('utf-8').split(self.NULL)
+        except ValueError:
+            return False
+
+        return self.auth.verify_password(zid, cid, passwd)

sasl/tests.py

@@ -0,0 +1,26 @@
+from __future__ import absolute_import
+import unittest
+from md import fluid
+from . import *
+
+USER = fluid.cell()
+PASS = fluid.cell()
+
+class TestPlain(unittest.TestCase):
+
+    def setUp(self):
+        users = { 'foo@bar.com': 'baz' }
+        self.auth = SimpleAuth(users, lambda: USER.value, lambda: PASS.value)
+        self.mech = plain.Plain(self.auth)
+
+    def test_success(self):
+        ch = self.mech.challenge()
+        with fluid.let((USER, 'foo@bar.com'), (PASS, 'baz')):
+            re = self.mech.respond(ch)
+        self.assert_(self.mech.verify_challenge(re))
+
+    def test_failure(self):
+        ch = self.mech.challenge()
+        with fluid.let((USER, 'foo@bar.com'), (PASS, '')):
+            re = self.mech.respond(ch)
+        self.assertFalse(self.mech.verify_challenge(re))