""" Manage the manifests of a repository. """ from app import label_validator from flask import request from endpoints.api import (resource, nickname, require_repo_read, require_repo_write, RepositoryParamResource, log_action, validate_json_request, path_param, parse_args, query_param, abort, api, disallow_for_app_repositories) from endpoints.exception import NotFound from manifest_models_pre_oci import pre_oci_model as model from data.model import InvalidLabelKeyException, InvalidMediaTypeException from digest import digest_tools from util.validation import VALID_LABEL_KEY_REGEX BASE_MANIFEST_ROUTE = '/v1/repository//manifest/' MANIFEST_DIGEST_ROUTE = BASE_MANIFEST_ROUTE.format(digest_tools.DIGEST_PATTERN) ALLOWED_LABEL_MEDIA_TYPES = ['text/plain', 'application/json'] @resource(MANIFEST_DIGEST_ROUTE + '/labels') @path_param('repository', 'The full path of the repository. e.g. namespace/name') @path_param('manifestref', 'The digest of the manifest') class RepositoryManifestLabels(RepositoryParamResource): """ Resource for listing the labels on a specific repository manifest. """ schemas = { 'AddLabel': { 'type': 'object', 'description': 'Adds a label to a manifest', 'required': [ 'key', 'value', 'media_type', ], 'properties': { 'key': { 'type': 'string', 'description': 'The key for the label', }, 'value': { 'type': 'string', 'description': 'The value for the label', }, 'media_type': { 'type': ['string', 'null'], 'description': 'The media type for this label', 'enum': ALLOWED_LABEL_MEDIA_TYPES + [None], }, }, }, } @require_repo_read @nickname('listManifestLabels') @disallow_for_app_repositories @parse_args() @query_param('filter', 'If specified, only labels matching the given prefix will be returned', type=str, default=None) def get(self, namespace_name, repository_name, manifestref, parsed_args): labels = model.get_manifest_labels(namespace_name, repository_name, manifestref, filter=parsed_args['filter']) if labels is None: raise NotFound() return { 'labels': [label.to_dict() for label in labels] } @require_repo_write @nickname('addManifestLabel') @disallow_for_app_repositories @validate_json_request('AddLabel') def post(self, namespace_name, repository_name, manifestref): """ Adds a new label into the tag manifest. """ label_data = request.get_json() # Check for any reserved prefixes. if label_validator.has_reserved_prefix(label_data['key']): abort(400, message='Label has a reserved prefix') label = None try: label = model.create_manifest_label(namespace_name, repository_name, manifestref, label_data['key'], label_data['value'], 'api', label_data['media_type']) except InvalidLabelKeyException: abort(400, message='Label is of an invalid format or missing please use %s format for labels'.format( VALID_LABEL_KEY_REGEX)) except InvalidMediaTypeException: abort(400, message='Media type is invalid please use a valid media type of text/plain or application/json') if label is None: raise NotFound() metadata = { 'id': label.uuid, 'key': label.key, 'value': label.value, 'manifest_digest': manifestref, 'media_type': label.media_type_name, 'namespace': namespace_name, 'repo': repository_name, } log_action('manifest_label_add', namespace_name, metadata, repo_name=repository_name) resp = {'label': label.to_dict()} repo_string = '%s/%s' % (namespace_name, repository_name) headers = { 'Location': api.url_for(ManageRepositoryManifestLabel, repository=repo_string, manifestref=manifestref, labelid=label.uuid), } return resp, 201, headers @resource(MANIFEST_DIGEST_ROUTE + '/labels/') @path_param('repository', 'The full path of the repository. e.g. namespace/name') @path_param('manifestref', 'The digest of the manifest') @path_param('labelid', 'The ID of the label') class ManageRepositoryManifestLabel(RepositoryParamResource): """ Resource for managing the labels on a specific repository manifest. """ @require_repo_read @nickname('getManifestLabel') @disallow_for_app_repositories def get(self, namespace_name, repository_name, manifestref, labelid): """ Retrieves the label with the specific ID under the manifest. """ label = model.get_manifest_label(namespace_name, repository_name, manifestref, labelid) if label is None: raise NotFound() return label.to_dict() @require_repo_write @nickname('deleteManifestLabel') @disallow_for_app_repositories def delete(self, namespace_name, repository_name, manifestref, labelid): """ Deletes an existing label from a manifest. """ deleted = model.delete_manifest_label(namespace_name, repository_name, manifestref, labelid) if deleted is None: raise NotFound() metadata = { 'id': labelid, 'key': deleted.key, 'value': deleted.value, 'manifest_digest': manifestref, 'namespace': namespace_name, 'repo': repository_name, } log_action('manifest_label_delete', namespace_name, metadata, repo_name=repository_name) return '', 204