Index: twisted/cred/checkers.py =================================================================== --- twisted/cred/checkers.py (revision 34963) +++ twisted/cred/checkers.py (working copy) @@ -8,6 +8,7 @@ from twisted.internet import defer from twisted.python import failure, log +from twisted.python.filepath import FilePath from twisted.cred import error, credentials @@ -97,7 +98,8 @@ class FilePasswordDB: - """A file-based, text-based username/password database. + """ + A file-based, text-based username/password database. Records in the datafile for this class are delimited by a particular string. The username appears in a fixed field of the columns delimited @@ -193,7 +195,7 @@ def _loadCredentials(self): try: - f = file(self.filename) + f = FilePath(self.filename).open() except: log.err() raise error.UnauthorizedLogin() Index: twisted/test/test_newcred.py =================================================================== --- twisted/test/test_newcred.py (revision 34963) +++ twisted/test/test_newcred.py (working copy) @@ -12,8 +12,10 @@ from twisted.trial import unittest from twisted.cred import portal, checkers, credentials, error from twisted.python import components +from twisted.python.filepath import FilePath from twisted.internet import defer -from twisted.internet.defer import deferredGenerator as dG, waitForDeferred as wFD +from twisted.internet.defer import (deferredGenerator as dG, + waitForDeferred as wFD) try: from crypt import crypt @@ -170,6 +172,8 @@ c = credentials.CramMD5Credentials() self.failIf(c.checkPassword('secret')) + + class OnDiskDatabaseTestCase(unittest.TestCase): users = [ ('user1', 'pass1'), @@ -177,50 +181,49 @@ ('user3', 'pass3'), ] + def setUp(self): + self.dbfile = FilePath(self.mktemp()) - def testUserLookup(self): - dbfile = self.mktemp() - db = checkers.FilePasswordDB(dbfile) - f = file(dbfile, 'w') - for (u, p) in self.users: - f.write('%s:%s\n' % (u, p)) - f.close() + def test_userLookup(self): + db = checkers.FilePasswordDB(self.dbfile.path) + with self.dbfile.open('w') as f: + for (u, p) in self.users: + f.write('%s:%s\n' % (u, p)) for (u, p) in self.users: self.failUnlessRaises(KeyError, db.getUser, u.upper()) self.assertEqual(db.getUser(u), (u, p)) - def testCaseInSensitivity(self): - dbfile = self.mktemp() - db = checkers.FilePasswordDB(dbfile, caseSensitive=0) - f = file(dbfile, 'w') - for (u, p) in self.users: - f.write('%s:%s\n' % (u, p)) - f.close() + + def test_caseInSensitivity(self): + db = checkers.FilePasswordDB(self.dbfile.path, caseSensitive=0) + with self.dbfile.open('w') as f: + for (u, p) in self.users: + f.write('%s:%s\n' % (u, p)) for (u, p) in self.users: self.assertEqual(db.getUser(u.upper()), (u, p)) - def testRequestAvatarId(self): - dbfile = self.mktemp() - db = checkers.FilePasswordDB(dbfile, caseSensitive=0) - f = file(dbfile, 'w') - for (u, p) in self.users: - f.write('%s:%s\n' % (u, p)) - f.close() + + def test_requestAvatarId(self): + db = checkers.FilePasswordDB(self.dbfile.path, caseSensitive=0) + with self.dbfile.open('w') as f: + for (u, p) in self.users: + f.write('%s:%s\n' % (u, p)) + creds = [credentials.UsernamePassword(u, p) for u, p in self.users] d = defer.gatherResults( [defer.maybeDeferred(db.requestAvatarId, c) for c in creds]) d.addCallback(self.assertEqual, [u for u, p in self.users]) return d - def testRequestAvatarId_hashed(self): - dbfile = self.mktemp() - db = checkers.FilePasswordDB(dbfile, caseSensitive=0) - f = file(dbfile, 'w') - for (u, p) in self.users: - f.write('%s:%s\n' % (u, p)) - f.close() + + def test_requestAvatarId_hashed(self): + db = checkers.FilePasswordDB(self.dbfile.path, caseSensitive=0) + with self.dbfile.open('w') as f: + for (u, p) in self.users: + f.write('%s:%s\n' % (u, p)) + creds = [credentials.UsernameHashedPassword(u, p) for u, p in self.users] d = defer.gatherResults( [defer.maybeDeferred(db.requestAvatarId, c) for c in creds]) @@ -240,24 +243,26 @@ def hash(self, u, p, s): return crypt(p, s) + def setUp(self): - dbfile = self.mktemp() - self.db = checkers.FilePasswordDB(dbfile, hash=self.hash) - f = file(dbfile, 'w') - for (u, p) in self.users: - f.write('%s:%s\n' % (u, crypt(p, u[:2]))) - f.close() + dbfile = FilePath(self.mktemp()) + self.db = checkers.FilePasswordDB(dbfile.path, hash=self.hash) + with dbfile.open('w') as f: + for (u, p) in self.users: + f.write('%s:%s\n' % (u, crypt(p, u[:2]))) r = TestRealm() self.port = portal.Portal(r) self.port.registerChecker(self.db) - def testGoodCredentials(self): + + def test_goodCredentials(self): goodCreds = [credentials.UsernamePassword(u, p) for u, p in self.users] d = defer.gatherResults([self.db.requestAvatarId(c) for c in goodCreds]) d.addCallback(self.assertEqual, [u for u, p in self.users]) return d - def testGoodCredentials_login(self): + + def test_goodCredentials_login(self): goodCreds = [credentials.UsernamePassword(u, p) for u, p in self.users] d = defer.gatherResults([self.port.login(c, None, ITestable) for c in goodCreds]) @@ -265,7 +270,8 @@ d.addCallback(self.assertEqual, [u for u, p in self.users]) return d - def testBadCredentials(self): + + def test_badCredentials(self): badCreds = [credentials.UsernamePassword(u, 'wrong password') for u, p in self.users] d = defer.DeferredList([self.port.login(c, None, ITestable) @@ -273,7 +279,8 @@ d.addCallback(self._assertFailures, error.UnauthorizedLogin) return d - def testHashedCredentials(self): + + def test_hashedCredentials(self): hashedCreds = [credentials.UsernameHashedPassword(u, crypt(p, u[:2])) for u, p in self.users] d = defer.DeferredList([self.port.login(c, None, ITestable) @@ -281,6 +288,7 @@ d.addCallback(self._assertFailures, error.UnhandledCredentials) return d + def _assertFailures(self, failures, *expectedFailures): for flag, failure in failures: self.assertEqual(flag, defer.FAILURE) @@ -290,6 +298,8 @@ if crypt is None: skip = "crypt module not available" + + class PluggableAuthenticationModulesTest(unittest.TestCase): def setUp(self): @@ -362,6 +372,8 @@ if not pamauth: skip = "Can't run without PyPAM" + + class CheckersMixin: def testPositive(self): for chk in self.getCheckers(): @@ -379,6 +391,8 @@ self.assertRaises(error.UnauthorizedLogin, r.getResult) testNegative = dG(testNegative) + + class HashlessFilePasswordDBMixin: credClass = credentials.UsernamePassword diskHash = None @@ -389,57 +403,74 @@ ('user2', 'password2'), ('user3', 'password3')] + def getGoodCredentials(self): for u, p in self._validCredentials: yield self.credClass(u, self.networkHash(p)), u + def getBadCredentials(self): for u, p in [('user1', 'password3'), ('user2', 'password1'), ('bloof', 'blarf')]: yield self.credClass(u, self.networkHash(p)) + def getCheckers(self): diskHash = self.diskHash or (lambda x: x) - hashCheck = self.diskHash and (lambda username, password, stored: self.diskHash(password)) + hashCheck = self.diskHash and (lambda username, password, + stored: self.diskHash(password)) for cache in True, False: - fn = self.mktemp() - fObj = file(fn, 'w') - for u, p in self._validCredentials: - fObj.write('%s:%s\n' % (u, diskHash(p))) - fObj.close() - yield checkers.FilePasswordDB(fn, cache=cache, hash=hashCheck) - - fn = self.mktemp() - fObj = file(fn, 'w') - for u, p in self._validCredentials: - fObj.write('%s dingle dongle %s\n' % (diskHash(p), u)) - fObj.close() - yield checkers.FilePasswordDB(fn, ' ', 3, 0, cache=cache, hash=hashCheck) - - fn = self.mktemp() - fObj = file(fn, 'w') - for u, p in self._validCredentials: - fObj.write('zip,zap,%s,zup,%s\n' % (u.title(), diskHash(p))) - fObj.close() - yield checkers.FilePasswordDB(fn, ',', 2, 4, False, cache=cache, hash=hashCheck) + fn = FilePath(self.mktemp()) + with fn.open('w') as fObj: + for u, p in self._validCredentials: + fObj.write('%s:%s\n' % (u, diskHash(p))) + yield checkers.FilePasswordDB(fn.path, cache=cache, hash=hashCheck) + + fn = FilePath(self.mktemp()) + with fn.open('w') as fObj: + for u, p in self._validCredentials: + fObj.write('%s dingle dongle %s\n' % (diskHash(p), u)) + yield checkers.FilePasswordDB(fn.path, ' ', 3, 0, cache=cache, + hash=hashCheck) + + fn = FilePath(self.mktemp()) + with fn.open('w') as fObj: + for u, p in self._validCredentials: + fObj.write('zip,zap,%s,zup,%s\n' % (u.title(), + diskHash(p))) + yield checkers.FilePasswordDB(fn.path, ',', 2, 4, False, + cache=cache, hash=hashCheck) + + class LocallyHashedFilePasswordDBMixin(HashlessFilePasswordDBMixin): diskHash = staticmethod(lambda x: x.encode('hex')) + + class NetworkHashedFilePasswordDBMixin(HashlessFilePasswordDBMixin): networkHash = staticmethod(lambda x: x.encode('hex')) class credClass(credentials.UsernameHashedPassword): def checkPassword(self, password): return self.hashed.decode('hex') == password -class HashlessFilePasswordDBCheckerTestCase(HashlessFilePasswordDBMixin, CheckersMixin, unittest.TestCase): + + +class HashlessFilePasswordDBCheckerTestCase(HashlessFilePasswordDBMixin, + CheckersMixin, unittest.TestCase): pass -class LocallyHashedFilePasswordDBCheckerTestCase(LocallyHashedFilePasswordDBMixin, CheckersMixin, unittest.TestCase): + + +class LocallyHashedFilePasswordDBCheckerTestCase(LocallyHashedFilePasswordDBMixin, + CheckersMixin, unittest.TestCase): pass -class NetworkHashedFilePasswordDBCheckerTestCase(NetworkHashedFilePasswordDBMixin, CheckersMixin, unittest.TestCase): + + +class NetworkHashedFilePasswordDBCheckerTestCase(NetworkHashedFilePasswordDBMixin, + CheckersMixin, unittest.TestCase): pass