Implement support for blob mounting via the mount parameter on blob uploads

Fixes https://jira.coreos.com/browse/QUAY-893

endpoints/v2/blob.py

@@ -9,16 +9,19 @@ import resumablehashlib
 
 from app import storage, app, get_app_url, metric_queue, model_cache
 from auth.registry_jwt_auth import process_registry_jwt_auth
+from auth.permissions import ReadRepositoryPermission
 from data import database
 from data.cache import cache_key
 from digest import digest_tools
 from endpoints.decorators import anon_protect, parse_repository_name
 from endpoints.v2 import v2_bp, require_repo_read, require_repo_write, get_input_stream
 from endpoints.v2.errors import (
-  BlobUnknown, BlobUploadInvalid, BlobUploadUnknown, Unsupported, NameUnknown, LayerTooLarge)
+  BlobUnknown, BlobUploadInvalid, BlobUploadUnknown, Unsupported, NameUnknown, LayerTooLarge,
+  InvalidRequest)
 from endpoints.v2.models_interface import Blob
 from endpoints.v2.models_pre_oci import data_model as model
 from util.cache import cache_control
+from util.names import parse_namespace_repository
 from util.registry.filelike import wrap_with_handler, StreamSlice
 from util.registry.gzipstream import calculate_size_handler
 from util.registry.torrent import PieceHasher
@@ -121,6 +124,59 @@ def download_blob(namespace_name, repo_name, digest):
         'Content-Type': BLOB_CONTENT_TYPE,}),)
 
 
+def _try_to_mount_blob(namespace_name, repo_name, mount_blob_digest):
+  """ Attempts to mount a blob requested by the user from another repository. """
+  logger.debug('Got mount request for blob `%s` into `%s/%s`', mount_blob_digest, namespace_name,
+               repo_name)
+  from_repo = request.args.get('from', None)
+  if from_repo is None:
+    raise InvalidRequest
+
+  # Ensure the user has access to the repository.
+  logger.debug('Got mount request for blob `%s` under repository `%s` into `%s/%s`',
+               mount_blob_digest, from_repo, namespace_name, repo_name)
+  from_namespace, from_repo_name = parse_namespace_repository(from_repo,
+                                                              app.config['LIBRARY_NAMESPACE'],
+                                                              include_tag=False)
+
+  # First check permission. This does not hit the DB so we do it first.
+  read_permission = ReadRepositoryPermission(from_namespace, from_repo_name).can()
+  if not read_permission:
+    # If no direct permission, check if the repostory is public.
+    if not model.is_repository_public(from_namespace, from_repo_name):
+      logger.debug('No permission to mount blob `%s` under repository `%s` into `%s/%s`',
+                   mount_blob_digest, from_repo, namespace_name, repo_name)
+      return None
+
+  # Lookup if the mount blob's digest exists in the repository.
+  mount_blob = model.get_blob_by_digest(from_namespace, from_repo_name, mount_blob_digest)
+  if mount_blob is None:
+    logger.debug('Blob `%s` under repository `%s` not found', mount_blob_digest, from_repo)
+    return None
+
+  logger.debug('Mounting blob `%s` under repository `%s` into `%s/%s`', mount_blob_digest,
+               from_repo, namespace_name, repo_name)
+
+  # Mount the blob into the current repository and return that we've completed the operation.
+  expiration_sec = app.config['PUSH_TEMP_TAG_EXPIRATION_SEC']
+  if not model.mount_blob_and_temp_tag(namespace_name, repo_name, mount_blob, expiration_sec):
+    logger.debug('Could not mount blob `%s` under repository `%s` not found', mount_blob_digest,
+                 from_repo)
+    return
+
+  # Return the response for the blob indicating that it was mounted, and including its content
+  # digest.
+  logger.debug('Mounted blob `%s` under repository `%s` into `%s/%s`', mount_blob_digest,
+               from_repo, namespace_name, repo_name)
+  return Response(
+    status=201,
+    headers={
+      'Docker-Content-Digest': mount_blob_digest,
+      'Location':
+        get_app_url() + url_for('v2.download_blob', repository='%s/%s' %
+                                (namespace_name, repo_name), digest=mount_blob_digest),},)
+
+
 @v2_bp.route('/<repopath:repository>/blobs/uploads/', methods=['POST'])
 @parse_repository_name()
 @process_registry_jwt_auth(scopes=['pull', 'push'])
@@ -135,6 +191,14 @@ def start_blob_upload(namespace_name, repo_name):
   if not repository_exists:
     raise NameUnknown()
 
+  # Check for mounting of a blob from another repository.
+  mount_blob_digest = request.args.get('mount', None)
+  if mount_blob_digest is not None:
+    response = _try_to_mount_blob(namespace_name, repo_name, mount_blob_digest)
+    if response is not None:
+      return response
+
+  # Check for a normal blob upload.
   digest = request.args.get('digest', None)
   if digest is None:
     # Short-circuit because the user will send the blob data in another request.