|
|
# -*- coding: utf-8 -*-
|
|
|
|
|
|
# Copyright (C) 2010-2017 RhodeCode GmbH
|
|
|
#
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
|
# it under the terms of the GNU Affero General Public License, version 3
|
|
|
# (only), as published by the Free Software Foundation.
|
|
|
#
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
|
# GNU General Public License for more details.
|
|
|
#
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
#
|
|
|
# This program is dual-licensed. If you wish to learn more about the
|
|
|
# RhodeCode Enterprise Edition, including its added features, Support services,
|
|
|
# and proprietary license terms, please see https://rhodecode.com/licenses/
|
|
|
|
|
|
import pytest
|
|
|
from rhodecode.model.db import User, UserIpMap
|
|
|
from rhodecode.model.meta import Session
|
|
|
from rhodecode.model.permission import PermissionModel
|
|
|
from rhodecode.model.ssh_key import SshKeyModel
|
|
|
from rhodecode.tests import (
|
|
|
TestController, clear_all_caches, assert_session_flash)
|
|
|
|
|
|
|
|
|
def route_path(name, params=None, **kwargs):
|
|
|
import urllib
|
|
|
from rhodecode.apps._base import ADMIN_PREFIX
|
|
|
|
|
|
base_url = {
|
|
|
'edit_user_ips':
|
|
|
ADMIN_PREFIX + '/users/{user_id}/edit/ips',
|
|
|
'edit_user_ips_add':
|
|
|
ADMIN_PREFIX + '/users/{user_id}/edit/ips/new',
|
|
|
'edit_user_ips_delete':
|
|
|
ADMIN_PREFIX + '/users/{user_id}/edit/ips/delete',
|
|
|
|
|
|
'admin_permissions_application':
|
|
|
ADMIN_PREFIX + '/permissions/application',
|
|
|
'admin_permissions_application_update':
|
|
|
ADMIN_PREFIX + '/permissions/application/update',
|
|
|
|
|
|
'admin_permissions_global':
|
|
|
ADMIN_PREFIX + '/permissions/global',
|
|
|
'admin_permissions_global_update':
|
|
|
ADMIN_PREFIX + '/permissions/global/update',
|
|
|
|
|
|
'admin_permissions_object':
|
|
|
ADMIN_PREFIX + '/permissions/object',
|
|
|
'admin_permissions_object_update':
|
|
|
ADMIN_PREFIX + '/permissions/object/update',
|
|
|
|
|
|
'admin_permissions_ips':
|
|
|
ADMIN_PREFIX + '/permissions/ips',
|
|
|
'admin_permissions_overview':
|
|
|
ADMIN_PREFIX + '/permissions/overview',
|
|
|
|
|
|
'admin_permissions_ssh_keys':
|
|
|
ADMIN_PREFIX + '/permissions/ssh_keys',
|
|
|
'admin_permissions_ssh_keys_data':
|
|
|
ADMIN_PREFIX + '/permissions/ssh_keys/data',
|
|
|
'admin_permissions_ssh_keys_update':
|
|
|
ADMIN_PREFIX + '/permissions/ssh_keys/update'
|
|
|
|
|
|
}[name].format(**kwargs)
|
|
|
|
|
|
if params:
|
|
|
base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
|
|
|
return base_url
|
|
|
|
|
|
|
|
|
class TestAdminPermissionsController(TestController):
|
|
|
|
|
|
@pytest.fixture(scope='class', autouse=True)
|
|
|
def prepare(self, request):
|
|
|
# cleanup and reset to default permissions after
|
|
|
@request.addfinalizer
|
|
|
def cleanup():
|
|
|
PermissionModel().create_default_user_permissions(
|
|
|
User.get_default_user(), force=True)
|
|
|
|
|
|
def test_index_application(self):
|
|
|
self.log_user()
|
|
|
self.app.get(route_path('admin_permissions_application'))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
'anonymous, default_register, default_register_message, default_password_reset,'
|
|
|
'default_extern_activate, expect_error, expect_form_error', [
|
|
|
(True, 'hg.register.none', '', 'hg.password_reset.enabled', 'hg.extern_activate.manual',
|
|
|
False, False),
|
|
|
(True, 'hg.register.manual_activate', '', 'hg.password_reset.enabled', 'hg.extern_activate.auto',
|
|
|
False, False),
|
|
|
(True, 'hg.register.auto_activate', '', 'hg.password_reset.enabled', 'hg.extern_activate.manual',
|
|
|
False, False),
|
|
|
(True, 'hg.register.auto_activate', '', 'hg.password_reset.enabled', 'hg.extern_activate.manual',
|
|
|
False, False),
|
|
|
(True, 'hg.register.XXX', '', 'hg.password_reset.enabled', 'hg.extern_activate.manual',
|
|
|
False, True),
|
|
|
(True, '', '', 'hg.password_reset.enabled', '', True, False),
|
|
|
])
|
|
|
def test_update_application_permissions(
|
|
|
self, anonymous, default_register, default_register_message, default_password_reset,
|
|
|
default_extern_activate, expect_error, expect_form_error):
|
|
|
|
|
|
self.log_user()
|
|
|
|
|
|
# TODO: anonymous access set here to False, breaks some other tests
|
|
|
params = {
|
|
|
'csrf_token': self.csrf_token,
|
|
|
'anonymous': anonymous,
|
|
|
'default_register': default_register,
|
|
|
'default_register_message': default_register_message,
|
|
|
'default_password_reset': default_password_reset,
|
|
|
'default_extern_activate': default_extern_activate,
|
|
|
}
|
|
|
response = self.app.post(route_path('admin_permissions_application_update'),
|
|
|
params=params)
|
|
|
if expect_form_error:
|
|
|
assert response.status_int == 200
|
|
|
response.mustcontain('Value must be one of')
|
|
|
else:
|
|
|
if expect_error:
|
|
|
msg = 'Error occurred during update of permissions'
|
|
|
else:
|
|
|
msg = 'Application permissions updated successfully'
|
|
|
assert_session_flash(response, msg)
|
|
|
|
|
|
def test_index_object(self):
|
|
|
self.log_user()
|
|
|
self.app.get(route_path('admin_permissions_object'))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
'repo, repo_group, user_group, expect_error, expect_form_error', [
|
|
|
('repository.none', 'group.none', 'usergroup.none', False, False),
|
|
|
('repository.read', 'group.read', 'usergroup.read', False, False),
|
|
|
('repository.write', 'group.write', 'usergroup.write',
|
|
|
False, False),
|
|
|
('repository.admin', 'group.admin', 'usergroup.admin',
|
|
|
False, False),
|
|
|
('repository.XXX', 'group.admin', 'usergroup.admin', False, True),
|
|
|
('', '', '', True, False),
|
|
|
])
|
|
|
def test_update_object_permissions(self, repo, repo_group, user_group,
|
|
|
expect_error, expect_form_error):
|
|
|
self.log_user()
|
|
|
|
|
|
params = {
|
|
|
'csrf_token': self.csrf_token,
|
|
|
'default_repo_perm': repo,
|
|
|
'overwrite_default_repo': False,
|
|
|
'default_group_perm': repo_group,
|
|
|
'overwrite_default_group': False,
|
|
|
'default_user_group_perm': user_group,
|
|
|
'overwrite_default_user_group': False,
|
|
|
}
|
|
|
response = self.app.post(route_path('admin_permissions_object_update'),
|
|
|
params=params)
|
|
|
if expect_form_error:
|
|
|
assert response.status_int == 200
|
|
|
response.mustcontain('Value must be one of')
|
|
|
else:
|
|
|
if expect_error:
|
|
|
msg = 'Error occurred during update of permissions'
|
|
|
else:
|
|
|
msg = 'Object permissions updated successfully'
|
|
|
assert_session_flash(response, msg)
|
|
|
|
|
|
def test_index_global(self):
|
|
|
self.log_user()
|
|
|
self.app.get(route_path('admin_permissions_global'))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
'repo_create, repo_create_write, user_group_create, repo_group_create,'
|
|
|
'fork_create, inherit_default_permissions, expect_error,'
|
|
|
'expect_form_error', [
|
|
|
('hg.create.none', 'hg.create.write_on_repogroup.false',
|
|
|
'hg.usergroup.create.false', 'hg.repogroup.create.false',
|
|
|
'hg.fork.none', 'hg.inherit_default_perms.false', False, False),
|
|
|
('hg.create.repository', 'hg.create.write_on_repogroup.true',
|
|
|
'hg.usergroup.create.true', 'hg.repogroup.create.true',
|
|
|
'hg.fork.repository', 'hg.inherit_default_perms.false',
|
|
|
False, False),
|
|
|
('hg.create.XXX', 'hg.create.write_on_repogroup.true',
|
|
|
'hg.usergroup.create.true', 'hg.repogroup.create.true',
|
|
|
'hg.fork.repository', 'hg.inherit_default_perms.false',
|
|
|
False, True),
|
|
|
('', '', '', '', '', '', True, False),
|
|
|
])
|
|
|
def test_update_global_permissions(
|
|
|
self, repo_create, repo_create_write, user_group_create,
|
|
|
repo_group_create, fork_create, inherit_default_permissions,
|
|
|
expect_error, expect_form_error):
|
|
|
self.log_user()
|
|
|
|
|
|
params = {
|
|
|
'csrf_token': self.csrf_token,
|
|
|
'default_repo_create': repo_create,
|
|
|
'default_repo_create_on_write': repo_create_write,
|
|
|
'default_user_group_create': user_group_create,
|
|
|
'default_repo_group_create': repo_group_create,
|
|
|
'default_fork_create': fork_create,
|
|
|
'default_inherit_default_permissions': inherit_default_permissions
|
|
|
}
|
|
|
response = self.app.post(route_path('admin_permissions_global_update'),
|
|
|
params=params)
|
|
|
if expect_form_error:
|
|
|
assert response.status_int == 200
|
|
|
response.mustcontain('Value must be one of')
|
|
|
else:
|
|
|
if expect_error:
|
|
|
msg = 'Error occurred during update of permissions'
|
|
|
else:
|
|
|
msg = 'Global permissions updated successfully'
|
|
|
assert_session_flash(response, msg)
|
|
|
|
|
|
def test_index_ips(self):
|
|
|
self.log_user()
|
|
|
response = self.app.get(route_path('admin_permissions_ips'))
|
|
|
# TODO: Test response...
|
|
|
response.mustcontain('All IP addresses are allowed')
|
|
|
|
|
|
def test_add_delete_ips(self):
|
|
|
self.log_user()
|
|
|
clear_all_caches()
|
|
|
|
|
|
# ADD
|
|
|
default_user_id = User.get_default_user().user_id
|
|
|
self.app.post(
|
|
|
route_path('edit_user_ips_add', user_id=default_user_id),
|
|
|
params={'new_ip': '127.0.0.0/24', 'csrf_token': self.csrf_token})
|
|
|
|
|
|
response = self.app.get(route_path('admin_permissions_ips'))
|
|
|
response.mustcontain('127.0.0.0/24')
|
|
|
response.mustcontain('127.0.0.0 - 127.0.0.255')
|
|
|
|
|
|
# DELETE
|
|
|
default_user_id = User.get_default_user().user_id
|
|
|
del_ip_id = UserIpMap.query().filter(UserIpMap.user_id ==
|
|
|
default_user_id).first().ip_id
|
|
|
|
|
|
response = self.app.post(
|
|
|
route_path('edit_user_ips_delete', user_id=default_user_id),
|
|
|
params={'del_ip_id': del_ip_id, 'csrf_token': self.csrf_token})
|
|
|
|
|
|
assert_session_flash(response, 'Removed ip address from user whitelist')
|
|
|
|
|
|
clear_all_caches()
|
|
|
response = self.app.get(route_path('admin_permissions_ips'))
|
|
|
response.mustcontain('All IP addresses are allowed')
|
|
|
response.mustcontain(no=['127.0.0.0/24'])
|
|
|
response.mustcontain(no=['127.0.0.0 - 127.0.0.255'])
|
|
|
|
|
|
def test_index_overview(self):
|
|
|
self.log_user()
|
|
|
self.app.get(route_path('admin_permissions_overview'))
|
|
|
|
|
|
def test_ssh_keys(self):
|
|
|
self.log_user()
|
|
|
self.app.get(route_path('admin_permissions_ssh_keys'), status=200)
|
|
|
|
|
|
def test_ssh_keys_data(self, user_util, xhr_header):
|
|
|
self.log_user()
|
|
|
response = self.app.get(route_path('admin_permissions_ssh_keys_data'),
|
|
|
extra_environ=xhr_header)
|
|
|
assert response.json == {u'data': [], u'draw': None,
|
|
|
u'recordsFiltered': 0, u'recordsTotal': 0}
|
|
|
|
|
|
dummy_user = user_util.create_user()
|
|
|
SshKeyModel().create(dummy_user, 'ab:cd:ef', 'KEYKEY', 'test_key')
|
|
|
Session().commit()
|
|
|
response = self.app.get(route_path('admin_permissions_ssh_keys_data'),
|
|
|
extra_environ=xhr_header)
|
|
|
assert response.json['data'][0]['fingerprint'] == 'ab:cd:ef'
|
|
|
|
|
|
def test_ssh_keys_update(self):
|
|
|
self.log_user()
|
|
|
response = self.app.post(
|
|
|
route_path('admin_permissions_ssh_keys_update'),
|
|
|
dict(csrf_token=self.csrf_token), status=302)
|
|
|
|
|
|
assert_session_flash(
|
|
|
response, 'SSH key support is disabled in .ini file')
|
|
|
|
