Add a check to ensure repository names are valid according to an extended set of rules.

Fixes #534

endpoints/api/repository.py

@@ -23,6 +23,7 @@ from auth.permissions import (ModifyRepositoryPermission, AdministerRepositoryPe
                               CreateRepositoryPermission)
 from auth.auth_context import get_authenticated_user
 from auth import scopes
+from util.names import REPOSITORY_NAME_REGEX
 
 
 logger = logging.getLogger(__name__)
@@ -104,6 +105,10 @@ class RepositoryList(ApiResource):
       if visibility == 'private':
         check_allowed_private_repos(namespace_name)
 
+      # Verify that the repository name is valid.
+      if not REPOSITORY_NAME_REGEX.match(repository_name):
+        raise InvalidRequest('Invalid repository name')
+
       repo = model.repository.create_repository(namespace_name, repository_name, owner, visibility)
       repo.description = req['description']
       repo.save()

endpoints/v1/index.py

@@ -9,7 +9,7 @@ from data import model
 from app import app, authentication, userevents, storage
 from auth.auth import process_auth, generate_signed_token
 from auth.auth_context import get_authenticated_user, get_validated_token, get_validated_oauth_token
-from util.names import parse_repository_name
+from util.names import parse_repository_name, REPOSITORY_NAME_REGEX
 from auth.permissions import (ModifyRepositoryPermission, UserAdminPermission,
                               ReadRepositoryPermission, CreateRepositoryPermission,
                               repository_read_grant, repository_write_grant)
@@ -173,6 +173,10 @@ def update_user(username):
 @generate_headers(scope=GrantType.WRITE_REPOSITORY, add_grant_for_status=201)
 @anon_allowed
 def create_repository(namespace, repository):
+  # Verify that the repository name is valid.
+  if not REPOSITORY_NAME_REGEX.match(repository):
+    abort(400, message='Invalid repository name. Repository names cannot contain slashes.')
+
   logger.debug('Looking up repository %s/%s', namespace, repository)
   repo = model.repository.get_repository(namespace, repository)
 

test/registry_tests.py

@@ -170,7 +170,7 @@ class RegistryTestCase(LiveServerTestCase):
     self.assertEquals(result.text, '"Username or email already exists"')
     self.conduct('GET', '/v1/users/', auth=(username, password))
 
-  def do_push(self, namespace, repository, username, password, images):
+  def do_push(self, namespace, repository, username, password, images, expected_code=201):
     auth = (username, password)
 
     # Ping!
@@ -180,7 +180,10 @@ class RegistryTestCase(LiveServerTestCase):
     data = [{"id": image['id']} for image in images]
     self.conduct('PUT', '/v1/repositories/%s/%s' % (namespace, repository),
                  data=json.dumps(data), auth=auth,
-                 expected_code=201)
+                 expected_code=expected_code)
+
+    if expected_code != 201:
+      return
 
     for image in images:
       # PUT /v1/images/{imageID}/json
@@ -230,6 +233,7 @@ class RegistryTestCase(LiveServerTestCase):
     # GET /v1/repositories/{namespace}/{repository}/
     self.conduct('GET', prefix + 'images', auth=auth, expected_code=expected_code)
     if expected_code != 200:
+      # Push was expected to fail, so nothing more to do for the push.
       return
 
     # GET /v1/repositories/{namespace}/{repository}/
@@ -254,6 +258,13 @@ class RegistryTestCase(LiveServerTestCase):
 
 
 class RegistryTests(RegistryTestCase):
+  def test_push_reponame_with_slashes(self):
+    # Attempt to add a repository name with slashes. This should fail as we do not support it.
+    images = [{
+      'id': 'onlyimagehere'
+    }]
+    self.do_push('public', 'newrepo/somesubrepo', 'public', 'password', images, expected_code=400)
+
   def test_pull_publicrepo_anonymous(self):
     # Add a new repository under the public user, so we have a real repository to pull.
     images = [{

test/test_api_usage.py

@@ -1272,6 +1272,17 @@ class TestDeleteOrganizationTeamMember(ApiTestCase):
 
 
 class TestCreateRepo(ApiTestCase):
+  def test_invalidreponame(self):
+    self.login(ADMIN_ACCESS_USER)
+
+    json = self.postJsonResponse(RepositoryList,
+                                 data=dict(repository='some/repo',
+                                           visibility='public',
+                                           description=''),
+                                 expected_code=400)
+
+    self.assertEquals('Invalid repository name', json['error_description'])
+
   def test_duplicaterepo(self):
     self.login(ADMIN_ACCESS_USER)
 

util/names.py

@@ -1,8 +1,10 @@
 import urllib
+import re
 
 from functools import wraps
 from uuid import uuid4
 
+REPOSITORY_NAME_REGEX = re.compile(r'^[\.a-zA-Z0-9_-]+$')
 
 def parse_namespace_repository(repository, include_tag=False):
   parts = repository.rstrip('/').split('/', 1)