Merge pull request #1795 from coreos-inc/fix-repo-lookup

Fix repo lookup

data/model/_basequery.py

@@ -34,6 +34,11 @@ def get_public_repo_visibility():
   return Visibility.get(name='public')
 
 
+@lru_cache(maxsize=3)
+def _lookup_team_role(name):
+  return TeamRole.get(name=name)
+
+
 def filter_to_repos_for_user(query, username=None, namespace=None, include_public=True,
                              start_id=None):
   if not include_public and not username:
@@ -85,7 +90,7 @@ def filter_to_repos_for_user(query, username=None, namespace=None, include_publi
                    .switch(Repository)
                    .join(Org, on=(Repository.namespace_user == Org.id))
                    .join(AdminTeam, on=(Org.id == AdminTeam.organization))
-                   .join(TeamRole, on=(AdminTeam.role == TeamRole.id))
+                   .where(AdminTeam.role == _lookup_team_role('admin'))
                    .switch(AdminTeam)
                    .join(AdminTeamMember, on=(AdminTeam.id == AdminTeamMember.team))
                    .join(AdminUser, on=(AdminTeamMember.user == AdminUser.id))

endpoints/api/repository.py

@@ -20,7 +20,7 @@ from endpoints.api.billing import lookup_allowed_private_repos, get_namespace_pl
 from endpoints.api.subscribe import check_repository_usage
 
 from auth.permissions import (ModifyRepositoryPermission, AdministerRepositoryPermission,
-                              CreateRepositoryPermission)
+                              CreateRepositoryPermission, ReadRepositoryPermission)
 from auth.auth_context import get_authenticated_user
 from auth import scopes
 from util.names import REPOSITORY_NAME_REGEX
@@ -158,8 +158,12 @@ class RepositoryList(ApiResource):
         # No repositories should be returned, as there is no user.
         abort(400)
 
-      # Return the full list of repos starred by the current user.
-      repos = list(model.repository.get_user_starred_repositories(user))
+      # Return the full list of repos starred by the current user that are still visible to them.
+      def can_view_repo(repo):
+        return ReadRepositoryPermission(repo.namespace_user.username, repo.name).can()
+
+      unfiltered_repos = model.repository.get_user_starred_repositories(user)
+      repos = [repo for repo in unfiltered_repos if can_view_repo(repo)]
     elif parsed_args['namespace']:
       # Repositories filtered by namespace do not need pagination (their results are fairly small),
       # so we just do the lookup directly.

test/test_api_usage.py

@@ -6,6 +6,7 @@ import logging
 import re
 import json as py_json
 
+from contextlib import contextmanager
 from calendar import timegm
 from StringIO import StringIO
 from urllib import urlencode
@@ -1556,6 +1557,85 @@ class TestListRepos(ApiTestCase):
     for repo in json['repositories']:
       self.assertEquals(ORGANIZATION, repo['namespace'])
 
+  def assertRepositoryVisible(self, namespace, name):
+    json = self.getJsonResponse(RepositoryList,
+                                params=dict(namespace=namespace,
+                                            public=False))
+    self.assertEquals(1, len(json['repositories']))
+    self.assertEquals(name, json['repositories'][0]['name'])
+
+  def assertRepositoryNotVisible(self, namespace, name):
+    json = self.getJsonResponse(RepositoryList, params=dict(namespace=namespace, public=False))
+    for repo in json['repositories']:
+      self.assertNotEquals(name, repo['name'])
+
+    json = self.getJsonResponse(RepositoryList, params=dict(starred=True))
+    for repo in json['repositories']:
+      self.assertNotEquals(name, repo['name'])
+
+  def test_listrepos_starred_filtered(self):
+    admin_user = model.user.get_user(ADMIN_ACCESS_USER)
+    reader_user = model.user.get_user(READ_ACCESS_USER)
+
+    # Create a new organization.
+    new_org = model.organization.create_organization('neworg', 'neworg@devtable.com', admin_user)
+    admin_team = model.team.create_team('admin', new_org, 'admin')
+
+    # Add a repository to the organization.
+    repo = model.repository.create_repository('neworg', 'somerepo', admin_user)
+
+    with self.add_to_team_temporarily(reader_user, admin_team):
+      # Star the repository for the user.
+      model.repository.star_repository(reader_user, repo)
+
+    # Verify that the user cannot see the repo, since they are no longer allowed to do so.
+    self.login(READ_ACCESS_USER)
+    self.assertRepositoryNotVisible('neworg', 'somerepo')
+
+  @contextmanager
+  def add_to_team_temporarily(self, user, team):
+    model.team.add_user_to_team(user, team)
+    yield
+    model.team.remove_user_from_team(team.organization.username, team.name, user.username,
+                                     ADMIN_ACCESS_USER)
+
+  def test_listrepos_org_filtered(self):
+    admin_user = model.user.get_user(ADMIN_ACCESS_USER)
+    reader_user = model.user.get_user(READ_ACCESS_USER)
+
+    # Create a new organization.
+    new_org = model.organization.create_organization('neworg', 'neworg@devtable.com', admin_user)
+
+    admin_team = model.team.create_team('admin', new_org, 'admin')
+    creator_team = model.team.create_team('creators', new_org, 'creator')
+    member_team = model.team.create_team('members', new_org, 'member')
+
+    # Add a repository to the organization.
+    model.repository.create_repository('neworg', 'somerepo', admin_user)
+
+    # Verify that the admin user can view it.
+    self.login(ADMIN_ACCESS_USER)
+    self.assertRepositoryVisible('neworg', 'somerepo')
+
+    # Add reader to a creator team under the org and verify they *cannot* see the repository.
+    with self.add_to_team_temporarily(reader_user, creator_team):
+      self.login(READ_ACCESS_USER)
+      self.assertRepositoryNotVisible('neworg', 'somerepo')
+
+    # Add reader to a member team under the org and verify they *cannot* see the repository.
+    with self.add_to_team_temporarily(reader_user, member_team):
+      self.login(READ_ACCESS_USER)
+      self.assertRepositoryNotVisible('neworg', 'somerepo')
+
+    # Add reader to an admin team under the org and verify they *can* see the repository.
+    with self.add_to_team_temporarily(reader_user, admin_team):
+      self.login(READ_ACCESS_USER)
+      self.assertRepositoryVisible('neworg', 'somerepo')
+
+    # Verify that the public user cannot see the repository.
+    self.login(PUBLIC_USER)
+    self.assertRepositoryNotVisible('neworg', 'somerepo')
+
 
 class TestViewPublicRepository(ApiTestCase):
   def test_normalview(self):