Add security checks and tests for xmlrpc interface.

Author: Richard Jones

roundup/xmlrpc.py

@@ -49,14 +49,15 @@ def __init__(self, tracker, username, password):
         self.tracker = tracker
         self.db = self.tracker.open('admin')
         try:
-            userid = self.db.user.lookup(username)
+            self.userid = self.db.user.lookup(username)
         except KeyError: # No such user
             self.db.close()
-            raise Unauthorised, 'Invalid user.'
-        stored = self.db.user.get(userid, 'password')
-        if stored != password: # Wrong password
+            raise Unauthorised, 'Invalid user'
+        stored = self.db.user.get(self.userid, 'password')
+        if stored != password:
+            # Wrong password
             self.db.close()
-            raise Unauthorised, 'Invalid user.'
+            raise Unauthorised, 'Invalid user'
         self.db.setCurrentUser(username)
 
     def close(self):
@@ -112,30 +113,41 @@ def __init__(self, tracker, verbose = False):
         self.tracker = roundup.instance.open(tracker)
         self.verbose = verbose
 
-    def list(self, username, password, classname, propname = None):
-
+    def list(self, username, password, classname, propname=None):
         r = RoundupRequest(self.tracker, username, password)
         cl = r.get_class(classname)
         if not propname:
             propname = cl.labelprop()
-        result = [cl.get(id, propname) for id in cl.list()]
+        def has_perm(itemid):
+            return True
+            r.db.security.hasPermission('View', r.userid, classname,
+                itemid=itemid, property=propname)
+        result = [cl.get(id, propname) for id in cl.list()
+            if has_perm(id)]
         r.close()
         return result
 
     def display(self, username, password, designator, *properties):
-
         r = RoundupRequest(self.tracker, username, password)
-        classname, nodeid = hyperdb.splitDesignator(designator)
+        classname, itemid = hyperdb.splitDesignator(designator)
+
+        if not r.db.security.hasPermission('View', r.userid, classname,
+                itemid=itemid):
+            raise Unauthorised('Permission to view %s denied'%designator)
+
         cl = r.get_class(classname)
         props = properties and list(properties) or cl.properties.keys()
         props.sort()
-        result = [(property, cl.get(nodeid, property)) for property in props]
+        result = [(property, cl.get(itemid, property)) for property in props]
         r.close()
         return dict(result)
 
     def create(self, username, password, classname, *args):
-
         r = RoundupRequest(self.tracker, username, password)
+
+        if not r.db.security.hasPermission('Create', r.userid, classname):
+            raise Unauthorised('Permission to create %s denied'%classname)
+
         cl = r.get_class(classname)
 
         # convert types
@@ -157,9 +169,13 @@ def create(self, username, password, classname, *args):
         return result
 
     def set(self, username, password, designator, *args):
-
         r = RoundupRequest(self.tracker, username, password)
         classname, itemid = hyperdb.splitDesignator(designator)
+
+        if not r.db.security.hasPermission('Edit', r.userid, classname,
+                itemid=itemid):
+            raise Unauthorised('Permission to edit %s denied'%designator)
+
         cl = r.get_class(classname)
 
         # convert types

test/test_xmlrpc.py

@@ -14,75 +14,85 @@
 
 NEEDS_INSTANCE = 1
 
-class TestCaseBase(unittest.TestCase):
-
+class TestCase(unittest.TestCase):
     def setUp(self):
-
         self.dirname = '_test_xmlrpc'
         # set up and open a tracker
         self.instance = db_test_base.setupTracker(self.dirname)
 
         # open the database
         self.db = self.instance.open('admin')
-        self.db.user.create(username='joe', password=password.Password('random'),
-                            address='[email protected]',
-                            realname='Joe Random', roles='User')
+        self.joeid = 'user' + self.db.user.create(username='joe',
+            password=password.Password('random'), address='[email protected]',
+            realname='Joe Random', roles='User')
 
         self.db.commit()
         self.db.close()
-        
-        self.server = RoundupServer(self.dirname)
 
+        self.server = RoundupServer(self.dirname)
 
     def tearDown(self):
-
         try:
             shutil.rmtree(self.dirname)
         except OSError, error:
             if error.errno not in (errno.ENOENT, errno.ESRCH): raise
 
-class AccessTestCase(TestCaseBase):
-
-    def test(self):
-
+    def testAccess(self):
         # Retrieve all three users.
         results = self.server.list('joe', 'random', 'user', 'id')
         self.assertEqual(len(results), 3)
+
         # Obtain data for 'joe'.
-        userid = 'user' + results[-1]
-        results = self.server.display('joe', 'random', userid)
+        results = self.server.display('joe', 'random', self.joeid)
         self.assertEqual(results['username'], 'joe')
         self.assertEqual(results['realname'], 'Joe Random')
+
+    def testChange(self):
         # Reset joe's 'realname'.
-        results = self.server.set('joe', 'random', userid, 'realname=Joe Doe')
-        results = self.server.display('joe', 'random', userid, 'realname')
+        results = self.server.set('joe', 'random', self.joeid,
+            'realname=Joe Doe')
+        results = self.server.display('joe', 'random', self.joeid,
+            'realname')
         self.assertEqual(results['realname'], 'Joe Doe')
-        # Create test
+
+    def testCreate(self):
         results = self.server.create('joe', 'random', 'issue', 'title=foo')
         issueid = 'issue' + results
         results = self.server.display('joe', 'random', issueid, 'title')
         self.assertEqual(results['title'], 'foo')
 
-class AuthenticationTestCase(TestCaseBase):
-
-    def test(self):
-
+    def testAuthUnknown(self):
         # Unknown user (caught in XMLRPC frontend).
         self.assertRaises(Unauthorised, self.server.list,
-                          'nobody', 'nobody', 'user', 'id')
+            'nobody', 'nobody', 'user', 'id')
+
+    def testAuthDeniedEdit(self):
         # Wrong permissions (caught by roundup security module).
-        results = self.server.list('joe', 'random', 'user', 'id')
-        userid = 'user' + results[0] # admin
         self.assertRaises(Unauthorised, self.server.set,
-                          'joe', 'random', userid, 'realname=someone')
+            'joe', 'random', 'user1', 'realname=someone')
 
+    def testAuthDeniedCreate(self):
+        self.assertRaises(Unauthorised, self.server.create,
+            'joe', 'random', 'user', {'username': 'blah'})
+
+    def testAuthAllowedEdit(self):
+        try:
+            self.server.set('admin', 'sekrit', 'user2', 'realname=someone')
+        except Unauthorised, err:
+            self.fail('raised %s'%err)
+
+    def testAuthAllowedCreate(self):
+        try:
+            self.server.create('admin', 'sekrit', 'user', 'username=blah')
+        except Unauthorised, err:
+            self.fail('raised %s'%err)
 
 def test_suite():
     suite = unittest.TestSuite()
-    suite.addTest(unittest.makeSuite(AccessTestCase))
-    suite.addTest(unittest.makeSuite(AuthenticationTestCase))
+    suite.addTest(unittest.makeSuite(TestCase))
     return suite
 
 if __name__ == '__main__':
     runner = unittest.TextTestRunner()
     unittest.main(testRunner=runner)
+