""" Manage the tags of a repository. """ from flask import request, abort from endpoints.api import (resource, nickname, require_repo_read, require_repo_write, RepositoryParamResource, log_action, validate_json_request, path_param, parse_args, query_param, truthy_bool, disallow_for_app_repositories) from endpoints.exception import NotFound from endpoints.api.image import image_view from data import model from auth.auth_context import get_authenticated_user from util.names import TAG_ERROR, TAG_REGEX @resource('/v1/repository//tag/') @path_param('repository', 'The full path of the repository. e.g. namespace/name') class ListRepositoryTags(RepositoryParamResource): """ Resource for listing full repository tag history, alive *and dead*. """ @require_repo_read @disallow_for_app_repositories @parse_args() @query_param('specificTag', 'Filters the tags to the specific tag.', type=str, default='') @query_param('limit', 'Limit to the number of results to return per page. Max 100.', type=int, default=50) @query_param('page', 'Page index for the results. Default 1.', type=int, default=1) @nickname('listRepoTags') def get(self, namespace, repository, parsed_args): repo = model.repository.get_repository(namespace, repository) if not repo: raise NotFound() def tag_view(tag): tag_info = { 'name': tag.name, 'docker_image_id': tag.image.docker_image_id, 'reversion': tag.reversion, } if tag.lifetime_start_ts > 0: tag_info['start_ts'] = tag.lifetime_start_ts if tag.lifetime_end_ts > 0: tag_info['end_ts'] = tag.lifetime_end_ts if tag.id in manifest_map: tag_info['manifest_digest'] = manifest_map[tag.id].digest return tag_info specific_tag = parsed_args.get('specificTag') or None page = max(1, parsed_args.get('page', 1)) limit = min(100, max(1, parsed_args.get('limit', 50))) tags, manifest_map, more = model.tag.list_repository_tag_history(repo, page=page, size=limit, specific_tag=specific_tag) return { 'tags': [tag_view(tag) for tag in tags], 'page': page, 'has_additional': more, } @resource('/v1/repository//tag/') @path_param('repository', 'The full path of the repository. e.g. namespace/name') @path_param('tag', 'The name of the tag') class RepositoryTag(RepositoryParamResource): """ Resource for managing repository tags. """ schemas = { 'MoveTag': { 'type': 'object', 'description': 'Description of to which image a new or existing tag should point', 'required': [ 'image', ], 'properties': { 'image': { 'type': 'string', 'description': 'Image identifier to which the tag should point', }, }, }, } @require_repo_write @disallow_for_app_repositories @nickname('changeTagImage') @validate_json_request('MoveTag') def put(self, namespace, repository, tag): """ Change which image a tag points to or create a new tag.""" if not TAG_REGEX.match(tag): abort(400, TAG_ERROR) image_id = request.get_json()['image'] image = model.image.get_repo_image(namespace, repository, image_id) if not image: raise NotFound() original_image_id = None try: original_tag_image = model.tag.get_repo_tag_image(image.repository, tag) if original_tag_image: original_image_id = original_tag_image.docker_image_id except model.DataModelException: # This is a new tag. pass model.tag.create_or_update_tag(namespace, repository, tag, image_id) username = get_authenticated_user().username log_action('move_tag' if original_image_id else 'create_tag', namespace, {'username': username, 'repo': repository, 'tag': tag, 'namespace': namespace, 'image': image_id, 'original_image': original_image_id}, repo=model.repository.get_repository(namespace, repository)) return 'Updated', 201 @require_repo_write @disallow_for_app_repositories @nickname('deleteFullTag') def delete(self, namespace, repository, tag): """ Delete the specified repository tag. """ model.tag.delete_tag(namespace, repository, tag) username = get_authenticated_user().username log_action('delete_tag', namespace, {'username': username, 'repo': repository, 'namespace': namespace, 'tag': tag}, repo=model.repository.get_repository(namespace, repository)) return '', 204 @resource('/v1/repository//tag//images') @path_param('repository', 'The full path of the repository. e.g. namespace/name') @path_param('tag', 'The name of the tag') class RepositoryTagImages(RepositoryParamResource): """ Resource for listing the images in a specific repository tag. """ @require_repo_read @nickname('listTagImages') @disallow_for_app_repositories @parse_args() @query_param('owned', 'If specified, only images wholely owned by this tag are returned.', type=truthy_bool, default=False) def get(self, namespace, repository, tag, parsed_args): """ List the images for the specified repository tag. """ try: tag_image = model.tag.get_tag_image(namespace, repository, tag) except model.DataModelException: raise NotFound() parent_images = model.image.get_parent_images(namespace, repository, tag_image) image_map = {} image_map[str(tag_image.id)] = tag_image for image in parent_images: image_map[str(image.id)] = image image_map_all = dict(image_map) all_images = [tag_image] + list(parent_images) # Filter the images returned to those not found in the ancestry of any of the other tags in # the repository. if parsed_args['owned']: all_tags = model.tag.list_repository_tags(namespace, repository) for current_tag in all_tags: if current_tag.name == tag: continue # Remove the tag's image ID. tag_image_id = str(current_tag.image_id) image_map.pop(tag_image_id, None) # Remove any ancestors: for ancestor_id in current_tag.image.ancestors.split('/'): image_map.pop(ancestor_id, None) return { 'images': [image_view(image, image_map_all) for image in all_images if not parsed_args['owned'] or (str(image.id) in image_map)] } @resource('/v1/repository//tag//restore') @path_param('repository', 'The full path of the repository. e.g. namespace/name') @path_param('tag', 'The name of the tag') class RestoreTag(RepositoryParamResource): """ Resource for restoring a repository tag back to a previous image. """ schemas = { 'RestoreTag': { 'type': 'object', 'description': 'Restores a tag to a specific image', 'required': [ 'image', ], 'properties': { 'image': { 'type': 'string', 'description': 'Image identifier to which the tag should point', }, 'manifest_digest': { 'type': 'string', 'description': 'If specified, the manifest digest that should be used', }, }, }, } @require_repo_write @disallow_for_app_repositories @nickname('restoreTag') @validate_json_request('RestoreTag') def post(self, namespace, repository, tag): """ Restores a repository tag back to a previous image in the repository. """ repo = model.repository.get_repository(namespace, repository) # Restore the tag back to the previous image. image_id = request.get_json()['image'] manifest_digest = request.get_json().get('manifest_digest', None) if manifest_digest is not None: existing_image = model.tag.restore_tag_to_manifest(repo, tag, manifest_digest) else: existing_image = model.tag.restore_tag_to_image(repo, tag, image_id) # Log the reversion/restoration. username = get_authenticated_user().username log_data = { 'username': username, 'repo': repository, 'tag': tag, 'image': image_id, } if existing_image is not None: log_data['original_image'] = existing_image.docker_image_id log_action('revert_tag', namespace, log_data, repo=model.repository.get_repository(namespace, repository)) return { 'image_id': image_id, 'original_image_id': existing_image.docker_image_id if existing_image else None, }