Fix auth and add V2 tests!
This commit is contained in:
Joseph Schorr
parent
afdd687192
commit
42dba8655c
5 changed files with 180 additions and 51 deletions
|
|
@ -1,5 +1,6 @@
|
|||
import unittest
|
||||
import requests
|
||||
import os
|
||||
|
||||
from flask import request, jsonify
|
||||
from flask.blueprints import Blueprint
|
||||
|
|
@ -7,6 +8,8 @@ from flask.ext.testing import LiveServerTestCase
|
|||
|
||||
from app import app
|
||||
from endpoints.v1 import v1_bp
|
||||
from endpoints.v2 import v2_bp
|
||||
from endpoints.v2.manifest import SignedManifestBuilder
|
||||
from endpoints.api import api_bp
|
||||
from initdb import wipe_database, initialize_database, populate_database
|
||||
from endpoints.csrf import generate_csrf_token
|
||||
|
|
@ -14,14 +17,20 @@ from endpoints.csrf import generate_csrf_token
|
|||
import endpoints.decorated
|
||||
import json
|
||||
import features
|
||||
import hashlib
|
||||
|
||||
import tarfile
|
||||
|
||||
from jwkest.jws import SIGNER_ALGS
|
||||
from jwkest.jwk import RSAKey
|
||||
from Crypto.PublicKey import RSA
|
||||
|
||||
from cStringIO import StringIO
|
||||
from digest.checksums import compute_simple
|
||||
|
||||
try:
|
||||
app.register_blueprint(v1_bp, url_prefix='/v1')
|
||||
app.register_blueprint(v2_bp, url_prefix='/v2')
|
||||
app.register_blueprint(api_bp, url_prefix='/api')
|
||||
except ValueError:
|
||||
# Blueprint was already registered
|
||||
|
|
@ -68,32 +77,8 @@ class TestFeature(object):
|
|||
data=json.dumps(dict(value=self.old_value)),
|
||||
headers={'Content-Type': 'application/json'})
|
||||
|
||||
class RegistryTestCase(LiveServerTestCase):
|
||||
maxDiff = None
|
||||
|
||||
def create_app(self):
|
||||
app.config['TESTING'] = True
|
||||
return app
|
||||
|
||||
def setUp(self):
|
||||
# Note: We cannot use the normal savepoint-based DB setup here because we are accessing
|
||||
# different app instances remotely via a live webserver, which is multiprocess. Therefore, we
|
||||
# completely clear the database between tests.
|
||||
wipe_database()
|
||||
initialize_database()
|
||||
populate_database()
|
||||
|
||||
self.clearSession()
|
||||
|
||||
def clearSession(self):
|
||||
self.session = requests.Session()
|
||||
self.signature = None
|
||||
self.docker_token = 'true'
|
||||
|
||||
# Load the CSRF token.
|
||||
self.csrf_token = ''
|
||||
self.csrf_token = self.conduct('GET', '/__test/csrf').text
|
||||
|
||||
class V1RegistryMixin(object):
|
||||
def conduct(self, method, url, headers=None, data=None, auth=None, expected_code=200):
|
||||
headers = headers or {}
|
||||
headers['X-Docker-Token'] = self.docker_token
|
||||
|
|
@ -118,17 +103,6 @@ class RegistryTestCase(LiveServerTestCase):
|
|||
def ping(self):
|
||||
self.conduct('GET', '/v1/_ping')
|
||||
|
||||
def do_login(self, username, password='password'):
|
||||
self.ping()
|
||||
result = self.conduct('POST', '/v1/users/',
|
||||
data=json.dumps(dict(username=username, password=password,
|
||||
email='bar@example.com')),
|
||||
headers={"Content-Type": "application/json"},
|
||||
expected_code=400)
|
||||
|
||||
self.assertEquals(result.text, '"Username or email already exists"')
|
||||
self.conduct('GET', '/v1/users/', auth=(username, password))
|
||||
|
||||
def do_push(self, namespace, repository, username, password, images):
|
||||
auth = (username, password)
|
||||
|
||||
|
|
@ -201,6 +175,147 @@ class RegistryTestCase(LiveServerTestCase):
|
|||
self.conduct('GET', image_prefix + 'json')
|
||||
self.conduct('GET', image_prefix + 'layer')
|
||||
|
||||
def clearSession(self):
|
||||
self.signature = None
|
||||
self.docker_token = 'true'
|
||||
|
||||
|
||||
class V2RegistryMixin(object):
|
||||
def conduct(self, method, url, headers=None, params=None, data=None, auth=None, expected_code=200):
|
||||
headers = headers or {}
|
||||
params = params or {}
|
||||
params['_csrf_token'] = self.csrf_token
|
||||
|
||||
if self.docker_token and not auth:
|
||||
headers['Authorization'] = 'Bearer ' + self.docker_token
|
||||
|
||||
response = self.session.request(method, self.get_server_url() + url, headers=headers, data=data,
|
||||
auth=auth, params=params)
|
||||
if response.status_code != expected_code:
|
||||
print response.text
|
||||
|
||||
self.assertEquals(response.status_code, expected_code)
|
||||
return response
|
||||
|
||||
def ping(self):
|
||||
self.conduct('GET', '/v2/', expected_code=200 if self.docker_token else 401)
|
||||
|
||||
|
||||
def do_auth(self, username, password, namespace, repository, expected_code=200, scopes=[]):
|
||||
auth = (username, password)
|
||||
params = {
|
||||
'account': username,
|
||||
'scope': 'repository:%s/%s:%s' % (namespace, repository, ','.join(scopes)),
|
||||
'service': 'quay'
|
||||
}
|
||||
|
||||
response = self.conduct('GET', '/v2/auth', params=params, auth=(username, password),
|
||||
expected_code=expected_code)
|
||||
|
||||
if expected_code == 200:
|
||||
response_json = json.loads(response.text)
|
||||
self.assertIsNotNone(response_json.get('token'))
|
||||
self.docker_token = response_json['token']
|
||||
|
||||
|
||||
def do_push(self, namespace, repository, username, password, images):
|
||||
# Ping!
|
||||
self.ping()
|
||||
|
||||
# Auth.
|
||||
self.do_auth(username, password, namespace, repository, scopes=['push', 'pull'])
|
||||
|
||||
# Build a fake manifest.
|
||||
images = [('somelayer', 'some fake data')]
|
||||
|
||||
tag_name = 'latest'
|
||||
builder = SignedManifestBuilder(namespace, repository, tag_name)
|
||||
for image_id, contents in images:
|
||||
checksum = 'sha256:' + hashlib.sha256(contents).hexdigest()
|
||||
builder.add_layer(checksum, json.dumps({'id': image_id, 'data': contents}))
|
||||
|
||||
# Push the image's layers.
|
||||
for image_id, contents in images:
|
||||
# Layer data should not yet exist.
|
||||
checksum = 'sha256:' + hashlib.sha256(contents).hexdigest()
|
||||
self.conduct('HEAD', '/v2/%s/%s/blobs/%s' % (namespace, repository, checksum),
|
||||
expected_code=404)
|
||||
|
||||
# Start a new upload of the layer data.
|
||||
response = self.conduct('POST', '/v2/%s/%s/blobs/uploads/' % (namespace, repository),
|
||||
expected_code=202)
|
||||
|
||||
location = response.headers['Location'][len(self.get_server_url()):]
|
||||
|
||||
# PATCH the image data into the layer.
|
||||
self.conduct('PATCH', location, data=contents, expected_code=204)
|
||||
|
||||
# Finish the layer upload with a PUT.
|
||||
self.conduct('PUT', location, params=dict(digest=checksum), expected_code=201)
|
||||
|
||||
# Write the manifest.
|
||||
new_key = RSA.generate(2048)
|
||||
jwk = RSAKey(key=new_key)
|
||||
manifest = builder.build(jwk)
|
||||
|
||||
self.conduct('PUT', '/v2/%s/%s/manifests/%s' % (namespace, repository, tag_name),
|
||||
data=manifest.bytes, expected_code=202,
|
||||
headers={'Content-Type': 'application/json'})
|
||||
|
||||
|
||||
def do_pull(self, namespace, repository, username=None, password='password', expected_code=200):
|
||||
auth = None
|
||||
if username:
|
||||
auth = (username, password)
|
||||
|
||||
# Ping!
|
||||
self.ping()
|
||||
|
||||
# Auth.
|
||||
self.do_auth(username, password, namespace, repository, scopes=['pull'],
|
||||
expected_code=expected_code)
|
||||
if expected_code != 200:
|
||||
return
|
||||
|
||||
# Retrieve the manifest for the tag.
|
||||
tag_name = 'latest'
|
||||
response = self.conduct('GET', '/v2/%s/%s/manifests/%s' % (namespace, repository, tag_name))
|
||||
manifest_data = json.loads(response.text)
|
||||
for layer in manifest_data['fsLayers']:
|
||||
blob_id = layer['blobSum']
|
||||
self.conduct('GET', '/v2/%s/%s/blobs/%s' % (namespace, repository, blob_id), expected_code=200)
|
||||
|
||||
|
||||
def clearSession(self):
|
||||
self.docker_token = None
|
||||
|
||||
|
||||
class RegistryTestCaseMixin(object):
|
||||
maxDiff = None
|
||||
|
||||
def create_app(self):
|
||||
app.config['TESTING'] = True
|
||||
app.config['DEBUG'] = True
|
||||
return app
|
||||
|
||||
def setUp(self):
|
||||
# Note: We cannot use the normal savepoint-based DB setup here because we are accessing
|
||||
# different app instances remotely via a live webserver, which is multiprocess. Therefore, we
|
||||
# completely clear the database between tests.
|
||||
wipe_database()
|
||||
initialize_database()
|
||||
populate_database()
|
||||
|
||||
self.clearTestSession()
|
||||
|
||||
def clearTestSession(self):
|
||||
self.session = requests.Session()
|
||||
self.clearSession()
|
||||
|
||||
# Load the CSRF token.
|
||||
self.csrf_token = ''
|
||||
self.csrf_token = self.conduct('GET', '/__test/csrf').text
|
||||
|
||||
def conduct_api_login(self, username, password):
|
||||
self.conduct('POST', '/api/v1/signin',
|
||||
data=json.dumps(dict(username=username, password=password)),
|
||||
|
|
@ -212,7 +327,7 @@ class RegistryTestCase(LiveServerTestCase):
|
|||
headers={'Content-Type': 'application/json'})
|
||||
|
||||
|
||||
class RegistryTests(RegistryTestCase):
|
||||
class RegistryTestsMixin(object):
|
||||
def test_pull_publicrepo_anonymous(self):
|
||||
# Add a new repository under the public user, so we have a real repository to pull.
|
||||
images = [{
|
||||
|
|
@ -385,5 +500,13 @@ class RegistryTests(RegistryTestCase):
|
|||
# org.
|
||||
self.do_pull('buynlarge', 'newrepo', 'devtable', 'password')
|
||||
|
||||
|
||||
class V1RegistryTests(V1RegistryMixin, RegistryTestsMixin, RegistryTestCaseMixin, LiveServerTestCase):
|
||||
""" Tests for V1 registry. """
|
||||
|
||||
class V2RegistryTests(V2RegistryMixin, RegistryTestsMixin, RegistryTestCaseMixin, LiveServerTestCase):
|
||||
""" Tests for V2 registry. """
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
Reference in a new issue