Ensure that verbs cannot be performed on disabled namespaces or by disabled users

endpoints/verbs/__init__.py

@@ -204,6 +204,10 @@ def _verify_repo_verb(_, namespace, repo_name, tag, verb, checker=None):
                  get_authenticated_user())
     abort(405)
 
+  # Make sure the repo's namespace isn't disabled.
+  if not model.is_namespace_enabled(namespace):
+    abort(400)
+
   # If there is a data checker, call it first.
   if checker is not None:
     if not checker(tag_image):

endpoints/verbs/models_interface.py

@@ -152,3 +152,8 @@ class VerbsDataInterface(object):
     or None if none.
     """
     pass
+
+  @abstractmethod
+  def is_namespace_enabled(self, namespace_name):
+    """ Returns whether the given namespace exists and is enabled. """
+    pass

endpoints/verbs/models_pre_oci.py

@@ -141,6 +141,10 @@ class PreOCIModel(VerbsDataInterface):
       v1_metadata=_docker_v1_metadata(namespace_name, repo_name, found),
       internal_db_id=found.id,)
 
+  def is_namespace_enabled(self, namespace_name):
+    namespace = model.user.get_namespace_user(namespace_name)
+    return namespace is not None and namespace.enabled
+
 
 pre_oci_model = PreOCIModel()
 

test/registry/registry_tests.py

@@ -799,6 +799,40 @@ def test_pull_torrent(pusher, basic_images, liveserver_session, liveserver,
     assert expected == found
 
 
+def test_squashed_image_disabled_namespace(pusher, sized_images, liveserver_session,
+                                           liveserver, registry_server_executor, app_reloader):
+  """ Test: Attempting to pull a squashed image from a disabled namespace. """
+  credentials = ('devtable', 'password')
+
+  # Push an image to download.
+  pusher.push(liveserver_session, 'buynlarge', 'newrepo', 'latest', sized_images,
+              credentials=credentials)
+
+  # Disable the buynlarge namespace.
+  registry_server_executor.on(liveserver).disable_namespace('buynlarge')
+
+  # Attempt to pull the squashed version.
+  response = liveserver_session.get('/c1/squash/buynlarge/newrepo/latest', auth=credentials)
+  assert response.status_code == 400
+
+
+def test_squashed_image_disabled_user(pusher, sized_images, liveserver_session,
+                                      liveserver, registry_server_executor, app_reloader):
+  """ Test: Attempting to pull a squashed image via a disabled user. """
+  credentials = ('devtable', 'password')
+
+  # Push an image to download.
+  pusher.push(liveserver_session, 'buynlarge', 'newrepo', 'latest', sized_images,
+              credentials=credentials)
+
+  # Disable the devtable namespace.
+  registry_server_executor.on(liveserver).disable_namespace('devtable')
+
+  # Attempt to pull the squashed version.
+  response = liveserver_session.get('/c1/squash/buynlarge/newrepo/latest', auth=credentials)
+  assert response.status_code == 403
+
+
 @pytest.mark.parametrize('use_estimates', [
   False,
   True,