luci-config: fine-grained acls

appengine/config_service/acl_test.py

Index: appengine/config_service/acl_test.py
diff --git a/appengine/config_service/acl_test.py b/appengine/config_service/acl_test.py
index c295f3b534ec0a7380b66b9e2355c4e0e9f6c49a..9b7cfd72da4a90453bc6ff640e2166eb567fd8d6 100755
--- a/appengine/config_service/acl_test.py
+++ b/appengine/config_service/acl_test.py
@@ -3,8 +3,6 @@
 # Use of this source code is governed by the Apache v2.0 license that can be
 # found in the LICENSE file.
 
-import wsgiref.headers
-
 from test_env import future
 import test_env
 test_env.setup_test_env()
@@ -14,19 +12,25 @@ import mock
 
 from components import auth
 
+from proto import project_config_pb2
 from proto import service_config_pb2
 import acl
+import projects
+import services
 import storage
 
 
 class AclTestCase(test_case.TestCase):
   def setUp(self):
     super(AclTestCase, self).setUp()
+    self.mock(auth, 'get_current_identity', mock.Mock())
+    auth.get_current_identity.return_value = auth.Anonymous
     self.mock(auth, 'is_admin', lambda *_: False)
     self.mock(auth, 'is_group_member', mock.Mock(return_value=False))
+    self.mock(services, 'get_service_async', mock.Mock())
+    services.get_service_async.side_effect = lambda sid: future(None)
 
     acl_cfg = service_config_pb2.AclCfg(
-        service_access_group='service-admins',
         project_access_group='project-admins',
     )
     self.mock(storage, 'get_self_config_async', lambda *_: future(acl_cfg))
@@ -35,30 +39,48 @@ class AclTestCase(test_case.TestCase):
     self.mock(auth, 'is_admin', mock.Mock(return_value=True))
     self.assertTrue(acl.can_read_config_set('services/swarming'))
     self.assertTrue(acl.can_read_config_set('projects/chromium'))
-    self.assertTrue(acl.can_read_project_list())
+    self.assertTrue(acl.has_project_access('chromium'))
 
-  def test_can_read_service_config(self):
-    auth.is_group_member.return_value = True
-    self.assertTrue(acl.can_read_config_set('services/swarming'))
-    auth.is_group_member.access_called_once_with('service-admins')
+  def test_has_service_access(self):
+    self.assertFalse(acl.can_read_config_set('services/swarming'))
 
-  def test_can_read_service_config_header(self):
-    headers = wsgiref.headers.Headers([
-      ('X-Appengine-Inbound-Appid', 'swarming'),
-    ])
-    self.assertTrue(
-        acl.can_read_config_set('services/swarming', headers=headers))
+    service_cfg = service_config_pb2.Service(
+        id='swarming', access=['group:swarming-app'])
+    services.get_service_async.side_effect = lambda sid: future(service_cfg)
+    auth.is_group_member.side_effect = lambda g: g == 'swarming-app'
 
-  def test_can_read_service_config_no_access(self):
+    self.assertTrue(acl.can_read_config_set('services/swarming'))
+
+  def test_has_service_access_no_access(self):
     self.assertFalse(acl.can_read_config_set('services/swarming'))
 
-  def test_can_read_project_config(self):
-    auth.is_group_member.return_value = True
-    self.assertTrue(acl.can_read_config_set('projects/swarming'))
-    auth.is_group_member.access_called_once_with('project-admins')
+  def test_has_project_access_group(self):
+    self.mock(projects, 'get_metadata', mock.Mock())
+    projects.get_metadata.return_value = project_config_pb2.ProjectCfg(
+        access=['group:googlers', 'a@a.com']
+    )
+
+    self.assertFalse(acl.can_read_config_set('projects/secret'))
+
+    auth.is_group_member.side_effect = lambda name: name == 'googlers'
+    self.assertTrue(acl.can_read_config_set('projects/secret'))
+
+    auth.is_group_member.side_effect = lambda name: name == 'project-admins'
+    self.assertTrue(acl.can_read_config_set('projects/secret'))
+
+  def test_has_project_access_identity(self):
+    self.mock(projects, 'get_metadata', mock.Mock())
+    projects.get_metadata.return_value = project_config_pb2.ProjectCfg(
+        access=['group:googlers', 'a@a.com']
+    )
+
+    self.assertFalse(acl.can_read_config_set('projects/secret'))
+
+    auth.get_current_identity.return_value = auth.Identity('user', 'a@a.com')
+    self.assertTrue(acl.can_read_config_set('projects/secret'))
 
   def test_can_read_project_config_no_access(self):
-    self.assertFalse(acl.can_read_config_set('projects/swarming'))
+    self.assertFalse(acl.has_project_access('projects/swarming'))
     self.assertFalse(acl.can_read_config_set('projects/swarming/refs/heads/x'))
 
   def test_malformed_config_set(self):