""" List, create and manage repositories. """
import logging
import datetime
import features
from collections import defaultdict
from datetime import timedelta, datetime
from flask import request, abort
from app import dockerfile_build_queue, tuf_metadata_api
from data import model, oci_model
from endpoints.api import (format_date, nickname, log_action, validate_json_request,
require_repo_read, require_repo_write, require_repo_admin,
RepositoryParamResource, resource, parse_args, ApiResource,
request_error, require_scope, path_param, page_support,
query_param, truthy_bool, show_if)
from endpoints.exception import (Unauthorized, NotFound, InvalidRequest, ExceedsLicenseException,
DownstreamIssue)
from endpoints.api.billing import lookup_allowed_private_repos, get_namespace_plan
from endpoints.api.subscribe import check_repository_usage
from auth.permissions import (ModifyRepositoryPermission, AdministerRepositoryPermission,
CreateRepositoryPermission, ReadRepositoryPermission)
from auth.auth_context import get_authenticated_user
from auth import scopes
from util.names import REPOSITORY_NAME_REGEX
logger = logging.getLogger(__name__)
REPOS_PER_PAGE = 100
MAX_DAYS_IN_3_MONTHS = 92
def check_allowed_private_repos(namespace):
""" Checks to see if the given namespace has reached its private repository limit. If so,
raises a ExceedsLicenseException.
"""
# Not enabled if billing is disabled.
if not features.BILLING:
return
if not lookup_allowed_private_repos(namespace):
raise ExceedsLicenseException()
@resource('/v1/repository')
class RepositoryList(ApiResource):
"""Operations for creating and listing repositories."""
schemas = {
'NewRepo': {
'type': 'object',
'description': 'Description of a new repository',
'required': [
'repository',
'visibility',
'description',
],
'properties': {
'repository': {
'type': 'string',
'description': 'Repository name',
},
'visibility': {
'type': 'string',
'description': 'Visibility which the repository will start with',
'enum': [
'public',
'private',
],
},
'namespace': {
'type': 'string',
'description': ('Namespace in which the repository should be created. If omitted, the '
'username of the caller is used'),
},
'description': {
'type': 'string',
'description': 'Markdown encoded description for the repository',
},
'repo_kind': {
'type': ['string', 'null'],
'description': 'The kind of repository',
'enum': ['image', 'application', None],
}
},
},
}
@require_scope(scopes.CREATE_REPO)
@nickname('createRepo')
@validate_json_request('NewRepo')
def post(self):
"""Create a new repository."""
owner = get_authenticated_user()
req = request.get_json()
if owner is None and 'namespace' not in 'req':
raise InvalidRequest('Must provide a namespace or must be logged in.')
namespace_name = req['namespace'] if 'namespace' in req else owner.username
permission = CreateRepositoryPermission(namespace_name)
if permission.can():
repository_name = req['repository']
visibility = req['visibility']
existing = model.repository.get_repository(namespace_name, repository_name)
if existing:
raise request_error(message='Repository already exists')
visibility = req['visibility']
if visibility == 'private':
check_allowed_private_repos(namespace_name)
# Verify that the repository name is valid.
if not REPOSITORY_NAME_REGEX.match(repository_name):
raise InvalidRequest('Invalid repository name')
kind = req.get('repo_kind', 'image') or 'image'
repo = model.repository.create_repository(namespace_name, repository_name, owner, visibility,
repo_kind=kind)
repo.description = req['description']
repo.save()
log_action('create_repo', namespace_name, {'repo': repository_name,
'namespace': namespace_name}, repo=repo)
return {
'namespace': namespace_name,
'name': repository_name,
'kind': kind,
}, 201
raise Unauthorized()
@require_scope(scopes.READ_REPO)
@nickname('listRepos')
@parse_args()
@query_param('namespace', 'Filters the repositories returned to this namespace', type=str)
@query_param('starred', 'Filters the repositories returned to those starred by the user',
type=truthy_bool, default=False)
@query_param('public', 'Adds any repositories visible to the user by virtue of being public',
type=truthy_bool, default=False)
@query_param('last_modified', 'Whether to include when the repository was last modified.',
type=truthy_bool, default=False)
@query_param('popularity', 'Whether to include the repository\'s popularity metric.',
type=truthy_bool, default=False)
@query_param('repo_kind', 'The kind of repositories to return', type=str, default='image')
@page_support()
def get(self, page_token, parsed_args):
""" Fetch the list of repositories visible to the current user under a variety of situations.
"""
# Ensure that the user requests either filtered by a namespace, only starred repositories,
# or public repositories. This ensures that the user is not requesting *all* visible repos,
# which can cause a surge in DB CPU usage.
if not parsed_args['namespace'] and not parsed_args['starred'] and not parsed_args['public']:
raise InvalidRequest('namespace, starred or public are required for this API call')
user = get_authenticated_user()
username = user.username if user else None
next_page_token = None
repos = None
repo_kind = parsed_args['repo_kind']
# Lookup the requested repositories (either starred or non-starred.)
if parsed_args['starred']:
if not username:
# No repositories should be returned, as there is no user.
abort(400)
# Return the full list of repos starred by the current user that are still visible to them.
def can_view_repo(repo):
return ReadRepositoryPermission(repo.namespace_user.username, repo.name).can()
unfiltered_repos = model.repository.get_user_starred_repositories(user, kind_filter=repo_kind)
repos = [repo for repo in unfiltered_repos if can_view_repo(repo)]
elif parsed_args['namespace']:
# Repositories filtered by namespace do not need pagination (their results are fairly small),
# so we just do the lookup directly.
repos = list(model.repository.get_visible_repositories(username=username,
include_public=parsed_args['public'],
namespace=parsed_args['namespace'],
kind_filter=repo_kind))
else:
# Determine the starting offset for pagination. Note that we don't use the normal
# model.modelutil.paginate method here, as that does not operate over UNION queries, which
# get_visible_repositories will return if there is a logged-in user (for performance reasons).
#
# Also note the +1 on the limit, as paginate_query uses the extra result to determine whether
# there is a next page.
start_id = model.modelutil.pagination_start(page_token)
repo_query = model.repository.get_visible_repositories(username=username,
include_public=parsed_args['public'],
start_id=start_id,
limit=REPOS_PER_PAGE+1,
kind_filter=repo_kind)
repos, next_page_token = model.modelutil.paginate_query(repo_query, limit=REPOS_PER_PAGE,
id_alias='rid')
# Collect the IDs of the repositories found for subequent lookup of popularity
# and/or last modified.
if parsed_args['last_modified'] or parsed_args['popularity']:
repository_ids = [repo.rid for repo in repos]
if parsed_args['last_modified']:
last_modified_map = model.repository.get_when_last_modified(repository_ids)
if parsed_args['popularity']:
action_sum_map = model.log.get_repositories_action_sums(repository_ids)
# Collect the IDs of the repositories that are starred for the user, so we can mark them
# in the returned results.
star_set = set()
if username:
starred_repos = model.repository.get_user_starred_repositories(user)
star_set = {starred.id for starred in starred_repos}
def repo_view(repo_obj):
repo = {
'namespace': repo_obj.namespace_user.username,
'name': repo_obj.name,
'description': repo_obj.description,
'is_public': repo_obj.visibility_id == model.repository.get_public_repo_visibility().id,
'kind': repo_kind,
}
repo_id = repo_obj.rid
if parsed_args['last_modified']:
repo['last_modified'] = last_modified_map.get(repo_id)
if parsed_args['popularity']:
repo['popularity'] = float(action_sum_map.get(repo_id, 0))
if username:
repo['is_starred'] = repo_id in star_set
return repo
return {
'repositories': [repo_view(repo) for repo in repos]
}, next_page_token
@resource('/v1/repository/<apirepopath:repository>')
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
class Repository(RepositoryParamResource):
"""Operations for managing a specific repository."""
schemas = {
'RepoUpdate': {
'type': 'object',
'description': 'Fields which can be updated in a repository.',
'required': [
'description',
],
'properties': {
'description': {
'type': 'string',
'description': 'Markdown encoded description for the repository',
},
}
}
}
@parse_args()
@query_param('includeStats', 'Whether to include action statistics', type=truthy_bool,
default=False)
@require_repo_read
@nickname('getRepo')
def get(self, namespace, repository, parsed_args):
"""Fetch the specified repository."""
logger.debug('Get repo: %s/%s' % (namespace, repository))
repo = model.repository.get_repository(namespace, repository)
if repo is None:
raise NotFound()
can_write = ModifyRepositoryPermission(namespace, repository).can()
can_admin = AdministerRepositoryPermission(namespace, repository).can()
is_starred = (model.repository.repository_is_starred(get_authenticated_user(), repo)
if get_authenticated_user() else False)
is_public = model.repository.is_repository_public(repo)
# Note: This is *temporary* code for the new OCI model stuff.
if repo.kind.name == 'application':
def channel_view(channel):
return {
'name': channel.name,
'release': channel.linked_tag.name,
'last_modified': format_date(datetime.fromtimestamp(channel.linked_tag.lifetime_start / 1000)),
}
def release_view(release):
return {
'name': release.name,
'last_modified': format_date(datetime.fromtimestamp(release.lifetime_start / 1000)),
'channels': releases_channels_map[release.name],
}
channels = oci_model.channel.get_repo_channels(repo)
releases_channels_map = defaultdict(list)
for channel in channels:
releases_channels_map[channel.linked_tag.name].append(channel.name)
repo_data = {
'namespace': namespace,
'name': repository,
'kind': repo.kind.name,
'description': repo.description,
'can_write': can_write,
'can_admin': can_admin,
'is_public': is_public,
'is_organization': repo.namespace_user.organization,
'is_starred': is_starred,
'channels': [channel_view(chan) for chan in channels],
'releases': [release_view(release) for release in oci_model.release.get_release_objs(repo)],
}
return repo_data
# Older image-only repo code.
def tag_view(tag):
tag_info = {
'name': tag.name,
'image_id': tag.image.docker_image_id,
'size': tag.image.aggregate_size
}
if tag.lifetime_start_ts > 0:
last_modified = format_date(datetime.fromtimestamp(tag.lifetime_start_ts))
tag_info['last_modified'] = last_modified
if tag.lifetime_end_ts:
expiration = format_date(datetime.fromtimestamp(tag.lifetime_end_ts))
tag_info['expiration'] = expiration
if tag.tagmanifest is not None:
tag_info['manifest_digest'] = tag.tagmanifest.digest
return tag_info
stats = None
tags = model.tag.list_active_repo_tags(repo)
tag_dict = {tag.name: tag_view(tag) for tag in tags}
if parsed_args['includeStats']:
stats = []
found_dates = {}
start_date = datetime.now() - timedelta(days=MAX_DAYS_IN_3_MONTHS)
counts = model.log.get_repository_action_counts(repo, start_date)
for count in counts:
stats.append({
'date': count.date.isoformat(),
'count': count.count,
})
found_dates['%s/%s' % (count.date.month, count.date.day)] = True
# Fill in any missing stats with zeros.
for day in range(1, MAX_DAYS_IN_3_MONTHS):
day_date = datetime.now() - timedelta(days=day)
key = '%s/%s' % (day_date.month, day_date.day)
if not key in found_dates:
stats.append({
'date': day_date.date().isoformat(),
'count': 0,
})
repo_data = {
'namespace': namespace,
'name': repository,
'kind': repo.kind.name,
'description': repo.description,
'tags': tag_dict,
'can_write': can_write,
'can_admin': can_admin,
'is_public': is_public,
'is_organization': repo.namespace_user.organization,
'is_starred': is_starred,
'status_token': repo.badge_token if not is_public else '',
'trust_enabled': bool(features.SIGNING) and repo.trust_enabled,
'tag_expiration_s': repo.namespace_user.removed_tag_expiration_s,
}
if stats is not None:
repo_data['stats'] = stats
return repo_data
@require_repo_write
@nickname('updateRepo')
@validate_json_request('RepoUpdate')
def put(self, namespace, repository):
""" Update the description in the specified repository. """
repo = model.repository.get_repository(namespace, repository)
if repo:
values = request.get_json()
repo.description = values['description']
repo.save()
log_action('set_repo_description', namespace,
{'repo': repository, 'namespace': namespace, 'description': values['description']},
repo=repo)
return {
'success': True
}
raise NotFound()
@require_repo_admin
@nickname('deleteRepository')
def delete(self, namespace, repository):
""" Delete a repository. """
model.repository.purge_repository(namespace, repository)
user = model.user.get_namespace_user(namespace)
if features.BILLING:
plan = get_namespace_plan(namespace)
check_repository_usage(user, plan)
# Remove any builds from the queue.
dockerfile_build_queue.delete_namespaced_items(namespace, repository)
log_action('delete_repo', namespace,
{'repo': repository, 'namespace': namespace})
return '', 204
@resource('/v1/repository/<apirepopath:repository>/changevisibility')
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
class RepositoryVisibility(RepositoryParamResource):
""" Custom verb for changing the visibility of the repository. """
schemas = {
'ChangeVisibility': {
'type': 'object',
'description': 'Change the visibility for the repository.',
'required': [
'visibility',
],
'properties': {
'visibility': {
'type': 'string',
'description': 'Visibility which the repository will start with',
'enum': [
'public',
'private',
],
},
}
}
}
@require_repo_admin
@nickname('changeRepoVisibility')
@validate_json_request('ChangeVisibility')
def post(self, namespace, repository):
""" Change the visibility of a repository. """
repo = model.repository.get_repository(namespace, repository)
if repo:
values = request.get_json()
visibility = values['visibility']
if visibility == 'private':
check_allowed_private_repos(namespace)
model.repository.set_repository_visibility(repo, visibility)
log_action('change_repo_visibility', namespace,
{'repo': repository, 'namespace': namespace, 'visibility': values['visibility']},
repo=repo)
return {'success': True}
@resource('/v1/repository/<apirepopath:repository>/changetrust')
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
class RepositoryTrust(RepositoryParamResource):
""" Custom verb for changing the trust settings of the repository. """
schemas = {
'ChangeRepoTrust': {
'type': 'object',
'description': 'Change the trust settings for the repository.',
'required': [
'trust_enabled',
],
'properties': {
'trust_enabled': {
'type': 'boolean',
'description': 'Whether or not signing is enabled for the repository.'
},
}
}
}
@show_if(features.SIGNING)
@require_repo_admin
@nickname('changeRepoTrust')
@validate_json_request('ChangeRepoTrust')
def post(self, namespace, repository):
""" Change the visibility of a repository. """
repo = model.repository.get_repository(namespace, repository)
if not repo:
raise NotFound()
tags, _ = tuf_metadata_api.get_default_tags_with_expiration(namespace, repository)
if tags and not tuf_metadata_api.delete_metadata(namespace, repository):
raise DownstreamIssue({'message': 'Unable to delete downstream trust metadata'})
values = request.get_json()
model.repository.set_trust(repo, values['trust_enabled'])
log_action('change_repo_trust', namespace,
{'repo': repository, 'namespace': namespace, 'trust_enabled': values['trust_enabled']},
repo=repo)
return {'success': True}