issue2550833 enhance the export csv action to include the keys for

liked items rather than id's. So for nosy list display usernames and
not numbers.

The original code was renamed and also made available. See change
document.

Author: rouilj

CHANGES.txt

@@ -43,6 +43,11 @@ Features:
   other than GET that doesn't have a payload. E.G. DELETE, PATCH or
   OPTIONS. Verbs like PUT and POST usually have payloads, so this
   patch doesn't touch processing of these methods. (John Rouillard)
+- issue2550833: the export_csv web action now returns labels/names
+  rather than id's. Replace calls to export_csv with the export_csv_id
+  action to return the same data as the old export_csv action. (Tom
+  Ekberg (tekberg), Andreas (anrounham14) edited/applied and tests
+  created by John Rouillard)
 
 Fixed:
 

roundup/cgi/actions.py

@@ -14,7 +14,7 @@
 __all__ = ['Action', 'ShowAction', 'RetireAction', 'RestoreAction', 'SearchAction',
            'EditCSVAction', 'EditItemAction', 'PassResetAction',
            'ConfRegoAction', 'RegisterAction', 'LoginAction', 'LogoutAction',
-           'NewItemAction', 'ExportCSVAction']
+           'NewItemAction', 'ExportCSVAction', 'ExportCSVWithIdAction']
 
 # used by a couple of routines
 chars = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'
@@ -1294,6 +1294,159 @@ def verifyPassword(self, userid, givenpw):
 class ExportCSVAction(Action):
     name = 'export'
     permissionType = 'View'
+    list_sep = ';'              # Separator for list types
+
+    def handle(self):
+        ''' Export the specified search query as CSV. '''
+        # figure the request
+        request = templating.HTMLRequest(self.client)
+        filterspec = request.filterspec
+        sort = request.sort
+        group = request.group
+        columns = request.columns
+        klass = self.db.getclass(request.classname)
+
+        # check if all columns exist on class
+        # the exception must be raised before sending header
+        props = klass.getprops()
+        for cname in columns:
+            if cname not in props:
+                # use error code 400: Bad Request. Do not use
+                # error code 404: Not Found.
+                self.client.response_code = 400
+                raise exceptions.NotFound(
+                    self._('Column "%(column)s" not found in %(class)s')
+                    % {'column': cgi.escape(cname), 'class': request.classname})
+
+        # full-text search
+        if request.search_text:
+            matches = self.db.indexer.search(
+                re.findall(r'\b\w{2,25}\b', request.search_text), klass)
+        else:
+            matches = None
+
+        header = self.client.additional_headers
+        header['Content-Type'] = 'text/csv; charset=%s' % self.client.charset
+        # some browsers will honor the filename here...
+        header['Content-Disposition'] = 'inline; filename=query.csv'
+
+        self.client.header()
+
+        if self.client.env['REQUEST_METHOD'] == 'HEAD':
+            # all done, return a dummy string
+            return 'dummy'
+
+        wfile = self.client.request.wfile
+        if self.client.charset != self.client.STORAGE_CHARSET:
+            wfile = codecs.EncodedFile(wfile,
+                self.client.STORAGE_CHARSET, self.client.charset, 'replace')
+
+        writer = csv.writer(wfile)
+
+        # handle different types of columns.
+        def repr_no_right(cls, col):
+            """User doesn't have the right to see the value of col."""
+            def fct(arg):
+                return "[hidden]"
+            return fct
+        def repr_link(cls, col):
+            """Generate a function which returns the string representation of
+            a link depending on `cls` and `col`."""
+            def fct(arg):
+                if arg == None:
+                    return ""
+                else:
+                    return str(cls.get(arg, col))
+            return fct
+        def repr_list(cls, col):
+            def fct(arg):
+                if arg == None:
+                    return ""
+                elif type(arg) is list:
+                    seq = [str(cls.get(val, col)) for val in arg]
+                    return self.list_sep.join(seq)
+            return fct
+        def repr_date():
+            def fct(arg):
+                if arg == None:
+                    return ""
+                else:
+                    if (arg.local(self.db.getUserTimezone()).pretty('%H:%M') ==
+                        '00:00'):
+                        fmt = '%Y-%m-%d'
+                    else:
+                        fmt = '%Y-%m-%d %H:%M'
+                return arg.local(self.db.getUserTimezone()).pretty(fmt)
+            return fct
+        def repr_val():
+            def fct(arg):
+                if arg == None:
+                    return ""
+                else:
+                    return str(arg)
+            return fct
+
+        props = klass.getprops()
+        
+        # Determine translation map.
+        ncols = []
+        represent = {}
+        for col in columns:
+            ncols.append(col)
+            represent[col] = repr_val()
+            if isinstance(props[col], hyperdb.Multilink):
+                cname = props[col].classname
+                cclass = self.db.getclass(cname)
+                represent[col] = repr_list(cclass, 'name')
+                if not self.hasPermission(self.permissionType, classname=cname):
+                    represent[col] = repr_no_right(cclass, 'name')
+                else:
+                    if 'name' in cclass.getprops():
+                        represent[col] = repr_list(cclass, 'name')
+                    elif cname == 'user':
+                        represent[col] = repr_list(cclass, 'realname')
+            if isinstance(props[col], hyperdb.Link):
+                cname = props[col].classname
+                cclass = self.db.getclass(cname)
+                if not self.hasPermission(self.permissionType, classname=cname):
+                    represent[col] = repr_no_right(cclass, 'name')
+                else:
+                    if 'name' in cclass.getprops():
+                        represent[col] = repr_link(cclass, 'name')
+                    elif cname == 'user':
+                        represent[col] = repr_link(cclass, 'realname')
+            if isinstance(props[col], hyperdb.Date):
+                represent[col] = repr_date()
+
+        columns = ncols
+        # generate the CSV output
+        self.client._socket_op(writer.writerow, columns)
+        # and search
+        for itemid in klass.filter(matches, filterspec, sort, group):
+            row = []
+            # don't put out a row of [hidden] fields if the user has
+            # no access to the issue.
+            if not self.hasPermission(self.permissionType, itemid=itemid,
+                                      classname=request.classname):
+                continue
+            for name in columns:
+                # check permission for this property on this item
+                # TODO: Permission filter doesn't work for the 'user' class
+                if not self.hasPermission(self.permissionType, itemid=itemid,
+                        classname=request.classname, property=name):
+                    repr_function = repr_no_right(request.classname, name)
+                else:
+                    repr_function = represent[name]
+                row.append(repr_function(klass.get(itemid, name)))
+            self.client._socket_op(writer.writerow, row)
+        return '\n'
+
+class ExportCSVWithIdAction(Action):
+    ''' A variation of ExportCSVAction that returns ID number rather than
+        names. This is the original csv export function.
+    '''
+    name = 'export'
+    permissionType = 'View'
 
     def handle(self):
         ''' Export the specified search query as CSV. '''
@@ -1346,10 +1499,24 @@ def handle(self):
         # and search
         for itemid in klass.filter(matches, filterspec, sort, group):
             row = []
+            # FIXME should this code raise an exception if an item
+            # is included that can't be accessed? Enabling this
+            # check will just skip the row for the inaccessible item.
+            # This makes it act more like the web interface.
+            #if not self.hasPermission(self.permissionType, itemid=itemid,
+            #                          classname=request.classname):
+            #    continue
             for name in columns:
                 # check permission to view this property on this item
-                if not self.hasPermission('View', itemid=itemid,
+                if not self.hasPermission(self.permissionType, itemid=itemid,
                         classname=request.classname, property=name):
+                    # FIXME: is this correct, or should we just
+                    # emit a '[hidden]' string. Note that this may
+                    # allow an attacker to figure out hidden schema
+                    # properties.
+                    # A bad property name will result in an exception.
+                    # A valid property results in a column of '[hidden]'
+                    #   values.
                     raise exceptions.Unauthorised(self._(
                         'You do not have permission to view %(class)s'
                     ) % {'class': request.classname})
@@ -1358,7 +1525,6 @@ def handle(self):
 
         return '\n'
 
-
 class Bridge(BaseAction):
     """Make roundup.actions.Action executable via CGI request.
 

roundup/cgi/client.py

@@ -1756,6 +1756,7 @@ def renderContext(self):
         ('retire',      actions.RetireAction),
         ('show',        actions.ShowAction),
         ('export_csv',  actions.ExportCSVAction),
+        ('export_csv_id',  actions.ExportCSVWithIdAction),
     )
     def handle_action(self):
         """ Determine whether there should be an Action called.

test/test_cgi.py

@@ -1507,16 +1507,40 @@ def testRoles(self):
         self.assert_(not item.hasRole(''))
 
     def testCSVExport(self):
-        cl = self._make_client({'@columns': 'id,name'}, nodeid=None,
-            userid='1')
-        cl.classname = 'status'
+        cl = self._make_client(
+            {'@columns': 'id,title,status,keyword,assignedto,nosy'},
+            nodeid=None, userid='1')
+        cl.classname = 'issue'
+
+        demo_id=self.db.user.create(username='demo', address='[email protected]',
+            roles='User', realname='demo')
+        key_id1=self.db.keyword.create(name='keyword1')
+        key_id2=self.db.keyword.create(name='keyword2')
+        self.db.issue.create(title='foo1', status='2', assignedto='4', nosy=['3',demo_id])
+        self.db.issue.create(title='bar2', status='1', assignedto='3', keyword=[key_id1,key_id2])
+        self.db.issue.create(title='baz32', status='4')
         output = StringIO()
         cl.request = MockNull()
         cl.request.wfile = output
+        # call export version that outputs names
         actions.ExportCSVAction(cl).handle()
-        self.assertEquals('id,name\r\n1,unread\r\n2,deferred\r\n3,chatting\r\n'
-            '4,need-eg\r\n5,in-progress\r\n6,testing\r\n7,done-cbb\r\n'
-            '8,resolved\r\n',
+        #print(output.getvalue())
+        self.assertEquals('id,title,status,keyword,assignedto,nosy\r\n'
+'1,foo1,deferred,,"Contrary, Mary","Bork, Chef;demo;Contrary, Mary"\r\n'
+'2,bar2,unread,keyword1;keyword2,"Bork, Chef","Bork, Chef"\r\n'
+'3,baz32,need-eg,,,\r\n',
+
+            output.getvalue())
+        output = StringIO()
+        cl.request = MockNull()
+        cl.request.wfile = output
+        # call export version that outputs id numbers
+        actions.ExportCSVWithIdAction(cl).handle()
+        #print(output.getvalue())
+        self.assertEquals('id,title,status,keyword,assignedto,nosy\r\n'
+                          "1,foo1,2,[],4,\"['3', '5', '4']\"\r\n"
+                          "2,bar2,1,\"['1', '2']\",3,['3']\r\n"
+                          '3,baz32,4,[],None,[]\r\n',
             output.getvalue())
 
     def testCSVExportBadColumnName(self):
@@ -1545,6 +1569,68 @@ def testCSVExportFailPermissionBadColumn(self):
             actions.ExportCSVAction(cl).handle)
 
     def testCSVExportFailPermissionValidColumn(self):
+        passwd=password.Password('foo')
+        demo_id=self.db.user.create(username='demo', address='[email protected]',
+                                    roles='User', realname='demo',
+                                    password=passwd)
+        cl = self._make_client({'@columns': 'id,username,address,password'},
+                               nodeid=None, userid=demo_id)
+        cl.classname = 'user'
+        output = StringIO()
+        cl.request = MockNull()
+        cl.request.wfile = output
+        # used to be self.assertRaises(exceptions.Unauthorised,
+        # but not acting like the column name is not found
+
+        actions.ExportCSVAction(cl).handle()
+        #print(output.getvalue())
+        self.assertEquals('id,username,address,password\r\n'
+                          '1,admin,[hidden],[hidden]\r\n'
+                          '2,anonymous,[hidden],[hidden]\r\n'
+                          '3,Chef,[hidden],[hidden]\r\n'
+                          '4,mary,[hidden],[hidden]\r\n'
+                          '5,demo,[email protected],%s\r\n'%(passwd),
+            output.getvalue())
+
+    def testCSVExportWithId(self):
+        cl = self._make_client({'@columns': 'id,name'}, nodeid=None,
+            userid='1')
+        cl.classname = 'status'
+        output = StringIO()
+        cl.request = MockNull()
+        cl.request.wfile = output
+        actions.ExportCSVWithIdAction(cl).handle()
+        self.assertEquals('id,name\r\n1,unread\r\n2,deferred\r\n3,chatting\r\n'
+            '4,need-eg\r\n5,in-progress\r\n6,testing\r\n7,done-cbb\r\n'
+            '8,resolved\r\n',
+            output.getvalue())
+
+    def testCSVExportWithIdBadColumnName(self):
+        cl = self._make_client({'@columns': 'falseid,name'}, nodeid=None,
+            userid='1')
+        cl.classname = 'status'
+        output = StringIO()
+        cl.request = MockNull()
+        cl.request.wfile = output
+        self.assertRaises(exceptions.NotFound,
+            actions.ExportCSVWithIdAction(cl).handle)
+
+    def testCSVExportWithIdFailPermissionBadColumn(self):
+        cl = self._make_client({'@columns': 'id,email,password'}, nodeid=None,
+            userid='2')
+        cl.classname = 'user'
+        output = StringIO()
+        cl.request = MockNull()
+        cl.request.wfile = output
+        # used to be self.assertRaises(exceptions.Unauthorised,
+        # but not acting like the column name is not found
+        # see issue2550755 - should this return Unauthorised?
+        # The unauthorised user should never get to the point where
+        # they can determine if the column name is valid or not.
+        self.assertRaises(exceptions.NotFound,
+            actions.ExportCSVWithIdAction(cl).handle)
+
+    def testCSVExportWithIdFailPermissionValidColumn(self):
         cl = self._make_client({'@columns': 'id,address,password'}, nodeid=None,
             userid='2')
         cl.classname = 'user'
@@ -1554,7 +1640,7 @@ def testCSVExportFailPermissionValidColumn(self):
         # used to be self.assertRaises(exceptions.Unauthorised,
         # but not acting like the column name is not found
         self.assertRaises(exceptions.Unauthorised,
-            actions.ExportCSVAction(cl).handle)
+            actions.ExportCSVWithIdAction(cl).handle)
 
 class TemplateHtmlRendering(unittest.TestCase):
     ''' try to test the rendering code for tal '''