Ticket #5808: cred-open-5808-2.patch

twisted/cred/checkers.py

@@ -8 +8 @@
 
 from twisted.internet import defer
 from twisted.python import failure, log
+from twisted.python.filepath import FilePath
 from twisted.cred import error, credentials
 
 
@@ -97 +98 @@
 
 
 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 +195 @@
 
     def _loadCredentials(self):
         try:
-            f = file(self.filename)
+            f = FilePath(self.filename).open()
         except:
             log.err()
             raise error.UnauthorizedLogin()

twisted/test/test_newcred.py

@@ -12 +12 @@
 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 +172 @@
         c = credentials.CramMD5Credentials()
         self.failIf(c.checkPassword('secret'))
 
+
+
 class OnDiskDatabaseTestCase(unittest.TestCase):
     users = [
         ('user1', 'pass1'),
@@ -177 +181 @@
         ('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 +243 @@
     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 +270 @@
         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 +279 @@
         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 +288 @@
         d.addCallback(self._assertFailures, error.UnhandledCredentials)
         return d
 
+
     def _assertFailures(self, failures, *expectedFailures):
         for flag, failure in failures:
             self.assertEqual(flag, defer.FAILURE)
@@ -290 +298 @@
     if crypt is None:
         skip = "crypt module not available"
 
+
+
 class PluggableAuthenticationModulesTest(unittest.TestCase):
 
     def setUp(self):
@@ -362 +372 @@
     if not pamauth:
         skip = "Can't run without PyPAM"
 
+
+
 class CheckersMixin:
     def testPositive(self):
         for chk in self.getCheckers():
@@ -379 +391 @@
                 self.assertRaises(error.UnauthorizedLogin, r.getResult)
     testNegative = dG(testNegative)
 
+
+
 class HashlessFilePasswordDBMixin:
     credClass = credentials.UsernamePassword
     diskHash = None
@@ -389 +403 @@
         ('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