Changeset 27082

Timestamp:

07/02/2009 12:00:19 PM (14 months ago)

Author:

z3p

Message:

merging forward

Location:

branches/spwd-3242-3/twisted/conch

Files:

• checkers.py (modified) (3 diffs)
• test/test_checkers.py (modified) (15 diffs)

branches/spwd-3242-3/twisted/conch/checkers.py

@@ -10 +10 @@
 try:
     import pwd
+    pwd
 except ImportError:
     pwd = None
@@ -16 +17 @@
 
 try:
-    # get this from http://www.twistedmatrix.com/users/z3p/files/pyshadow-0.2.tar.gz
-    import shadow
-except:
-    shadow = None
+    # Python 2.5 got spwd to interface with shadow passwords
+    import spwd as shadow
+    shadow
+except ImportError:
+    try:
+        # get this from http://www.twistedmatrix.com/users/z3p/files/pyshadow-0.2.tar.gz
+        import shadow
+        shadow
+    except ImportError:
+        shadow = None
 
 try:
     from twisted.cred import pamauth
+    pamauth
 except ImportError:
     pamauth = None
@@ -37 +45 @@
 from twisted.python.util import runAsEffectiveUser
 
+
+
 def verifyCryptedPassword(crypted, pw):
-    if crypted[0] == '$': # md5_crypt encrypted
-        salt = '$1$' + crypted.split('$')[2]
-    else:
-        salt = crypted[:2]
-    return crypt.crypt(pw, salt) == crypted
+    return crypt.crypt(pw, crypted) == crypted
+
+
+
+def getpwnamPasswd(username):
+    """
+    Look up a user in the /etc/passwd database using the pwd module.  If the
+    pwd module is not available, return None.
+
+    @param username: the username of the user to return the passwd database
+    information for.
+    """
+    if not pwd:
+        return None
+    return pwd.getpwnam(username)
+
+
+
+def getpwnamShadow(username):
+    """
+    Look up a user in the /etc/shadow database using the spwd or shadow
+    modules.  If neither module is available, return None.
+
+    @param username: the username of the user to return the shadow database
+    information for.
+    """
+    if not shadow:
+        return None
+    return runAsEffectiveUser(0, 0, shadow.getspnam, username)
+
+
 
 class UNIXPasswordDatabase:
+    """
+    A checker which validates users out of the UNIX password databases, or a
+    databases a compatible format.
+
+    @ivar getpwnamFunctions: a C{tuple} of functions which are called in order
+    to valid a user.  The default value is (getpwnamPasswd, getpwnamShadow),
+    which tries the /etc/passwd database first, followed by the /etc/shadow
+    database.
+    """
     credentialInterfaces = IUsernamePassword,
     implements(ICredentialsChecker)
 
+
+    def __init__(self, getpwnamFunctions=(getpwnamPasswd, getpwnamShadow)):
+        self.getpwnamFunctions = getpwnamFunctions
+
+
     def requestAvatarId(self, credentials):
-        if pwd:
+        for func in self.getpwnamFunctions:
             try:
-                cryptedPass = pwd.getpwnam(credentials.username)[1]
+                pwnam = func(credentials.username)
             except KeyError:
                 return defer.fail(UnauthorizedLogin("invalid username"))
             else:
-                if cryptedPass not in ['*', 'x'] and \
-                    verifyCryptedPassword(cryptedPass, credentials.password):
-                    return defer.succeed(credentials.username)
-        if shadow:
-            gid = os.getegid()
-            uid = os.geteuid()
-            os.setegid(0)
-            os.seteuid(0)
-            try:
-                shadowPass = shadow.getspnam(credentials.username)[1]
-            except KeyError:
-                os.setegid(gid)
-                os.seteuid(uid)
-                return defer.fail(UnauthorizedLogin("invalid username"))
-            os.setegid(gid)
-            os.seteuid(uid)
-            if verifyCryptedPassword(shadowPass, credentials.password):
-                return defer.succeed(credentials.username)
-            return defer.fail(UnauthorizedLogin("invalid password"))
-
+                if pwnam is not None:
+                    cryptedPass = pwnam[1]
+                    if verifyCryptedPassword(cryptedPass,
+                                             credentials.password):
+                        return defer.succeed(credentials.username)
+        # fallback
         return defer.fail(UnauthorizedLogin("unable to verify password"))
+
 
 

branches/spwd-3242-3/twisted/conch/test/test_checkers.py

@@ -5 +5 @@
 Tests for L{twisted.conch.checkers}.
 """
-
-try:
-    import pwd
-except ImportError:
-    pwd = None
-
 import os, base64
 
@@ -23 +17 @@
 try:
     import Crypto.Cipher.DES3
+    Crypto
     import pyasn1
 except ImportError:
-    SSHPublicKeyDatabase = None
+    Crypto = None
 else:
     from twisted.conch.ssh import keys
-    from twisted.conch.checkers import SSHPublicKeyDatabase, SSHProtocolChecker
+    from twisted.conch import checkers
     from twisted.conch.error import NotEnoughAuthentication, ValidPublicKey
     from twisted.conch.test import keydata
@@ -38 +33 @@
     """
 
-    if pwd is None:
+    if checkers.pwd is None:
         skip = "Cannot run without pwd module"
-    elif SSHPublicKeyDatabase is None:
+    elif Crypto is None:
         skip = "Cannot run without PyCrypto or PyASN1"
 
     def setUp(self):
-        self.checker = SSHPublicKeyDatabase()
+        self.checker = checkers.SSHPublicKeyDatabase()
         self.sshDir = FilePath(self.mktemp())
         self.sshDir.makedirs()
@@ -55 +50 @@
         self.mockos.path = self.sshDir.path
         self.patch(os.path, "expanduser", self.mockos.expanduser)
-        self.patch(pwd, "getpwnam", self.mockos.getpwnam)
+        self.patch(checkers.pwd, "getpwnam", self.mockos.getpwnam)
         self.patch(os, "seteuid", self.mockos.seteuid)
         self.patch(os, "setegid", self.mockos.setegid)
@@ -122 +117 @@
             return True
         self.patch(self.checker, 'checkKey', _checkKey)
-        credentials = SSHPrivateKey('test', 'ssh-rsa', keydata.publicRSA_openssh,
-                                    'foo', keys.Key.fromString(keydata.privateRSA_openssh).sign('foo'))
+        credentials = SSHPrivateKey('test', 'ssh-rsa',
+                                    keydata.publicRSA_openssh, 'foo',
+                                    keys.Key.fromString(
+                keydata.privateRSA_openssh).sign('foo'))
         d = self.checker.requestAvatarId(credentials)
         def _verify(avatarId):
@@ -140 +137 @@
             return True
         self.patch(self.checker, 'checkKey', _checkKey)
-        credentials = SSHPrivateKey('test', 'ssh-rsa', keydata.publicRSA_openssh, None, None)
+        credentials = SSHPrivateKey('test', 'ssh-rsa',
+                                    keydata.publicRSA_openssh, None, None)
         d = self.checker.requestAvatarId(credentials)
         return self.assertFailure(d, ValidPublicKey)
@@ -166 +164 @@
             return True
         self.patch(self.checker, 'checkKey', _checkKey)
-        credentials = SSHPrivateKey('test', 'ssh-rsa', keydata.publicRSA_openssh,
-                                    'foo', keys.Key.fromString(keydata.privateDSA_openssh).sign('foo'))
+        credentials = SSHPrivateKey('test', 'ssh-rsa',
+                                    keydata.publicRSA_openssh, 'foo',
+                                    keys.Key.fromString(
+                keydata.privateDSA_openssh).sign('foo'))
         d = self.checker.requestAvatarId(credentials)
         return self.assertFailure(d, UnauthorizedLogin)
@@ -190 +190 @@
 
 
+
 class SSHProtocolCheckerTestCase(TestCase):
     """
@@ -195 +196 @@
     """
 
-    if SSHPublicKeyDatabase is None:
+    if Crypto is None:
         skip = "Cannot run without PyCrypto"
 
@@ -203 +204 @@
         the list of registered checkers.
         """
-        checker = SSHProtocolChecker()
+        checker = checkers.SSHProtocolChecker()
         self.assertEquals(checker.credentialInterfaces, [])
-        checker.registerChecker(SSHPublicKeyDatabase(), )
+        checker.registerChecker(checkers.SSHPublicKeyDatabase(), )
         self.assertEquals(checker.credentialInterfaces, [ISSHPrivateKey])
         self.assertIsInstance(checker.checkers[ISSHPrivateKey],
-                              SSHPublicKeyDatabase)
+                              checkers.SSHPublicKeyDatabase)
 
 
@@ -218 +219 @@
         credentialIntefaces.
         """
-        checker = SSHProtocolChecker()
+        checker = checkers.SSHProtocolChecker()
         self.assertEquals(checker.credentialInterfaces, [])
-        checker.registerChecker(SSHPublicKeyDatabase(), IUsernamePassword)
+        checker.registerChecker(checkers.SSHPublicKeyDatabase(),
+                                IUsernamePassword)
         self.assertEquals(checker.credentialInterfaces, [IUsernamePassword])
         self.assertIsInstance(checker.checkers[IUsernamePassword],
-                              SSHPublicKeyDatabase)
+                              checkers.SSHPublicKeyDatabase)
 
 
@@ -231 +233 @@
         registered checkers to authenticate a user.
         """
-        checker = SSHProtocolChecker()
+        checker = checkers.SSHProtocolChecker()
         passwordDatabase = InMemoryUsernamePasswordDatabaseDontUse()
         passwordDatabase.addUser('test', 'test')
@@ -247 +249 @@
         L{NotEnoughAuthentication}.
         """
-        checker = SSHProtocolChecker()
+        checker = checkers.SSHProtocolChecker()
         def _areDone(avatarId):
             return False
@@ -264 +266 @@
         L{SSHProtocolChecker} should raise L{UnhandledCredentials}.
         """
-        checker = SSHProtocolChecker()
+        checker = checkers.SSHProtocolChecker()
         d = checker.requestAvatarId(UsernamePassword('test', 'test'))
         return self.assertFailure(d, UnhandledCredentials)
@@ -273 +275 @@
         The default L{SSHProcotolChecker.areDone} should simply return True.
         """
-        self.assertEquals(SSHProtocolChecker().areDone(None), True)
+        self.assertEquals(checkers.SSHProtocolChecker().areDone(None), True)
+
+
+
+class HelperFunctionTests(TestCase):
+    """
+    Tests for helper functions L{verifyCryptedPassword}, L{getpwnamPasswd} and
+    L{getpwnamShadow}.
+    """
+    if checkers.pwd is None:
+        skip = 'cannot run without crypt module'
+    elif not getattr(os, 'O_NOCTTY', None):
+        skip = 'cannot run without os.O_NOCTTY'
+
+
+    def setUp(self):
+        self.mockos = MockOS()
+
+
+    def test_verifyCryptedPassword(self):
+        """
+        L{verifyCryptedPassword} returns True if the provided cleartext
+        password matches the provided password hash.
+        """
+        password = 'password'
+        salt = 'sa'
+        encrypted = checkers.crypt.crypt(password, salt)
+        self.assertTrue(checkers.verifyCryptedPassword(
+                encrypted,
+                password), '%r not valid encrypted password for %s' % (
+                encrypted, password))
+
+
+    def test_verifyCryptedPassword_md5(self):
+        """
+        L{verifyCryptedPassword} returns True if the provided cleartext
+        password matches the provided MD5 password hash.
+        """
+        password = 'password'
+        salt = '$1$salt'
+        encrypted = checkers.crypt.crypt(password, salt)
+        self.assertTrue(checkers.verifyCryptedPassword(
+                encrypted,
+                password), '%r not valid encrypted password for %s' % (
+                encrypted, password))
+
+
+    def test_getpwnamPasswd(self):
+        """
+        L{getpwnamPasswd} returns a tuple of items from the UNIX /etc/passwd
+        database if the L{pwd} module is present.  If that module isn't
+        present, it returns None.
+        """
+        if checkers.pwd is None:
+            self.assertIdentical(checkers.getpwnamPasswd('user'), None)
+        else:
+            self.patch(checkers.pwd, "getpwnam", self.mockos.getpwnam)
+            user_line = checkers.getpwnamPasswd('user')
+            self.assertEquals(user_line, (0, 0, 1, 2)) # what MockOS.getpwnam
+                                                       # returns
+
+
+    def test_getpwnamShadow(self):
+        """
+        L{getpwnamShadow} returns a tuple of items from the UNIX /etc/shadow
+        database if the L{spwd} or L{shadow} modules are present.  If the
+        modules are not present, it returns None.
+        """
+        if checkers.shadow is None:
+            self.assertIdentical(checkers.getpwnamShadow('user'), None)
+        else:
+            self.patch(checkers.shadow, 'getspnam', lambda u: (0, 1, 2, 3))
+            self.mockos.euid = 2345
+            self.mockos.egid = 1234
+            self.patch(os, "geteuid", self.mockos.geteuid)
+            self.patch(os, "getegid", self.mockos.getegid)
+            self.patch(os, "seteuid", self.mockos.seteuid)
+            self.patch(os, "setegid", self.mockos.setegid)
+            user_line = checkers.getpwnamShadow('user')
+            self.assertEquals(user_line, (0, 1, 2, 3))
+            self.assertEquals(self.mockos.seteuidCalls, [0, 2345])
+            self.assertEquals(self.mockos.setegidCalls, [0, 1234])
+
+
+
+class UNIXPasswordDatabaseTests(TestCase):
+    """
+    Tests for L{UNIXPasswordDatabase}.
+    """
+    def test_defaultCheckers(self):
+        """
+        L{UNIXPasswordDatabase} with no arguments has L{getpwnamPasswd} and
+        L{getpwnamShadow} as the getpwnamFunctions.
+        """
+        checker = checkers.UNIXPasswordDatabase()
+        self.assertEquals(checker.getpwnamFunctions,
+                          (checkers.getpwnamPasswd,
+                           checkers.getpwnamShadow))
+
+
+    def assertLoggedIn(self, d):
+        """
+        Assert that the L{Deferred} passed in is called back with the value
+        'username'.  This represents a valid login for this TestCase.
+
+        NOTE: To work, this method's return value must be returned from the
+        test method, or otherwise hooked up to the test machinery.
+
+        @param d: a L{Deferred} from an L{IChecker.requestAvatarId} method.
+        @type d: L{Deferred}
+        @rtype: L{Deferred}
+        """
+        def _cbRequestAvatarId(username):
+            self.assertEquals(username, 'username')
+
+        return d.addCallback(_cbRequestAvatarId)
+
+
+    def assertUnauthorizedLogin(self, d):
+        """
+        Asserts that the L{Deferred} passed in is erred back with an
+        L{UnauthorizedLogin} L{Failure}.  This reprsents an invalid login for
+        this TestCase.
+
+        NOTE: To work, this method's return value must be returned from the
+        test method, or otherwise hooked up to the test machinery.
+
+        @param d: a L{Deferred} from an L{IChecker.requestAvatarId} method.
+        @type d: L{Deferred}
+        @rtype: L{Deferred}
+        """
+        def _ebRequestAvatarId(reason):
+            reason.trap(checkers.UnauthorizedLogin)
+        return d.addErrback(_ebRequestAvatarId)
+
+
+    def test_passInCheckers(self):
+        """
+        L{UNIXPasswordDatabase} takes a list of functions to check for UNIX
+        user information.
+        """
+        def getpwnam(username): pass
+        checker = checkers.UNIXPasswordDatabase([getpwnam])
+        self.assertEquals(checker.getpwnamFunctions,
+                          [getpwnam])
+
+
+    def test_verifyPassword(self):
+        """
+        If the encrypted password provided by the getpwnam function is valid
+        (verified by the L{verifyCryptedPassword} function), we callback the
+        C{requestAvatarId} L{Deferred} with the username.
+        """
+        def verifyCryptedPassword(crypted, pw):
+            return crypted == pw
+        def getpwnam(username):
+            return [username, username]
+        self.patch(checkers, 'verifyCryptedPassword', verifyCryptedPassword)
+        checker = checkers.UNIXPasswordDatabase([getpwnam])
+        credential = UsernamePassword('username', 'username')
+        d = checker.requestAvatarId(credential)
+        return self.assertLoggedIn(d)
+
+
+    def test_failOnKeyError(self):
+        """
+        If the getpwnam function raises a KeyError, the login fails with an
+        L{UnauthorizedLogin} exception.
+        """
+        def getpwnam(username):
+            raise KeyError(username)
+        checker = checkers.UNIXPasswordDatabase([getpwnam])
+        credential = UsernamePassword('username', 'username')
+        d = checker.requestAvatarId(credential)
+        return self.assertUnauthorizedLogin(d)
+
+
+    def test_failOnBadPassword(self):
+        """
+        If the verifyCryptedPassword function doesn't verify the password, the
+        login fails with an L{UnauthorizedLogin} exception.
+        """
+        def verifyCryptedPassword(crypted, pw):
+            return False
+        def getpwnam(username):
+            return [username, username]
+        self.patch(checkers, 'verifyCryptedPassword', verifyCryptedPassword)
+        checker = checkers.UNIXPasswordDatabase([getpwnam])
+        credential = UsernamePassword('username', 'username')
+        d = checker.requestAvatarId(credential)
+        return self.assertUnauthorizedLogin(d)
+
+
+    def test_loopThroughFunctions(self):
+        """
+        UNIXPasswordDatabase.requestAvatarId loops through each getpwnam
+        function associated with it and returns a L{Deferred} which fires with
+        the result of the first one which returns a value other than None.
+        ones do not verify the password.
+        """
+        def verifyCryptedPassword(crypted, pw):
+            return crypted == pw
+        def getpwnam1(username):
+            return [username, 'not the password']
+        def getpwnam2(username):
+            return [username, username]
+        self.patch(checkers, 'verifyCryptedPassword', verifyCryptedPassword)
+        checker = checkers.UNIXPasswordDatabase([getpwnam1, getpwnam2])
+        credential = UsernamePassword('username', 'username')
+        d = checker.requestAvatarId(credential)
+        return self.assertLoggedIn(d)