quay/endpoints/v1/registry.py

import logging
import json

from functools import wraps
from datetime import datetime
from time import time

from flask import make_response, request, session, Response, redirect, abort as flask_abort

from app import storage as store, app, metric_queue
from auth.auth_context import get_authenticated_user
from auth.decorators import extract_namespace_repo_from_session, process_auth
from auth.permissions import (ReadRepositoryPermission, ModifyRepositoryPermission)
from data import model, database
from digest import checksums
from endpoints.v1 import v1_bp
from endpoints.v1.models_pre_oci import pre_oci_model as model
from endpoints.v1.index import ensure_namespace_enabled
from endpoints.decorators import anon_protect
from util.http import abort, exact_abort
from util.registry.filelike import SocketReader
from util.registry import gzipstream
from util.registry.replication import queue_storage_replication
from util.registry.torrent import PieceHasher

logger = logging.getLogger(__name__)


def _finish_image(namespace, repository, image_id):
  # Checksum is ok, we remove the marker
  blob_ref = model.update_image_uploading(namespace, repository, image_id, False)

  # Send a job to the work queue to replicate the image layer.
  queue_storage_replication(namespace, blob_ref)


def require_completion(f):
  """This make sure that the image push correctly finished."""

  @wraps(f)
  def wrapper(namespace, repository, *args, **kwargs):
    image_id = kwargs['image_id']
    if model.is_image_uploading(namespace, repository, image_id):
      abort(400, 'Image %(image_id)s is being uploaded, retry later', issue='upload-in-progress',
            image_id=image_id)
    return f(namespace, repository, *args, **kwargs)

  return wrapper


def set_cache_headers(f):
  """Returns HTTP headers suitable for caching."""

  @wraps(f)
  def wrapper(*args, **kwargs):
    # Set TTL to 1 year by default
    ttl = 31536000
    expires = datetime.fromtimestamp(int(time()) + ttl)
    expires = expires.strftime('%a, %d %b %Y %H:%M:%S GMT')
    headers = {
      'Cache-Control': 'public, max-age={0}'.format(ttl),
      'Expires': expires,
      'Last-Modified': 'Thu, 01 Jan 1970 00:00:00 GMT',}
    if 'If-Modified-Since' in request.headers:
      response = make_response('Not modified', 304)
      response.headers.extend(headers)
      return response
    kwargs['headers'] = headers
    # Prevent the Cookie to be sent when the object is cacheable
    session.modified = False
    return f(*args, **kwargs)

  return wrapper


@v1_bp.route('/images/<image_id>/layer', methods=['HEAD'])
@process_auth
@extract_namespace_repo_from_session
@ensure_namespace_enabled
@require_completion
@set_cache_headers
@anon_protect
def head_image_layer(namespace, repository, image_id, headers):
  permission = ReadRepositoryPermission(namespace, repository)

  logger.debug('Checking repo permissions')
  if permission.can() or model.repository_is_public(namespace, repository):
    repo = model.get_repository(namespace, repository)
    if repo.kind != 'image':
      msg = 'This repository is for managing %s resources and not container images.' % repo.kind
      abort(405, message=msg, image_id=image_id)

    logger.debug('Looking up placement locations')
    locations, _ = model.placement_locations_and_path_docker_v1(namespace, repository, image_id)
    if locations is None:
      logger.debug('Could not find any blob placement locations')
      abort(404, 'Image %(image_id)s not found', issue='unknown-image', image_id=image_id)

    # Add the Accept-Ranges header if the storage engine supports resumable
    # downloads.
    extra_headers = {}
    if store.get_supports_resumable_downloads(locations):
      logger.debug('Storage supports resumable downloads')
      extra_headers['Accept-Ranges'] = 'bytes'

    resp = make_response('')
    resp.headers.extend(headers)
    resp.headers.extend(extra_headers)
    return resp

  abort(403)


@v1_bp.route('/images/<image_id>/layer', methods=['GET'])
@process_auth
@extract_namespace_repo_from_session
@ensure_namespace_enabled
@require_completion
@set_cache_headers
@anon_protect
def get_image_layer(namespace, repository, image_id, headers):
  permission = ReadRepositoryPermission(namespace, repository)

  logger.debug('Checking repo permissions')
  if permission.can() or model.repository_is_public(namespace, repository):
    repo = model.get_repository(namespace, repository)
    if repo.kind != 'image':
      msg = 'This repository is for managing %s resources and not container images.' % repo.kind
      abort(405, message=msg, image_id=image_id)

    logger.debug('Looking up placement locations and path')
    locations, path = model.placement_locations_and_path_docker_v1(namespace, repository, image_id)
    if not locations or not path:
      abort(404, 'Image %(image_id)s not found', issue='unknown-image', image_id=image_id)
    try:
      logger.debug('Looking up the direct download URL for path: %s', path)
      direct_download_url = store.get_direct_download_url(locations, path, request.remote_addr)
      if direct_download_url:
        logger.debug('Returning direct download URL')
        resp = redirect(direct_download_url)
        return resp

      # Close the database handle here for this process before we send the long download.
      database.close_db_filter(None)
      logger.debug('Streaming layer data')
      return Response(store.stream_read(locations, path), headers=headers)
    except (IOError, AttributeError):
      logger.exception('Image layer data not found')
      abort(404, 'Image %(image_id)s not found', issue='unknown-image', image_id=image_id)

  abort(403)


@v1_bp.route('/images/<image_id>/layer', methods=['PUT'])
@process_auth
@extract_namespace_repo_from_session
@ensure_namespace_enabled
@anon_protect
def put_image_layer(namespace, repository, image_id):
  logger.debug('Checking repo permissions')
  permission = ModifyRepositoryPermission(namespace, repository)
  if not permission.can():
    abort(403)

  repo = model.get_repository(namespace, repository)
  if repo.kind != 'image':
    msg = 'This repository is for managing %s resources and not container images.' % repo.kind
    abort(405, message=msg, image_id=image_id)

  logger.debug('Retrieving image')
  if model.storage_exists(namespace, repository, image_id):
    exact_abort(409, 'Image already exists')

  v1_metadata = model.docker_v1_metadata(namespace, repository, image_id)
  if v1_metadata is None:
    abort(404)

  logger.debug('Storing layer data')

  input_stream = request.stream
  if request.headers.get('transfer-encoding') == 'chunked':
    # Careful, might work only with WSGI servers supporting chunked
    # encoding (Gunicorn)
    input_stream = request.environ['wsgi.input']

  # Create a socket reader to read the input stream containing the layer data.
  sr = SocketReader(input_stream)

  # Add a handler that copies the data into a temp file. This is used to calculate the tarsum,
  # which is only needed for older versions of Docker.
  requires_tarsum = session.get('checksum_format') == 'tarsum'
  if requires_tarsum:
    tmp, tmp_hndlr = store.temp_store_handler()
    sr.add_handler(tmp_hndlr)

  # Add a handler to compute the compressed and uncompressed sizes of the layer.
  size_info, size_hndlr = gzipstream.calculate_size_handler()
  sr.add_handler(size_hndlr)

  # Add a handler to hash the chunks of the upload for torrenting
  piece_hasher = PieceHasher(app.config['BITTORRENT_PIECE_SIZE'])
  sr.add_handler(piece_hasher.update)

  # Add a handler which computes the checksum.
  h, sum_hndlr = checksums.simple_checksum_handler(v1_metadata.compat_json)
  sr.add_handler(sum_hndlr)

  # Add a handler which computes the content checksum only
  ch, content_sum_hndlr = checksums.content_checksum_handler()
  sr.add_handler(content_sum_hndlr)

  # Stream write the data to storage.
  locations, path = model.placement_locations_and_path_docker_v1(namespace, repository, image_id)
  with database.CloseForLongOperation(app.config):
    try:
      start_time = time()
      store.stream_write(locations, path, sr)
      metric_queue.chunk_size.Observe(size_info.compressed_size, labelvalues=[list(locations)[0]])
      metric_queue.chunk_upload_time.Observe(time() - start_time, labelvalues=[list(locations)[0]])
    except IOError:
      logger.exception('Exception when writing image data')
      abort(520, 'Image %(image_id)s could not be written. Please try again.', image_id=image_id)

  # Save the size of the image.
  model.update_image_sizes(namespace, repository, image_id, size_info.compressed_size,
                           size_info.uncompressed_size)

  # Save the BitTorrent pieces.
  model.create_bittorrent_pieces(namespace, repository, image_id,
                                 piece_hasher.final_piece_hashes())

  # Append the computed checksum.
  csums = []
  csums.append('sha256:{0}'.format(h.hexdigest()))

  try:
    if requires_tarsum:
      tmp.seek(0)
      csums.append(checksums.compute_tarsum(tmp, v1_metadata.compat_json))
      tmp.close()
  except (IOError, checksums.TarError) as exc:
    logger.debug('put_image_layer: Error when computing tarsum %s', exc)

  if v1_metadata.checksum is None:
    # We don't have a checksum stored yet, that's fine skipping the check.
    # Not removing the mark though, image is not downloadable yet.
    session['checksum'] = csums
    session['content_checksum'] = 'sha256:{0}'.format(ch.hexdigest())
    return make_response('true', 200)

  # We check if the checksums provided matches one the one we computed
  if v1_metadata.checksum not in csums:
    logger.warning('put_image_layer: Wrong checksum')
    abort(400, 'Checksum mismatch; ignoring the layer for image %(image_id)s',
          issue='checksum-mismatch', image_id=image_id)

  # Mark the image as uploaded.
  _finish_image(namespace, repository, image_id)

  return make_response('true', 200)


@v1_bp.route('/images/<image_id>/checksum', methods=['PUT'])
@process_auth
@extract_namespace_repo_from_session
@ensure_namespace_enabled
@anon_protect
def put_image_checksum(namespace, repository, image_id):
  logger.debug('Checking repo permissions')
  permission = ModifyRepositoryPermission(namespace, repository)
  if not permission.can():
    abort(403)

  repo = model.get_repository(namespace, repository)
  if repo.kind != 'image':
    msg = 'This repository is for managing %s resources and not container images.' % repo.kind
    abort(405, message=msg, image_id=image_id)

  # Docker Version < 0.10 (tarsum+sha):
  old_checksum = request.headers.get('X-Docker-Checksum')

  # Docker Version >= 0.10 (sha):
  new_checksum = request.headers.get('X-Docker-Checksum-Payload')

  # Store whether we need to calculate the tarsum.
  if new_checksum:
    session['checksum_format'] = 'sha256'
  else:
    session['checksum_format'] = 'tarsum'

  checksum = new_checksum or old_checksum
  if not checksum:
    abort(400, "Missing checksum for image %(image_id)s", issue='missing-checksum',
          image_id=image_id)

  if not session.get('checksum'):
    abort(400, 'Checksum not found in Cookie for image %(image_id)s',
          issue='missing-checksum-cookie', image_id=image_id)

  logger.debug('Looking up repo image')
  v1_metadata = model.docker_v1_metadata(namespace, repository, image_id)
  if not v1_metadata:
    abort(404, 'Image not found: %(image_id)s', issue='unknown-image', image_id=image_id)

  logger.debug('Looking up repo layer data')
  if not v1_metadata.compat_json:
    abort(404, 'Image not found: %(image_id)s', issue='unknown-image', image_id=image_id)

  logger.debug('Marking image path')
  if not model.is_image_uploading(namespace, repository, image_id):
    abort(409, 'Cannot set checksum for image %(image_id)s', issue='image-write-error',
          image_id=image_id)

  logger.debug('Storing image and content checksums')

  content_checksum = session.get('content_checksum', None)
  checksum_parts = checksum.split(':')
  if len(checksum_parts) != 2:
    abort(400, 'Invalid checksum format')

  model.store_docker_v1_checksums(namespace, repository, image_id, checksum, content_checksum)

  if checksum not in session.get('checksum', []):
    logger.debug('session checksums: %s', session.get('checksum', []))
    logger.debug('client supplied checksum: %s', checksum)
    logger.debug('put_image_checksum: Wrong checksum')
    abort(400, 'Checksum mismatch for image: %(image_id)s', issue='checksum-mismatch',
          image_id=image_id)

  # Mark the image as uploaded.
  _finish_image(namespace, repository, image_id)

  return make_response('true', 200)


@v1_bp.route('/images/<image_id>/json', methods=['GET'])
@process_auth
@extract_namespace_repo_from_session
@ensure_namespace_enabled
@require_completion
@set_cache_headers
@anon_protect
def get_image_json(namespace, repository, image_id, headers):
  logger.debug('Checking repo permissions')
  permission = ReadRepositoryPermission(namespace, repository)
  if not permission.can() and not model.repository_is_public(namespace, repository):
    abort(403)

  repo = model.get_repository(namespace, repository)
  if repo.kind != 'image':
    msg = 'This repository is for managing %s resources and not container images.' % repo.kind
    abort(405, message=msg, image_id=image_id)

  logger.debug('Looking up repo image')
  v1_metadata = model.docker_v1_metadata(namespace, repository, image_id)
  if v1_metadata is None:
    flask_abort(404)

  logger.debug('Looking up repo layer size')
  size = model.get_image_size(namespace, repository, image_id)
  if size is not None:
    # Note: X-Docker-Size is optional and we *can* end up with a NULL image_size,
    # so handle this case rather than failing.
    headers['X-Docker-Size'] = str(size)

  response = make_response(v1_metadata.compat_json, 200)
  response.headers.extend(headers)
  return response


@v1_bp.route('/images/<image_id>/ancestry', methods=['GET'])
@process_auth
@extract_namespace_repo_from_session
@ensure_namespace_enabled
@require_completion
@set_cache_headers
@anon_protect
def get_image_ancestry(namespace, repository, image_id, headers):
  logger.debug('Checking repo permissions')
  permission = ReadRepositoryPermission(namespace, repository)
  if not permission.can() and not model.repository_is_public(namespace, repository):
    abort(403)

  repo = model.get_repository(namespace, repository)
  if repo.kind != 'image':
    msg = 'This repository is for managing %s resources and not container images.' % repo.kind
    abort(405, message=msg, image_id=image_id)

  ancestry_docker_ids = model.image_ancestry(namespace, repository, image_id)
  if ancestry_docker_ids is None:
    abort(404, 'Image %(image_id)s not found', issue='unknown-image', image_id=image_id)

  # We can not use jsonify here because we are returning a list not an object
  response = make_response(json.dumps(ancestry_docker_ids), 200)
  response.headers.extend(headers)
  return response


@v1_bp.route('/images/<image_id>/json', methods=['PUT'])
@process_auth
@extract_namespace_repo_from_session
@ensure_namespace_enabled
@anon_protect
def put_image_json(namespace, repository, image_id):
  logger.debug('Checking repo permissions')
  permission = ModifyRepositoryPermission(namespace, repository)
  if not permission.can():
    abort(403)

  repo = model.get_repository(namespace, repository)
  if repo.kind != 'image':
    msg = 'This repository is for managing %s resources and not container images.' % repo.kind
    abort(405, message=msg, image_id=image_id)

  logger.debug('Parsing image JSON')
  try:
    uploaded_metadata = request.data
    data = json.loads(uploaded_metadata.decode('utf8'))
  except ValueError:
    pass

  if not data or not isinstance(data, dict):
    abort(400, 'Invalid JSON for image: %(image_id)s\nJSON: %(json)s', issue='invalid-request',
          image_id=image_id, json=request.data)

  if 'id' not in data:
    abort(400, 'Missing key `id` in JSON for image: %(image_id)s', issue='invalid-request',
          image_id=image_id)

  if image_id != data['id']:
    abort(400, 'JSON data contains invalid id for image: %(image_id)s', issue='invalid-request',
          image_id=image_id)

  logger.debug('Looking up repo image')

  if not model.repository_exists(namespace, repository):
    abort(404, 'Repository does not exist: %(namespace)s/%(repository)s', issue='no-repo',
          namespace=namespace, repository=repository)

  parent_id = data.get('parent', None)
  if parent_id:
    logger.debug('Looking up parent image')
    if model.docker_v1_metadata(namespace, repository, parent_id) is None:
      abort(400, 'Image %(image_id)s depends on non existing parent image %(parent_id)s',
            issue='invalid-request', image_id=image_id, parent_id=parent_id)

  v1_metadata = model.docker_v1_metadata(namespace, repository, image_id)
  if v1_metadata is None:
    username = get_authenticated_user() and get_authenticated_user().username
    logger.debug('Image not found, creating or linking image with initiating user context: %s',
                 username)
    location_pref = store.preferred_locations[0]
    model.create_or_link_image(username, namespace, repository, image_id, location_pref)
    v1_metadata = model.docker_v1_metadata(namespace, repository, image_id)

  # Create a temporary tag to prevent this image from getting garbage collected while the push
  # is in progress.
  model.create_temp_hidden_tag(namespace, repository, image_id,
                               app.config['PUSH_TEMP_TAG_EXPIRATION_SEC'])

  logger.debug('Checking if image already exists')
  if v1_metadata and not model.is_image_uploading(namespace, repository, image_id):
    exact_abort(409, 'Image already exists')

  model.update_image_uploading(namespace, repository, image_id, True)

  # If we reach that point, it means that this is a new image or a retry
  # on a failed push, save the metadata
  command_list = data.get('container_config', {}).get('Cmd', None)
  command = json.dumps(command_list) if command_list else None

  logger.debug('Setting image metadata')
  model.update_docker_v1_metadata(namespace, repository, image_id,
                                  data.get('created'),
                                  data.get('comment'), command, uploaded_metadata, parent_id)

  return make_response('true', 200)