Open Source Monorail

appengine/monorail/framework/test/monorailrequest_test.py

+# Copyright 2016 The Chromium Authors. All rights reserved.
+# Use of this source code is govered by a BSD-style
+# license that can be found in the LICENSE file or at
+# https://developers.google.com/open-source/licenses/bsd
+
+"""Unit tests for the monorailrequest module."""
+
+import re
+import unittest
+
+import mox
+
+from google.appengine.api import users
+
+import webapp2
+
+from framework import monorailrequest
+from framework import permissions
+from framework import profiler
+from proto import project_pb2
+from proto import tracker_pb2
+from services import service_manager
+from testing import fake
+from testing import testing_helpers
+from tracker import tracker_constants
+
+
+class HostportReTest(unittest.TestCase):
+
+  def testGood(self):
+    test_data = [
+      'localhost:8080',
+      'app.appspot.com',
+      'bugs-staging.chromium.org',
+      'vers10n-h3x-dot-app-id.appspot.com',
+      ]
+    for hostport in test_data:
+      self.assertTrue(monorailrequest._HOSTPORT_RE.match(hostport),
+                      msg='Incorrectly rejected %r' % hostport)
+
+  def testBad(self):
+    test_data = [
+      '',
+      ' ',
+      '\t',
+      '\n',
+      '\'',
+      '"',
+      'version"cruft-dot-app-id.appspot.com',
+      '\nother header',
+      'version&cruft-dot-app-id.appspot.com',
+      ]
+    for hostport in test_data:
+      self.assertFalse(monorailrequest._HOSTPORT_RE.match(hostport),
+                       msg='Incorrectly accepted %r' % hostport)
+
+class AuthDataTest(unittest.TestCase):
+
+  def setUp(self):
+    self.mox = mox.Mox()
+
+  def tearDown(self):
+    self.mox.UnsetStubs()
+
+  def testGetUserID(self):
+    pass  # TODO(jrobbins): re-impement
+
+  def testExamineRequestUserID(self):
+    pass  # TODO(jrobbins): re-implement
+
+
+class MonorailRequestUnitTest(unittest.TestCase):
+
+  def setUp(self):
+    self.services = service_manager.Services(
+        project=fake.ProjectService(),
+        user=fake.UserService(),
+        usergroup=fake.UserGroupService())
+    self.project = self.services.project.TestAddProject('proj')
+    self.services.user.TestAddUser('jrobbins@example.com', 111)
+
+    self.profiler = profiler.Profiler()
+    self.mox = mox.Mox()
+    self.mox.StubOutWithMock(users, 'get_current_user')
+    users.get_current_user().AndReturn(None)
+    self.mox.ReplayAll()
+
+  def tearDown(self):
+    self.mox.UnsetStubs()
+
+  def testGetIntParamConvertsQueryParamToInt(self):
+    notice_id = 12345
+    mr = testing_helpers.MakeMonorailRequest(
+        path='/foo?notice=%s' % notice_id)
+
+    value = mr.GetIntParam('notice')
+    self.assert_(isinstance(value, int))
+    self.assertEqual(notice_id, value)
+
+  def testGetIntParamConvertsQueryParamToLong(self):
+    notice_id = 12345678901234567890
+    mr = testing_helpers.MakeMonorailRequest(
+        path='/foo?notice=%s' % notice_id)
+
+    value = mr.GetIntParam('notice')
+    self.assertTrue(isinstance(value, long))
+    self.assertEqual(notice_id, value)
+
+  def testGetIntListParamNoParam(self):
+    mr = monorailrequest.MonorailRequest()
+    mr.ParseRequest(
+        webapp2.Request.blank('servlet'), self.services, self.profiler)
+    self.assertEquals(mr.GetIntListParam('ids'), None)
+    self.assertEquals(mr.GetIntListParam('ids', default_value=['test']),
+                      ['test'])
+
+  def testGetIntListParamOneValue(self):
+    mr = monorailrequest.MonorailRequest()
+    mr.ParseRequest(
+        webapp2.Request.blank('servlet?ids=11'), self.services, self.profiler)
+    self.assertEquals(mr.GetIntListParam('ids'), [11])
+    self.assertEquals(mr.GetIntListParam('ids', default_value=['test']),
+                      [11])
+
+  def testGetIntListParamMultiValue(self):
+    mr = monorailrequest.MonorailRequest()
+    mr.ParseRequest(
+        webapp2.Request.blank('servlet?ids=21,22,23'), self.services,
+        self.profiler)
+    self.assertEquals(mr.GetIntListParam('ids'), [21, 22, 23])
+    self.assertEquals(mr.GetIntListParam('ids', default_value=['test']),
+                      [21, 22, 23])
+
+  def testGetIntListParamBogusValue(self):
+    mr = monorailrequest.MonorailRequest()
+    mr.ParseRequest(
+        webapp2.Request.blank('servlet?ids=not_an_int'), self.services,
+        self.profiler)
+    self.assertEquals(mr.GetIntListParam('ids'), None)
+    self.assertEquals(mr.GetIntListParam('ids', default_value=['test']),
+                      ['test'])
+
+  def testGetIntListParamMalformed(self):
+    mr = monorailrequest.MonorailRequest()
+    mr.ParseRequest(
+        webapp2.Request.blank('servlet?ids=31,32,,'), self.services,
+        self.profiler)
+    self.assertEquals(mr.GetIntListParam('ids'), None)
+    self.assertEquals(mr.GetIntListParam('ids', default_value=['test']),
+                      ['test'])
+
+  def testDefaultValuesNoUrl(self):
+    """If request has no param, default param values should be used."""
+    mr = monorailrequest.MonorailRequest()
+    mr.ParseRequest(
+        webapp2.Request.blank('servlet'), self.services, self.profiler)
+    self.assertEquals(mr.GetParam('r', 3), 3)
+    self.assertEquals(mr.GetIntParam('r', 3), 3)
+    self.assertEquals(mr.GetPositiveIntParam('r', 3), 3)
+    self.assertEquals(mr.GetIntListParam('r', [3, 4]), [3, 4])
+
+  def _MRWithMockRequest(
+      self, path, headers=None, *mr_args, **mr_kwargs):
+    request = webapp2.Request.blank(path, headers=headers)
+    mr = monorailrequest.MonorailRequest(*mr_args, **mr_kwargs)
+    mr.ParseRequest(request, self.services, self.profiler)
+    return mr
+
+  def testParseQueryParameters(self):
+    mr = self._MRWithMockRequest(
+        '/p/proj/issues/list?q=foo+OR+bar&num=50')
+    self.assertEquals('foo OR bar', mr.query)
+    self.assertEquals(50, mr.num)
+
+  def testParseRequest_Scheme(self):
+    mr = self._MRWithMockRequest('/p/proj/')
+    self.assertEquals('http', mr.request.scheme)
+
+  def testParseRequest_HostportAndCurrentPageURL(self):
+    mr = self._MRWithMockRequest('/p/proj/', headers={
+        'Host': 'example.com',
+        'Cookie': 'asdf',
+        })
+    self.assertEquals('http', mr.request.scheme)
+    self.assertEquals('example.com', mr.request.host)
+    self.assertEquals('http://example.com/p/proj/', mr.current_page_url)
+
+  def testViewedUser_WithEmail(self):
+    mr = self._MRWithMockRequest('/u/jrobbins@example.com/')
+    self.assertEquals('jrobbins@example.com', mr.viewed_username)
+    self.assertEquals(111, mr.viewed_user_auth.user_id)
+    self.assertEquals(
+        self.services.user.GetUser('fake cnxn', 111),
+        mr.viewed_user_auth.user_pb)
+
+  def testViewedUser_WithUserID(self):
+    mr = self._MRWithMockRequest('/u/111/')
+    self.assertEquals('jrobbins@example.com', mr.viewed_username)
+    self.assertEquals(111, mr.viewed_user_auth.user_id)
+    self.assertEquals(
+        self.services.user.GetUser('fake cnxn', 111),
+        mr.viewed_user_auth.user_pb)
+
+  def testViewedUser_NoSuchEmail(self):
+    try:
+      self._MRWithMockRequest('/u/unknownuser@example.com/')
+      self.fail()
+    except webapp2.HTTPException as e:
+      self.assertEquals(404, e.code)
+
+  def testViewedUser_NoSuchUserID(self):
+    with self.assertRaises(webapp2.HTTPException) as e:
+      self._MRWithMockRequest('/u/234521111/')
+      self.assertEquals(404, e.code)
+
+  def testGetParam(self):
+    mr = testing_helpers.MakeMonorailRequest(
+        path='/foo?syn=error!&a=a&empty=',
+        params=dict(over1='over_value1', over2='over_value2'))
+
+    # test tampering
+    self.assertRaises(monorailrequest.InputException, mr.GetParam, 'a',
+                      antitamper_re=re.compile(r'^$'))
+    self.assertRaises(monorailrequest.InputException, mr.GetParam,
+                      'undefined', default_value='default',
+                      antitamper_re=re.compile(r'^$'))
+
+    # test empty value
+    self.assertEquals('', mr.GetParam(
+        'empty', default_value='default', antitamper_re=re.compile(r'^$')))
+
+    # test default
+    self.assertEquals('default', mr.GetParam(
+        'undefined', default_value='default'))
+
+  def testComputeColSpec(self):
+    # No config passed, and nothing in URL
+    mr = testing_helpers.MakeMonorailRequest(
+        path='/p/proj/issues/detail?id=123')
+    mr.ComputeColSpec(None)
+    self.assertEquals(tracker_constants.DEFAULT_COL_SPEC, mr.col_spec)
+
+    # No config passed, but set in URL
+    mr = testing_helpers.MakeMonorailRequest(
+        path='/p/proj/issues/detail?id=123&colspec=a b C')
+    mr.ComputeColSpec(None)
+    self.assertEquals('a b C', mr.col_spec)
+
+    config = tracker_pb2.ProjectIssueConfig()
+
+    # No default in the config, and nothing in URL
+    mr = testing_helpers.MakeMonorailRequest(
+        path='/p/proj/issues/detail?id=123')
+    mr.ComputeColSpec(config)
+    self.assertEquals(tracker_constants.DEFAULT_COL_SPEC, mr.col_spec)
+
+    # No default in the config, but set in URL
+    mr = testing_helpers.MakeMonorailRequest(
+        path='/p/proj/issues/detail?id=123&colspec=a b C')
+    mr.ComputeColSpec(config)
+    self.assertEquals('a b C', mr.col_spec)
+
+    config.default_col_spec = 'd e f'
+
+    # Default in the config, and nothing in URL
+    mr = testing_helpers.MakeMonorailRequest(
+        path='/p/proj/issues/detail?id=123')
+    mr.ComputeColSpec(config)
+    self.assertEquals('d e f', mr.col_spec)
+
+    # Default in the config, but overrided via URL
+    mr = testing_helpers.MakeMonorailRequest(
+        path='/p/proj/issues/detail?id=123&colspec=a b C')
+    mr.ComputeColSpec(config)
+    self.assertEquals('a b C', mr.col_spec)
+
+  def testComputeColSpec_XSS(self):
+    config_1 = tracker_pb2.ProjectIssueConfig()
+    config_2 = tracker_pb2.ProjectIssueConfig()
+    config_2.default_col_spec = "id '+alert(1)+'"
+    mr_1 = testing_helpers.MakeMonorailRequest(
+        path='/p/proj/issues/detail?id=123')
+    mr_2 = testing_helpers.MakeMonorailRequest(
+        path="/p/proj/issues/detail?id=123&colspec=id '+alert(1)+'")
+
+    # Normal colspec in config but malicious request
+    self.assertRaises(
+        monorailrequest.InputException,
+        mr_2.ComputeColSpec, config_1)
+
+    # Malicious colspec in config but normal request
+    self.assertRaises(
+        monorailrequest.InputException,
+        mr_1.ComputeColSpec, config_2)
+
+    # Malicious colspec in config and malicious request
+    self.assertRaises(
+        monorailrequest.InputException,
+        mr_2.ComputeColSpec, config_2)
+
+
+class TestMonorailRequestFunctions(unittest.TestCase):
+
+  def testExtractPathIdenifiers_ProjectOnly(self):
+    username, project_name = monorailrequest._ParsePathIdentifiers(
+        '/p/proj/issues/list?q=foo+OR+bar&ts=1234')
+    self.assertIsNone(username)
+    self.assertEquals('proj', project_name)
+
+  def testExtractPathIdenifiers_ViewedUserOnly(self):
+    username, project_name = monorailrequest._ParsePathIdentifiers(
+        '/u/jrobbins@example.com/')
+    self.assertEquals('jrobbins@example.com', username)
+    self.assertIsNone(project_name)
+
+  def testExtractPathIdenifiers_ViewedUserURLSpace(self):
+    username, project_name = monorailrequest._ParsePathIdentifiers(
+        '/u/jrobbins@example.com/updates')
+    self.assertEquals('jrobbins@example.com', username)
+    self.assertIsNone(project_name)
+
+  def testExtractPathIdenifiers_ViewedGroupURLSpace(self):
+    username, project_name = monorailrequest._ParsePathIdentifiers(
+        '/g/user-group@example.com/updates')
+    self.assertEquals('user-group@example.com', username)
+    self.assertIsNone(project_name)
+
+  def testParseColSpec(self):
+    parse = monorailrequest.ParseColSpec
+    self.assertEqual(['PageName', 'Summary', 'Changed', 'ChangedBy'],
+                     parse(u'PageName Summary Changed ChangedBy'))
+    self.assertEqual(['Foo-Bar', 'Foo-Bar-Baz', 'Release-1.2', 'Hey', 'There'],
+                     parse('Foo-Bar Foo-Bar-Baz Release-1.2 Hey!There'))
+    self.assertEqual(
+        ['\xe7\xaa\xbf\xe8\x8b\xa5\xe7\xb9\xb9'.decode('utf-8'),
+         '\xe5\x9f\xba\xe5\x9c\xb0\xe3\x81\xaf'.decode('utf-8')],
+        parse('\xe7\xaa\xbf\xe8\x8b\xa5\xe7\xb9\xb9 '
+              '\xe5\x9f\xba\xe5\x9c\xb0\xe3\x81\xaf'.decode('utf-8')))
+
+
+class TestPermissionLookup(unittest.TestCase):
+  OWNER_ID = 1
+  OTHER_USER_ID = 2
+
+  def setUp(self):
+    self.services = service_manager.Services(
+        project=fake.ProjectService(),
+        user=fake.UserService(),
+        usergroup=fake.UserGroupService())
+    self.services.user.TestAddUser('owner@gmail.com', self.OWNER_ID)
+    self.services.user.TestAddUser('user@gmail.com', self.OTHER_USER_ID)
+    self.live_project = self.services.project.TestAddProject(
+        'live', owner_ids=[self.OWNER_ID])
+    self.archived_project = self.services.project.TestAddProject(
+        'archived', owner_ids=[self.OWNER_ID],
+        state=project_pb2.ProjectState.ARCHIVED)
+    self.members_only_project = self.services.project.TestAddProject(
+        'members-only', owner_ids=[self.OWNER_ID],
+        access=project_pb2.ProjectAccess.MEMBERS_ONLY)
+
+    self.mox = mox.Mox()
+
+  def tearDown(self):
+    self.mox.UnsetStubs()
+
+  def CheckPermissions(self, perms, expect_view, expect_commit, expect_edit):
+    may_view = perms.HasPerm(permissions.VIEW, None, None)
+    self.assertEqual(expect_view, may_view)
+    may_commit = perms.HasPerm(permissions.COMMIT, None, None)
+    self.assertEqual(expect_commit, may_commit)
+    may_edit = perms.HasPerm(permissions.EDIT_PROJECT, None, None)
+    self.assertEqual(expect_edit, may_edit)
+
+  def MakeRequestAsUser(self, project_name, email):
+    self.mox.StubOutWithMock(users, 'get_current_user')
+    users.get_current_user().AndReturn(testing_helpers.Blank(
+        email=lambda: email))
+    self.mox.ReplayAll()
+
+    request = webapp2.Request.blank('/p/' + project_name)
+    mr = monorailrequest.MonorailRequest()
+    prof = profiler.Profiler()
+    with prof.Phase('parse user info'):
+      mr.ParseRequest(request, self.services, prof)
+    return mr
+
+  def testOwnerPermissions_Live(self):
+    mr = self.MakeRequestAsUser('live', 'owner@gmail.com')
+    self.CheckPermissions(mr.perms, True, True, True)
+
+  def testOwnerPermissions_Archived(self):
+    mr = self.MakeRequestAsUser('archived', 'owner@gmail.com')
+    self.CheckPermissions(mr.perms, True, False, True)
+
+  def testOwnerPermissions_MembersOnly(self):
+    mr = self.MakeRequestAsUser('members-only', 'owner@gmail.com')
+    self.CheckPermissions(mr.perms, True, True, True)
+
+  def testExternalUserPermissions_Live(self):
+    mr = self.MakeRequestAsUser('live', 'user@gmail.com')
+    self.CheckPermissions(mr.perms, True, False, False)
+
+  def testExternalUserPermissions_Archived(self):
+    mr = self.MakeRequestAsUser('archived', 'user@gmail.com')
+    self.CheckPermissions(mr.perms, False, False, False)
+
+  def testExternalUserPermissions_MembersOnly(self):
+    mr = self.MakeRequestAsUser('members-only', 'user@gmail.com')
+    self.CheckPermissions(mr.perms, False, False, False)
+
+
+if __name__ == '__main__':
+  unittest.main()