vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Disallow non-apps-supported APIs for application repositories

Browse source
This commit is contained in:
Joseph Schorr 2017-03-22 14:30:13 -04:00
parent c3402fff5a
commit 30b532254c
16 changed files with 236 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

0
endpoints/api/test/__init__.py Normal file
View file

58
endpoints/api/test/shared.py Normal file
View file

@ -0,0 +1,58 @@
import datetime
import json
from contextlib import contextmanager
from data import model
from endpoints.api import api
CSRF_TOKEN_KEY = '_csrf_token'
CSRF_TOKEN = '123csrfforme'
@contextmanager
def client_with_identity(auth_username, client):
with client.session_transaction() as sess:
if auth_username:
if auth_username is not None:
loaded = model.user.get_user(auth_username)
sess['user_id'] = loaded.uuid
sess['login_time'] = datetime.datetime.now()
sess[CSRF_TOKEN_KEY] = CSRF_TOKEN
yield client
with client.session_transaction() as sess:
sess['user_id'] = None
sess['login_time'] = None
sess[CSRF_TOKEN_KEY] = None
def add_csrf_param(params):
""" Returns a params dict with the CSRF parameter added. """
params = params or {}
params[CSRF_TOKEN_KEY] = CSRF_TOKEN
return params
def conduct_api_call(client, resource, method, params, body=None, expected_code=200):
""" Conducts an API call to the given resource via the given client, and ensures its returned
status matches the code given.
Returns the response.
"""
params = add_csrf_param(params)
final_url = api.url_for(resource, **params)
headers = {}
headers.update({"Content-Type": "application/json"})
if body is not None:
body = json.dumps(body)
rv = client.open(final_url, method=method, data=body, headers=headers)
msg = '%s %s: got %s expected: %s | %s' % (method, final_url, rv.status_code, expected_code,
rv.data)
assert rv.status_code == expected_code, msg
return rv

80
endpoints/api/test/test_disallow_for_apps.py Normal file
View file

@ -0,0 +1,80 @@
import pytest
from data import model
from endpoints.api.repository import Repository
from endpoints.api.build import (RepositoryBuildList, RepositoryBuildResource,
RepositoryBuildStatus, RepositoryBuildLogs)
from endpoints.api.image import RepositoryImageList, RepositoryImage
from endpoints.api.manifest import RepositoryManifestLabels, ManageRepositoryManifestLabel
from endpoints.api.repositorynotification import (RepositoryNotification,
RepositoryNotificationList,
TestRepositoryNotification)
from endpoints.api.secscan import RepositoryImageSecurity, RepositoryManifestSecurity
from endpoints.api.tag import ListRepositoryTags, RepositoryTag, RepositoryTagImages, RestoreTag
from endpoints.api.trigger import (BuildTriggerList, BuildTrigger, BuildTriggerSubdirs,
BuildTriggerActivate, BuildTriggerAnalyze, ActivateBuildTrigger,
TriggerBuildList, BuildTriggerFieldValues, BuildTriggerSources,
BuildTriggerSourceNamespaces)
from endpoints.api.test.shared import client_with_identity, conduct_api_call
from endpoints.test.fixtures import app, appconfig, database_uri, init_db_path, sqlitedb_file
BUILD_ARGS = {'build_uuid': '1234'}
IMAGE_ARGS = {'imageid': '1234', 'image_id': 1234}
MANIFEST_ARGS = {'manifestref': 'sha256:abcd1234'}
LABEL_ARGS = {'manifestref': 'sha256:abcd1234', 'labelid': '1234'}
NOTIFICATION_ARGS = {'uuid': '1234'}
TAG_ARGS = {'tag': 'foobar'}
TRIGGER_ARGS = {'trigger_uuid': '1234'}
FIELD_ARGS = {'trigger_uuid': '1234', 'field_name': 'foobar'}
@pytest.mark.parametrize('resource, method, params', [
(Repository, 'delete', None),
(RepositoryBuildList, 'get', None),
(RepositoryBuildList, 'post', None),
(RepositoryBuildResource, 'get', BUILD_ARGS),
(RepositoryBuildResource, 'delete', BUILD_ARGS),
(RepositoryBuildStatus, 'get', BUILD_ARGS),
(RepositoryBuildLogs, 'get', BUILD_ARGS),
(RepositoryImageList, 'get', None),
(RepositoryImage, 'get', IMAGE_ARGS),
(RepositoryManifestLabels, 'get', MANIFEST_ARGS),
(RepositoryManifestLabels, 'post', MANIFEST_ARGS),
(ManageRepositoryManifestLabel, 'get', LABEL_ARGS),
(ManageRepositoryManifestLabel, 'delete', LABEL_ARGS),
(RepositoryNotificationList, 'get', None),
(RepositoryNotificationList, 'post', None),
(RepositoryNotification, 'get', NOTIFICATION_ARGS),
(RepositoryNotification, 'delete', NOTIFICATION_ARGS),
(TestRepositoryNotification, 'post', NOTIFICATION_ARGS),
(RepositoryImageSecurity, 'get', IMAGE_ARGS),
(RepositoryManifestSecurity, 'get', MANIFEST_ARGS),
(ListRepositoryTags, 'get', None),
(RepositoryTag, 'put', TAG_ARGS),
(RepositoryTag, 'delete', TAG_ARGS),
(RepositoryTagImages, 'get', TAG_ARGS),
(RestoreTag, 'post', TAG_ARGS),
(BuildTriggerList, 'get', None),
(BuildTrigger, 'get', TRIGGER_ARGS),
(BuildTrigger, 'delete', TRIGGER_ARGS),
(BuildTriggerSubdirs, 'post', TRIGGER_ARGS),
(BuildTriggerActivate, 'post', TRIGGER_ARGS),
(BuildTriggerAnalyze, 'post', TRIGGER_ARGS),
(ActivateBuildTrigger, 'post', TRIGGER_ARGS),
(TriggerBuildList, 'get', TRIGGER_ARGS),
(BuildTriggerFieldValues, 'post', FIELD_ARGS),
(BuildTriggerSources, 'post', TRIGGER_ARGS),
(BuildTriggerSourceNamespaces, 'get', TRIGGER_ARGS),
])
def test_disallowed_for_apps(resource, method, params, client):
namespace = 'devtable'
repository = 'someapprepo'
devtable = model.user.get_user('devtable')
model.repository.create_repository(namespace, repository, devtable, repo_kind='application')
params = params or {}
params['repository'] = '%s/%s' % (namespace, repository)
with client_with_identity('devtable', client) as cl:
conduct_api_call(cl, resource, method, params, None, 501)

53
endpoints/api/test/test_security.py
View file

@ -1,44 +1,29 @@
import datetime
import pytest
from data import model
from endpoints.api import api
from endpoints.api.test.shared import client_with_identity, conduct_api_call
from endpoints.api.superuser import SuperUserRepositoryBuildLogs, SuperUserRepositoryBuildResource
from endpoints.api.superuser import SuperUserRepositoryBuildStatus
from endpoints.test.fixtures import app, appconfig, database_uri, init_db_path, sqlitedb_file
TEAM_PARAMS = {'orgname': 'buynlarge', 'teamname': 'owners'}
BUILD_PARAMS = {'build_uuid': 'test-1234'}
def client_with_identity(auth_username, client):
with client.session_transaction() as sess:
if auth_username:
if auth_username is not None:
loaded = model.user.get_user(auth_username)
sess['user_id'] = loaded.uuid
sess['login_time'] = datetime.datetime.now()
return client
@pytest.mark.parametrize('resource,method,params,body,identity,expected', [
(SuperUserRepositoryBuildLogs, 'GET', BUILD_PARAMS, None, None, 401),
(SuperUserRepositoryBuildLogs, 'GET', BUILD_PARAMS, None, 'freshuser', 403),
(SuperUserRepositoryBuildLogs, 'GET', BUILD_PARAMS, None, 'reader', 403),
(SuperUserRepositoryBuildLogs, 'GET', BUILD_PARAMS, None, 'devtable', 400),
(SuperUserRepositoryBuildStatus, 'GET', BUILD_PARAMS, None, None, 401),
(SuperUserRepositoryBuildStatus, 'GET', BUILD_PARAMS, None, 'freshuser', 403),
(SuperUserRepositoryBuildStatus, 'GET', BUILD_PARAMS, None, 'reader', 403),
(SuperUserRepositoryBuildStatus, 'GET', BUILD_PARAMS, None, 'devtable', 400),
@pytest.mark.parametrize('resource,identity,expected', [
(SuperUserRepositoryBuildLogs, None, 401),
(SuperUserRepositoryBuildLogs, 'freshuser', 403),
(SuperUserRepositoryBuildLogs, 'reader', 403),
(SuperUserRepositoryBuildLogs, 'devtable', 400),
(SuperUserRepositoryBuildStatus, None, 401),
(SuperUserRepositoryBuildStatus, 'freshuser', 403),
(SuperUserRepositoryBuildStatus, 'reader', 403),
(SuperUserRepositoryBuildStatus, 'devtable', 400),
(SuperUserRepositoryBuildResource, None, 401),
(SuperUserRepositoryBuildResource, 'freshuser', 403),
(SuperUserRepositoryBuildResource, 'reader', 403),
(SuperUserRepositoryBuildResource, 'devtable', 404),
(SuperUserRepositoryBuildResource, 'GET', BUILD_PARAMS, None, None, 401),
(SuperUserRepositoryBuildResource, 'GET', BUILD_PARAMS, None, 'freshuser', 403),
(SuperUserRepositoryBuildResource, 'GET', BUILD_PARAMS, None, 'reader', 403),
(SuperUserRepositoryBuildResource, 'GET', BUILD_PARAMS, None, 'devtable', 404),
])
def test_super_user_build_endpoints(resource, identity, expected, client):
cl = client_with_identity(identity, client)
final_url = api.url_for(resource, build_uuid='1234')
rv = cl.open(final_url)
msg = '%s %s: %s expected: %s' % ('GET', final_url, rv.status_code, expected)
assert rv.status_code == expected, msg
def test_api_security(resource, method, params, body, identity, expected, client):
with client_with_identity(identity, client) as cl:
conduct_api_call(cl, resource, method, params, body, expected)