Fix issue with Docker 1.8.3 and pulling public repos with no creds
We now return the valid subset of auth scopes requested. Adds a test for this case and adds testing of all returned JWTs in the V2 login tests
This commit is contained in:
Joseph Schorr
parent
566a91f003
commit
8cd38569d6
5 changed files with 202 additions and 148 deletions
|
|
@ -8,9 +8,12 @@ import resumablehashlib
|
|||
import binascii
|
||||
|
||||
import Crypto.Random
|
||||
from cachetools import lru_cache
|
||||
from flask import request, jsonify
|
||||
from flask.blueprints import Blueprint
|
||||
from flask.ext.testing import LiveServerTestCase
|
||||
from cryptography.x509 import load_pem_x509_certificate
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
|
||||
from app import app, storage
|
||||
from data.database import close_db_filter, configure
|
||||
|
|
@ -24,6 +27,7 @@ from initdb import wipe_database, initialize_database, populate_database
|
|||
from endpoints.csrf import generate_csrf_token
|
||||
from tempfile import NamedTemporaryFile
|
||||
from jsonschema import validate as validate_schema
|
||||
from util.security import strictjwt
|
||||
|
||||
import endpoints.decorated
|
||||
import json
|
||||
|
|
@ -137,6 +141,23 @@ _PORT_NUMBER = 5001
|
|||
_CLEAN_DATABASE_PATH = None
|
||||
_JWK = RSAKey(key=RSA.generate(2048))
|
||||
|
||||
class FailureCodes:
|
||||
""" Defines tuples representing the HTTP status codes for various errors. The tuple
|
||||
is defined as ('errordescription', V1HTTPStatusCode, V2HTTPStatusCode). """
|
||||
|
||||
UNAUTHENTICATED = ('unauthenticated', 401, 401)
|
||||
UNAUTHORIZED = ('unauthorized', 403, 401)
|
||||
INVALID_REGISTRY = ('invalidregistry', 404, 404)
|
||||
DOES_NOT_EXIST = ('doesnotexist', 404, 404)
|
||||
INVALID_REQUEST = ('invalidrequest', 400, 400)
|
||||
|
||||
def _get_expected_code(expected_failure, version, success_status_code):
|
||||
""" Returns the HTTP status code for the expected failure under the specified protocol version
|
||||
(1 or 2). If none, returns the success status code. """
|
||||
if not expected_failure:
|
||||
return success_status_code
|
||||
|
||||
return expected_failure[version]
|
||||
|
||||
def _get_repo_name(namespace, name):
|
||||
if namespace == '':
|
||||
|
|
@ -290,7 +311,7 @@ class V1RegistryMixin(BaseRegistryMixin):
|
|||
|
||||
|
||||
class V1RegistryPushMixin(V1RegistryMixin):
|
||||
def do_push(self, namespace, repository, username, password, images=None, expected_code=201):
|
||||
def do_push(self, namespace, repository, username, password, images=None, expect_failure=None):
|
||||
images = images or self._get_default_images()
|
||||
auth = (username, password)
|
||||
repo_name = _get_repo_name(namespace, repository)
|
||||
|
|
@ -299,6 +320,7 @@ class V1RegistryPushMixin(V1RegistryMixin):
|
|||
self.v1_ping()
|
||||
|
||||
# PUT /v1/repositories/{namespace}/{repository}/
|
||||
expected_code = _get_expected_code(expect_failure, 1, 201)
|
||||
self.conduct('PUT', '/v1/repositories/%s' % repo_name,
|
||||
data=json.dumps(images), auth=auth,
|
||||
expected_code=expected_code)
|
||||
|
|
@ -341,7 +363,7 @@ class V1RegistryPushMixin(V1RegistryMixin):
|
|||
|
||||
|
||||
class V1RegistryPullMixin(V1RegistryMixin):
|
||||
def do_pull(self, namespace, repository, username=None, password='password', expected_code=200,
|
||||
def do_pull(self, namespace, repository, username=None, password='password', expect_failure=None,
|
||||
images=None):
|
||||
images = images or self._get_default_images()
|
||||
repo_name = _get_repo_name(namespace, repository)
|
||||
|
|
@ -356,6 +378,7 @@ class V1RegistryPullMixin(V1RegistryMixin):
|
|||
prefix = '/v1/repositories/%s/' % repo_name
|
||||
|
||||
# GET /v1/repositories/{namespace}/{repository}/
|
||||
expected_code = _get_expected_code(expect_failure, 1, 200)
|
||||
self.conduct('GET', prefix + 'images', auth=auth, expected_code=expected_code)
|
||||
if expected_code != 200:
|
||||
return
|
||||
|
|
@ -427,7 +450,10 @@ class V2RegistryMixin(BaseRegistryMixin):
|
|||
|
||||
|
||||
def do_auth(self, username, password, namespace, repository, expected_code=200, scopes=[]):
|
||||
auth = (username, password)
|
||||
auth = None
|
||||
if username and password:
|
||||
auth = (username, password)
|
||||
|
||||
repo_name = _get_repo_name(namespace, repository)
|
||||
|
||||
params = {
|
||||
|
|
@ -436,7 +462,7 @@ class V2RegistryMixin(BaseRegistryMixin):
|
|||
'service': app.config['SERVER_HOSTNAME'],
|
||||
}
|
||||
|
||||
response = self.conduct('GET', '/v2/auth', params=params, auth=(username, password),
|
||||
response = self.conduct('GET', '/v2/auth', params=params, auth=auth,
|
||||
expected_code=expected_code)
|
||||
|
||||
if expected_code == 200:
|
||||
|
|
@ -449,18 +475,21 @@ class V2RegistryMixin(BaseRegistryMixin):
|
|||
|
||||
class V2RegistryPushMixin(V2RegistryMixin):
|
||||
def do_push(self, namespace, repository, username, password, images=None, tag_name=None,
|
||||
cancel=False, invalid=False, expected_manifest_code=202, expected_auth_code=200,
|
||||
scopes=None):
|
||||
cancel=False, invalid=False, expect_failure=None, scopes=None):
|
||||
images = images or self._get_default_images()
|
||||
repo_name = _get_repo_name(namespace, repository)
|
||||
|
||||
# Ping!
|
||||
self.v2_ping()
|
||||
|
||||
# Auth.
|
||||
# Auth. If the expected failure is an invalid registry, in V2 we'll receive that error from
|
||||
# the auth endpoint first, rather than just the V2 requests below.
|
||||
expected_auth_code = 200
|
||||
if expect_failure == FailureCodes.INVALID_REGISTRY:
|
||||
expected_auth_code = 400
|
||||
|
||||
self.do_auth(username, password, namespace, repository, scopes=scopes or ['push', 'pull'],
|
||||
expected_code=expected_auth_code)
|
||||
|
||||
if expected_auth_code != 200:
|
||||
return
|
||||
|
||||
|
|
@ -478,6 +507,8 @@ class V2RegistryPushMixin(V2RegistryMixin):
|
|||
|
||||
builder.add_layer(checksum, json.dumps(image_data))
|
||||
|
||||
expected_code = _get_expected_code(expect_failure, 2, 404)
|
||||
|
||||
# Build the manifest.
|
||||
manifest = builder.build(_JWK)
|
||||
|
||||
|
|
@ -493,6 +524,11 @@ class V2RegistryPushMixin(V2RegistryMixin):
|
|||
self.conduct('HEAD', '/v2/%s/blobs/%s' % (repo_name, checksum),
|
||||
expected_code=404, auth='jwt')
|
||||
|
||||
# If we expected a non-404 status code, then the HEAD operation has failed and we cannot
|
||||
# continue performing the push.
|
||||
if expected_code != 404:
|
||||
return
|
||||
|
||||
# Start a new upload of the layer data.
|
||||
response = self.conduct('POST', '/v2/%s/blobs/uploads/' % repo_name,
|
||||
expected_code=202, auth='jwt')
|
||||
|
|
@ -548,8 +584,9 @@ class V2RegistryPushMixin(V2RegistryMixin):
|
|||
self.assertEquals(response.headers['Docker-Content-Digest'], checksum)
|
||||
self.assertEquals(response.headers['Content-Length'], str(len(layer_bytes)))
|
||||
|
||||
# Write the manifest.
|
||||
put_code = 404 if invalid else expected_manifest_code
|
||||
# Write the manifest. If we expect it to be invalid, we expect a 404 code. Otherwise, we expect
|
||||
# a 202 response for success.
|
||||
put_code = 404 if invalid else 202
|
||||
self.conduct('PUT', '/v2/%s/manifests/%s' % (repo_name, tag_name),
|
||||
data=manifest.bytes, expected_code=put_code,
|
||||
headers={'Content-Type': 'application/json'}, auth='jwt')
|
||||
|
|
@ -558,25 +595,32 @@ class V2RegistryPushMixin(V2RegistryMixin):
|
|||
|
||||
|
||||
class V2RegistryPullMixin(V2RegistryMixin):
|
||||
def do_pull(self, namespace, repository, username=None, password='password', expected_code=200,
|
||||
manifest_id=None, expected_manifest_code=200, images=None):
|
||||
def do_pull(self, namespace, repository, username=None, password='password', expect_failure=None,
|
||||
manifest_id=None, images=None):
|
||||
images = images or self._get_default_images()
|
||||
repo_name = _get_repo_name(namespace, repository)
|
||||
|
||||
# Ping!
|
||||
self.v2_ping()
|
||||
|
||||
# Auth.
|
||||
# Auth. If the failure expected is unauthenticated, then the auth endpoint will 401 before
|
||||
# we reach any of the registry operations.
|
||||
expected_auth_code = 200
|
||||
if expect_failure == FailureCodes.UNAUTHENTICATED:
|
||||
expected_auth_code = 401
|
||||
|
||||
self.do_auth(username, password, namespace, repository, scopes=['pull'],
|
||||
expected_code=expected_code)
|
||||
if expected_code != 200:
|
||||
expected_code=expected_auth_code)
|
||||
if expected_auth_code != 200:
|
||||
return
|
||||
|
||||
# Retrieve the manifest for the tag or digest.
|
||||
manifest_id = manifest_id or 'latest'
|
||||
|
||||
expected_code = _get_expected_code(expect_failure, 2, 200)
|
||||
response = self.conduct('GET', '/v2/%s/manifests/%s' % (repo_name, manifest_id),
|
||||
auth='jwt', expected_code=expected_manifest_code)
|
||||
if expected_manifest_code != 200:
|
||||
auth='jwt', expected_code=expected_code)
|
||||
if expected_code != 200:
|
||||
return
|
||||
|
||||
manifest_data = json.loads(response.text)
|
||||
|
|
@ -621,20 +665,23 @@ class V1RegistryLoginMixin(object):
|
|||
|
||||
|
||||
class V2RegistryLoginMixin(object):
|
||||
def do_login(self, username, password, scope, expect_success=True, expected_code=None):
|
||||
def do_login(self, username, password, scope, expect_success=True):
|
||||
params = {
|
||||
'account': username,
|
||||
'scope': scope,
|
||||
'service': app.config['SERVER_HOSTNAME'],
|
||||
}
|
||||
|
||||
if expected_code is None:
|
||||
if expect_success:
|
||||
expected_code = 200
|
||||
else:
|
||||
expected_code = 403
|
||||
if expect_success:
|
||||
expected_code = 200
|
||||
else:
|
||||
expected_code = 401
|
||||
|
||||
response = self.conduct('GET', '/v2/auth', params=params, auth=(username, password),
|
||||
auth = None
|
||||
if username and password:
|
||||
auth = (username, password)
|
||||
|
||||
response = self.conduct('GET', '/v2/auth', params=params, auth=auth,
|
||||
expected_code=expected_code)
|
||||
return response
|
||||
|
||||
|
|
@ -739,7 +786,9 @@ class RegistryTestsMixin(object):
|
|||
|
||||
# First try to pull the (currently private) repo anonymously, which should fail (since it is
|
||||
# private)
|
||||
self.do_pull('public', 'newrepo', expected_code=403)
|
||||
self.do_pull('public', 'newrepo', expect_failure=FailureCodes.UNAUTHORIZED)
|
||||
self.do_pull('public', 'newrepo', 'devtable', 'password',
|
||||
expect_failure=FailureCodes.UNAUTHORIZED)
|
||||
|
||||
# Make the repository public.
|
||||
self.conduct_api_login('public', 'password')
|
||||
|
|
@ -757,7 +806,8 @@ class RegistryTestsMixin(object):
|
|||
|
||||
# First try to pull the (currently private) repo as devtable, which should fail as it belongs
|
||||
# to public.
|
||||
self.do_pull('public', 'newrepo', 'devtable', 'password', expected_code=403)
|
||||
self.do_pull('public', 'newrepo', 'devtable', 'password',
|
||||
expect_failure=FailureCodes.UNAUTHORIZED)
|
||||
|
||||
# Make the repository public.
|
||||
self.conduct_api_login('public', 'password')
|
||||
|
|
@ -775,7 +825,8 @@ class RegistryTestsMixin(object):
|
|||
|
||||
# First try to pull the (currently private) repo as public, which should fail as it belongs
|
||||
# to devtable.
|
||||
self.do_pull('devtable', 'newrepo', 'public', 'password', expected_code=403)
|
||||
self.do_pull('devtable', 'newrepo', 'public', 'password',
|
||||
expect_failure=FailureCodes.UNAUTHORIZED)
|
||||
|
||||
# Pull the repository as devtable, which should succeed because the repository is owned by
|
||||
# devtable.
|
||||
|
|
@ -791,7 +842,8 @@ class RegistryTestsMixin(object):
|
|||
|
||||
# First try to pull the (currently private) repo as devtable, which should fail as it belongs
|
||||
# to public.
|
||||
self.do_pull('public', 'newrepo', 'devtable', 'password', expected_code=403)
|
||||
self.do_pull('public', 'newrepo', 'devtable', 'password',
|
||||
expect_failure=FailureCodes.UNAUTHORIZED)
|
||||
|
||||
# Make the repository public.
|
||||
self.conduct_api_login('public', 'password')
|
||||
|
|
@ -811,7 +863,8 @@ class RegistryTestsMixin(object):
|
|||
|
||||
# First try to pull the (currently private) repo as devtable, which should fail as it belongs
|
||||
# to public.
|
||||
self.do_pull('public', 'newrepo', 'devtable', 'password', expected_code=403)
|
||||
self.do_pull('public', 'newrepo', 'devtable', 'password',
|
||||
expect_failure=FailureCodes.UNAUTHORIZED)
|
||||
|
||||
# Pull the repository as public, which should succeed because the repository is owned by public.
|
||||
self.do_pull('public', 'newrepo', 'public', 'password')
|
||||
|
|
@ -826,7 +879,7 @@ class RegistryTestsMixin(object):
|
|||
|
||||
# First try to pull the (currently private) repo as anonymous, which should fail as it
|
||||
# is private.
|
||||
self.do_pull('public', 'newrepo', expected_code=401)
|
||||
self.do_pull('public', 'newrepo', expect_failure=FailureCodes.UNAUTHENTICATED)
|
||||
|
||||
# Make the repository public.
|
||||
self.conduct_api_login('public', 'password')
|
||||
|
|
@ -835,7 +888,7 @@ class RegistryTestsMixin(object):
|
|||
|
||||
# Try again to pull the (currently public) repo as anonymous, which should fail as
|
||||
# anonymous access is disabled.
|
||||
self.do_pull('public', 'newrepo', expected_code=401)
|
||||
self.do_pull('public', 'newrepo', expect_failure=FailureCodes.UNAUTHENTICATED)
|
||||
|
||||
# Pull the repository as public, which should succeed because the repository is owned by public.
|
||||
self.do_pull('public', 'newrepo', 'public', 'password')
|
||||
|
|
@ -898,7 +951,8 @@ class V1RegistryTests(V1RegistryPullMixin, V1RegistryPushMixin, RegistryTestsMix
|
|||
'id': 'onlyimagehere',
|
||||
'contents': 'somecontents',
|
||||
}]
|
||||
self.do_push('public', 'newrepo/somesubrepo', 'public', 'password', images, expected_code=404)
|
||||
self.do_push('public', 'newrepo/somesubrepo', 'public', 'password', images,
|
||||
expect_failure=FailureCodes.INVALID_REGISTRY)
|
||||
|
||||
def test_push_unicode_metadata(self):
|
||||
self.conduct_api_login('devtable', 'password')
|
||||
|
|
@ -948,7 +1002,7 @@ class V2RegistryTests(V2RegistryPullMixin, V2RegistryPushMixin, RegistryTestsMix
|
|||
|
||||
# Ensure the tag no longer exists.
|
||||
self.do_pull('devtable', 'newrepo', 'devtable', 'password',
|
||||
expected_manifest_code=404)
|
||||
expect_failure=FailureCodes.DOES_NOT_EXIST)
|
||||
|
||||
def test_push_only_push_scope(self):
|
||||
images = [{
|
||||
|
|
@ -959,15 +1013,6 @@ class V2RegistryTests(V2RegistryPullMixin, V2RegistryPushMixin, RegistryTestsMix
|
|||
self.do_push('devtable', 'somenewrepo', 'devtable', 'password', images,
|
||||
scopes=['push'])
|
||||
|
||||
def test_attempt_push_only_push_scope(self):
|
||||
images = [{
|
||||
'id': 'onlyimagehere',
|
||||
'contents': 'foobar',
|
||||
}]
|
||||
|
||||
self.do_push('public', 'somenewrepo', 'devtable', 'password', images,
|
||||
scopes=['push'], expected_auth_code=403)
|
||||
|
||||
def test_push_reponame_with_slashes(self):
|
||||
# Attempt to add a repository name with slashes. This should fail as we do not support it.
|
||||
images = [{
|
||||
|
|
@ -976,7 +1021,7 @@ class V2RegistryTests(V2RegistryPullMixin, V2RegistryPushMixin, RegistryTestsMix
|
|||
}]
|
||||
|
||||
self.do_push('public', 'newrepo/somesubrepo', 'devtable', 'password', images,
|
||||
expected_auth_code=400)
|
||||
expect_failure=FailureCodes.INVALID_REGISTRY)
|
||||
|
||||
def test_invalid_push(self):
|
||||
self.do_push('devtable', 'newrepo', 'devtable', 'password', invalid=True)
|
||||
|
|
@ -999,7 +1044,7 @@ class V2RegistryTests(V2RegistryPullMixin, V2RegistryPushMixin, RegistryTestsMix
|
|||
|
||||
# Attempt to pull the invalid tag.
|
||||
self.do_pull('devtable', 'newrepo', 'devtable', 'password', manifest_id='invalid',
|
||||
expected_manifest_code=404)
|
||||
expect_failure=FailureCodes.INVALID_REGISTRY)
|
||||
|
||||
|
||||
def test_partial_upload_below_5mb(self):
|
||||
|
|
@ -1097,7 +1142,7 @@ class V2RegistryTests(V2RegistryPullMixin, V2RegistryPushMixin, RegistryTestsMix
|
|||
]
|
||||
|
||||
self.do_push('devtable', 'newrepo', 'devtable', 'password', images=images,
|
||||
expected_manifest_code=400)
|
||||
expect_failure=FailureCodes.INVALID_REQUEST)
|
||||
|
||||
def test_multiple_layers(self):
|
||||
# Push a manifest with multiple layers.
|
||||
|
|
@ -1116,7 +1161,8 @@ class V2RegistryTests(V2RegistryPullMixin, V2RegistryPushMixin, RegistryTestsMix
|
|||
self.do_push('devtable', 'newrepo', 'devtable', 'password', images=images)
|
||||
|
||||
def test_invalid_regname(self):
|
||||
self.do_push('devtable', 'this/is/a/repo', 'devtable', 'password', expected_auth_code=400)
|
||||
self.do_push('devtable', 'this/is/a/repo', 'devtable', 'password',
|
||||
expect_failure=FailureCodes.INVALID_REGISTRY)
|
||||
|
||||
def test_multiple_tags(self):
|
||||
latest_images = [
|
||||
|
|
@ -1432,49 +1478,108 @@ class V1LoginTests(V1RegistryLoginMixin, LoginTests, RegistryTestCaseMixin, Base
|
|||
class V2LoginTests(V2RegistryLoginMixin, LoginTests, RegistryTestCaseMixin, BaseRegistryMixin, LiveServerTestCase):
|
||||
""" Tests for V2 login. """
|
||||
|
||||
@staticmethod
|
||||
@lru_cache(maxsize=1)
|
||||
def load_public_key(certificate_file_path):
|
||||
with open(certificate_file_path) as cert_file:
|
||||
cert_obj = load_pem_x509_certificate(cert_file.read(), default_backend())
|
||||
return cert_obj.public_key()
|
||||
|
||||
|
||||
def do_logincheck(self, username, password, scope, expected_actions=[], expect_success=True):
|
||||
response = self.do_login(username, password, scope, expect_success=expect_success)
|
||||
|
||||
if not expect_success:
|
||||
return
|
||||
|
||||
# Validate the returned JWT.
|
||||
encoded = response.json()['token']
|
||||
|
||||
expected_issuer = app.config['JWT_AUTH_TOKEN_ISSUER']
|
||||
audience = app.config['SERVER_HOSTNAME']
|
||||
|
||||
max_signed_s = app.config.get('JWT_AUTH_MAX_FRESH_S', 3660)
|
||||
certificate_file_path = app.config['JWT_AUTH_CERTIFICATE_PATH']
|
||||
|
||||
public_key = V2LoginTests.load_public_key(certificate_file_path)
|
||||
|
||||
max_exp = strictjwt.exp_max_s_option(max_signed_s)
|
||||
payload = strictjwt.decode(encoded, public_key, algorithms=['RS256'], audience=audience,
|
||||
issuer=expected_issuer, options=max_exp)
|
||||
|
||||
if scope is None:
|
||||
self.assertEquals(0, len(payload['access']))
|
||||
else:
|
||||
self.assertEquals(1, len(payload['access']))
|
||||
self.assertEquals(payload['access'][0]['actions'], expected_actions)
|
||||
|
||||
|
||||
def test_nouser_noscope(self):
|
||||
self.do_login('', '', expected_code=401, scope='')
|
||||
self.do_logincheck('', '', expect_success=False, scope=None)
|
||||
|
||||
def test_validuser_unknownrepo(self):
|
||||
self.do_login('devtable', 'password', expect_success=False,
|
||||
scope='repository:invalidnamespace/simple:pull')
|
||||
self.do_logincheck('devtable', 'password', expect_success=True,
|
||||
scope='repository:invalidnamespace/simple:pull',
|
||||
expected_actions=[])
|
||||
|
||||
def test_validuser_unknownnamespacerepo(self):
|
||||
self.do_login('devtable', 'password', expect_success=True,
|
||||
scope='repository:devtable/newrepo:push')
|
||||
self.do_logincheck('devtable', 'password', expect_success=True,
|
||||
scope='repository:devtable/newrepo:push',
|
||||
expected_actions=['push'])
|
||||
|
||||
def test_validuser_noaccess(self):
|
||||
self.do_login('public', 'password', expect_success=False,
|
||||
scope='repository:devtable/simple:pull')
|
||||
self.do_logincheck('public', 'password', expect_success=True,
|
||||
scope='repository:devtable/simple:pull',
|
||||
expected_actions=[])
|
||||
|
||||
def test_validuser_noscope(self):
|
||||
self.do_login('public', 'password', expect_success=True, scope=None)
|
||||
self.do_logincheck('public', 'password', expect_success=True, scope=None)
|
||||
|
||||
def test_invaliduser_noscope(self):
|
||||
self.do_login('invaliduser', 'invalidpass', expected_code=401, scope=None)
|
||||
self.do_logincheck('invaliduser', 'invalidpass', expect_success=False, scope=None)
|
||||
|
||||
def test_invalidpassword_noscope(self):
|
||||
self.do_login('public', 'invalidpass', expected_code=401, scope=None)
|
||||
self.do_logincheck('public', 'invalidpass', expect_success=False, scope=None)
|
||||
|
||||
def test_oauth_noaccess(self):
|
||||
self.do_login('$oauthtoken', 'test', expect_success=False,
|
||||
scope='repository:public/publicrepo:pull,push')
|
||||
self.do_logincheck('$oauthtoken', 'test', expect_success=True,
|
||||
scope='repository:freshuser/unknownrepo:pull,push',
|
||||
expected_actions=[])
|
||||
|
||||
def test_oauth_public(self):
|
||||
self.do_logincheck('$oauthtoken', 'test', expect_success=True,
|
||||
scope='repository:public/publicrepo:pull,push',
|
||||
expected_actions=['pull'])
|
||||
|
||||
def test_nouser_pull_publicrepo(self):
|
||||
self.do_login('', '', expect_success=True, scope='repository:public/publicrepo:pull')
|
||||
self.do_logincheck('', '', expect_success=True, scope='repository:public/publicrepo:pull',
|
||||
expected_actions=['pull'])
|
||||
|
||||
def test_nouser_push_publicrepo(self):
|
||||
self.do_login('', '', expected_code=401, scope='repository:public/publicrepo:push')
|
||||
self.do_logincheck('', '', expect_success=True, scope='repository:public/publicrepo:push',
|
||||
expected_actions=[])
|
||||
|
||||
def test_library_invaliduser(self):
|
||||
self.do_login('invaliduser', 'password', expected_code=401, scope='repository:librepo:pull,push')
|
||||
self.do_logincheck('invaliduser', 'password', expect_success=False,
|
||||
scope='repository:librepo:pull,push')
|
||||
|
||||
def test_library_noaccess(self):
|
||||
self.do_login('freshuser', 'password', expected_code=403, scope='repository:librepo:pull,push')
|
||||
self.do_logincheck('freshuser', 'password', expect_success=True,
|
||||
scope='repository:librepo:pull,push',
|
||||
expected_actions=[])
|
||||
|
||||
def test_library_access(self):
|
||||
self.do_login('devtable', 'password', expect_success=200, scope='repository:librepo:pull,push')
|
||||
self.do_logincheck('devtable', 'password', expect_success=True,
|
||||
scope='repository:librepo:pull,push',
|
||||
expected_actions=['push', 'pull'])
|
||||
|
||||
def test_nouser_pushpull_publicrepo(self):
|
||||
# Note: Docker 1.8.3 will ask for both push and pull scopes at all times. For public pulls
|
||||
# with no credentials, we were returning a 401. This test makes sure we get back just a pull
|
||||
# token.
|
||||
self.do_logincheck('', '', expect_success=True,
|
||||
scope='repository:public/publicrepo:pull,push',
|
||||
expected_actions=['pull'])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
Reference in a new issue