authentication: enabled authentication with auth_token and repository scope.

- fixes #3932
- vcs type tokens now with repo scope narrows the usage to a repository.
This commit is contained in:
Marcin Kuzminski 2017-03-03 20:48:23 +01:00
parent 94cd7b2237
commit 8732733a54
7 changed files with 169 additions and 18 deletions

View file

@ -132,11 +132,14 @@ def repos(request, pylonsapp):
@pytest.fixture(scope="module")
def rc_web_server_config(pylons_config):
def rc_web_server_config(testini_factory):
"""
Configuration file used for the fixture `rc_web_server`.
"""
return pylons_config
CUSTOM_PARAMS = [
{'handler_console': {'level': 'DEBUG'}},
]
return testini_factory(CUSTOM_PARAMS)
@pytest.fixture(scope="module")
@ -150,7 +153,8 @@ def rc_web_server(
env = os.environ.copy()
env['RC_NO_TMP_PATH'] = '1'
server_out = open(RC_LOG, 'w')
rc_log = RC_LOG
server_out = open(rc_log, 'w')
# TODO: Would be great to capture the output and err of the subprocess
# and make it available in a section of the py.test report in case of an
@ -158,11 +162,11 @@ def rc_web_server(
host_url = 'http://' + get_host_url(rc_web_server_config)
assert_no_running_instance(host_url)
command = ['rcserver', rc_web_server_config]
command = ['pserve', rc_web_server_config]
print('Starting rcserver: {}'.format(host_url))
print('Command: {}'.format(command))
print('Logfile: {}'.format(RC_LOG))
print('Logfile: {}'.format(rc_log))
proc = subprocess32.Popen(
command, bufsize=0, env=env, stdout=server_out, stderr=server_out)
@ -173,8 +177,9 @@ def rc_web_server(
def stop_web_server():
# TODO: Find out how to integrate with the reporting of py.test to
# make this information available.
print "\nServer log file written to %s" % (RC_LOG, )
print("\nServer log file written to %s" % (rc_log, ))
proc.kill()
server_out.flush()
server_out.close()
return RcWebServer(rc_web_server_config)
@ -210,12 +215,17 @@ def enable_auth_plugins(request, pylonsapp, csrf_token):
override = override or {}
params = {
'auth_plugins': ','.join(plugins_list),
'csrf_token': csrf_token,
}
# helper translate some names to others
name_map = {
'token': 'authtoken'
}
for module in plugins_list:
plugin = rhodecode.authentication.base.loadplugin(module)
plugin_name = plugin.name
plugin_name = module.partition('#')[-1]
if plugin_name in name_map:
plugin_name = name_map[plugin_name]
enabled_plugin = 'auth_%s_enabled' % plugin_name
cache_ttl = 'auth_%s_cache_ttl' % plugin_name

View file

@ -35,6 +35,7 @@ import pytest
from rhodecode.lib.vcs.backends.git.repository import GitRepository
from rhodecode.lib.vcs.nodes import FileNode
from rhodecode.model.auth_token import AuthTokenModel
from rhodecode.model.db import Repository, UserIpMap, CacheKey
from rhodecode.model.meta import Session
from rhodecode.model.user import UserModel
@ -46,7 +47,7 @@ from rhodecode.tests.other.vcs_operations import (
@pytest.mark.usefixtures("disable_locking")
class TestVCSOperations:
class TestVCSOperations(object):
def test_clone_hg_repo_by_admin(self, rc_web_server, tmpdir):
clone_url = rc_web_server.repo_clone_url(HG_REPO)
@ -322,6 +323,115 @@ class TestVCSOperations:
cmd.assert_returncode_success()
_check_proper_clone(stdout, stderr, 'git')
def test_clone_by_auth_token(
self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
'egg:rhodecode-enterprise-ce#rhodecode'])
user = user_util.create_user()
token = user.auth_tokens[1]
clone_url = rc_web_server.repo_clone_url(
HG_REPO, user=user.username, passwd=token)
stdout, stderr = Command('/tmp').execute(
'hg clone', clone_url, tmpdir.strpath)
_check_proper_clone(stdout, stderr, 'hg')
def test_clone_by_auth_token_expired(
self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
'egg:rhodecode-enterprise-ce#rhodecode'])
user = user_util.create_user()
auth_token = AuthTokenModel().create(
user.user_id, 'test-token', -10, AuthTokenModel.cls.ROLE_VCS)
token = auth_token.api_key
clone_url = rc_web_server.repo_clone_url(
HG_REPO, user=user.username, passwd=token)
stdout, stderr = Command('/tmp').execute(
'hg clone', clone_url, tmpdir.strpath)
assert 'abort: authorization failed' in stderr
def test_clone_by_auth_token_bad_role(
self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
'egg:rhodecode-enterprise-ce#rhodecode'])
user = user_util.create_user()
auth_token = AuthTokenModel().create(
user.user_id, 'test-token', -1, AuthTokenModel.cls.ROLE_API)
token = auth_token.api_key
clone_url = rc_web_server.repo_clone_url(
HG_REPO, user=user.username, passwd=token)
stdout, stderr = Command('/tmp').execute(
'hg clone', clone_url, tmpdir.strpath)
assert 'abort: authorization failed' in stderr
def test_clone_by_auth_token_user_disabled(
self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
'egg:rhodecode-enterprise-ce#rhodecode'])
user = user_util.create_user()
user.active = False
Session().add(user)
Session().commit()
token = user.auth_tokens[1]
clone_url = rc_web_server.repo_clone_url(
HG_REPO, user=user.username, passwd=token)
stdout, stderr = Command('/tmp').execute(
'hg clone', clone_url, tmpdir.strpath)
assert 'abort: authorization failed' in stderr
def test_clone_by_auth_token_with_scope(
self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
'egg:rhodecode-enterprise-ce#rhodecode'])
user = user_util.create_user()
auth_token = AuthTokenModel().create(
user.user_id, 'test-token', -1, AuthTokenModel.cls.ROLE_VCS)
token = auth_token.api_key
# manually set scope
auth_token.repo = Repository.get_by_repo_name(HG_REPO)
Session().add(auth_token)
Session().commit()
clone_url = rc_web_server.repo_clone_url(
HG_REPO, user=user.username, passwd=token)
stdout, stderr = Command('/tmp').execute(
'hg clone', clone_url, tmpdir.strpath)
_check_proper_clone(stdout, stderr, 'hg')
def test_clone_by_auth_token_with_wrong_scope(
self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
'egg:rhodecode-enterprise-ce#rhodecode'])
user = user_util.create_user()
auth_token = AuthTokenModel().create(
user.user_id, 'test-token', -1, AuthTokenModel.cls.ROLE_VCS)
token = auth_token.api_key
# manually set scope
auth_token.repo = Repository.get_by_repo_name(GIT_REPO)
Session().add(auth_token)
Session().commit()
clone_url = rc_web_server.repo_clone_url(
HG_REPO, user=user.username, passwd=token)
stdout, stderr = Command('/tmp').execute(
'hg clone', clone_url, tmpdir.strpath)
assert 'abort: authorization failed' in stderr
def test_git_sets_default_branch_if_not_master(
backend_git, tmpdir, disable_locking, rc_web_server):