Fix an NPE when trying to pull a manifest without a legacy image via V1

data/registry_model/interface.py

@@ -12,6 +12,12 @@ class RegistryDataInterface(object):
     """ Returns whether the implementation of the data interface supports schema 2 format
         manifests. """
 
+  @abstractmethod
+  def get_tag_legacy_image_id(self, repository_ref, tag_name, storage):
+    """ Returns the legacy image ID for the tag with a legacy images in
+        the repository. Returns None if None.
+    """
+
   @abstractmethod
   def get_legacy_tags_map(self, repository_ref, storage):
     """ Returns a map from tag name to its legacy image ID, for all tags with legacy images in

data/registry_model/registry_oci_model.py

@@ -30,6 +30,25 @@ class OCIModel(SharedModel, RegistryDataInterface):
         manifests. """
     return True
 
+  def get_tag_legacy_image_id(self, repository_ref, tag_name, storage):
+    """ Returns the legacy image ID for the tag with a legacy images in
+        the repository. Returns None if None.
+    """
+    tag = self.get_repo_tag(repository_ref, tag_name, include_legacy_image=True)
+    if tag is None:
+      return None
+
+    if tag.legacy_image_if_present is not None:
+      return tag.legacy_image_if_present.docker_image_id
+
+    if tag.manifest.media_type == DOCKER_SCHEMA2_MANIFESTLIST_CONTENT_TYPE:
+      # See if we can lookup a schema1 legacy image.
+      v1_compatible = self.get_schema1_parsed_manifest(tag.manifest, '', '', '', storage)
+      if v1_compatible is not None:
+        return v1_compatible.leaf_layer_v1_image_id
+
+    return None
+
   def get_legacy_tags_map(self, repository_ref, storage):
     """ Returns a map from tag name to its legacy image ID, for all tags with legacy images in
         the repository. Note that this can be a *very* heavy operation.
@@ -52,7 +71,6 @@ class OCIModel(SharedModel, RegistryDataInterface):
             if v1_id is not None:
               tags_map[tag.name] = v1_id
 
-
     return tags_map
 
   def _get_legacy_compatible_image_for_manifest(self, manifest, storage):

data/registry_model/registry_pre_oci_model.py

@@ -33,6 +33,16 @@ class PreOCIModel(SharedModel, RegistryDataInterface):
         manifests. """
     return False
 
+  def get_tag_legacy_image_id(self, repository_ref, tag_name, storage):
+    """ Returns the legacy image ID for the tag with a legacy images in
+        the repository. Returns None if None.
+    """
+    tag = self.get_repo_tag(repository_ref, tag_name, include_legacy_image=True)
+    if tag is None:
+      return None
+
+    return tag.legacy_image.docker_image_id
+
   def get_legacy_tags_map(self, repository_ref, storage):
     """ Returns a map from tag name to its legacy image, for all tags with legacy images in
         the repository.

endpoints/v1/tag.py

@@ -44,11 +44,11 @@ def get_tag(namespace_name, repo_name, tag):
     if repository_ref is None:
       abort(404)
 
-    tag = registry_model.get_repo_tag(repository_ref, tag, include_legacy_image=True)
+    image_id = registry_model.get_tag_legacy_image_id(repository_ref, tag, storage)
-    if tag is None:
+    if image_id is None:
       abort(404)
 
-    resp = make_response('"%s"' % tag.legacy_image.docker_image_id)
+    resp = make_response('"%s"' % image_id)
     resp.headers['Content-Type'] = 'application/json'
     return resp
 

test/registry/protocol_v1.py

@@ -15,6 +15,7 @@ class V1ProtocolSteps(Enum):
   PUT_TAG = 'put-tag'
   PUT_IMAGE_JSON = 'put-image-json'
   DELETE_TAG = 'delete-tag'
+  GET_TAG = 'get-tag'
   GET_LAYER = 'get-layer'
 
 
@@ -49,6 +50,9 @@ class V1Protocol(RegistryProtocol):
     V1ProtocolSteps.GET_LAYER: {
       Failures.GEO_BLOCKED: 403,
     },
+    V1ProtocolSteps.GET_TAG: {
+      Failures.UNKNOWN_TAG: 404,
+    },
   }
 
   def __init__(self, jwk):
@@ -99,11 +103,18 @@ class V1Protocol(RegistryProtocol):
     image_ids = self.conduct(session, 'GET', prefix + 'tags', headers=headers).json()
 
     for tag_name in tag_names:
+      # GET /v1/repositories/{namespace}/{repository}/tags/<tag_name>
+      image_id_data = self.conduct(session, 'GET', prefix + 'tags/' + tag_name,
+                                   headers=headers,
+                                   expected_status=(200, expected_failure,
+                                                    V1ProtocolSteps.GET_TAG))
+
       if tag_name not in image_ids:
         assert expected_failure == Failures.UNKNOWN_TAG
         return None
 
       tag_image_id = image_ids[tag_name]
+      assert image_id_data.json() == tag_image_id
 
       # Retrieve the ancestry of the tagged image.
       image_prefix = '/v1/images/%s/' % tag_image_id

test/registry/registry_tests.py

@@ -1803,3 +1803,22 @@ def test_push_pull_unicode_direct(pusher, puller, unicode_images, liveserver_ses
   # Pull the repository to verify.
   puller.pull(liveserver_session, 'devtable', 'newrepo', 'latest', unicode_images,
               credentials=credentials, options=options)
+
+
+def test_push_legacy_pull_not_allowed(v22_protocol, v1_protocol, remote_images, liveserver_session,
+                                      app_reloader, data_model):
+  """ Test: Push a V2 Schema 2 manifest and attempt to pull via V1 when there is no assigned legacy
+      image.
+  """
+  if data_model != 'oci_model':
+    return
+
+  credentials = ('devtable', 'password')
+
+  # Push a new repository.
+  v22_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest', remote_images,
+                    credentials=credentials)
+
+  # Attempt to pull. Should fail with a 404.
+  v1_protocol.pull(liveserver_session, 'devtable', 'newrepo', 'latest', remote_images,
+                   credentials=credentials, expected_failure=Failures.UNKNOWN_TAG)