engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

repo-schemas: refactor repository schemas and use it in API update/create functions.

Browse source 
- now it uses consistent way of serializing/validating data
- parent groups are consistenty handled by name parameter
- fixes #4133
- fixes other problems with bad data
- changes API slightly
- validation chain ordering, first permissions, then uniqness.
  Helps prevent resource disvovery
This commit is contained in:
Marcin Kuzminski 2016-11-24 23:48:10 +01:00
parent 7aa1116308
commit c081457f2c
14 changed files with 1050 additions and 322 deletions
Download patch file Download diff file Expand all files Collapse all files

122
rhodecode/api/tests/test_create_repo.py
View file

@ -23,8 +23,11 @@ import json
import mock
import pytest
from rhodecode.lib.utils2 import safe_unicode
from rhodecode.lib.vcs import settings
from rhodecode.model.meta import Session
from rhodecode.model.repo import RepoModel
from rhodecode.model.user import UserModel
from rhodecode.tests import TEST_USER_ADMIN_LOGIN
from rhodecode.api.tests.utils import (
build_data, api_call, assert_ok, assert_error, crash)
@ -36,29 +39,37 @@ fixture = Fixture()
@pytest.mark.usefixtures("testuser_api", "app")
class TestCreateRepo(object):
def test_api_create_repo(self, backend):
repo_name = 'api-repo-1'
@pytest.mark.parametrize('given, expected_name, expected_exc', [
('api repo-1', 'api-repo-1', False),
('api-repo 1-ąć', 'api-repo-1-ąć', False),
(u'unicode-ąć', u'unicode-ąć', False),
('some repo v1.2', 'some-repo-v1.2', False),
('v2.0', 'v2.0', False),
])
def test_api_create_repo(self, backend, given, expected_name, expected_exc):
id_, params = build_data(
self.apikey,
'create_repo',
repo_name=repo_name,
repo_name=given,
owner=TEST_USER_ADMIN_LOGIN,
repo_type=backend.alias,
)
response = api_call(self.app, params)
repo = RepoModel().get_by_repo_name(repo_name)
assert repo is not None
ret = {
'msg': 'Created new repository `%s`' % (repo_name,),
'msg': 'Created new repository `%s`' % (expected_name,),
'success': True,
'task': None,
}
expected = ret
assert_ok(id_, expected, given=response.body)
id_, params = build_data(self.apikey, 'get_repo', repoid=repo_name)
repo = RepoModel().get_by_repo_name(safe_unicode(expected_name))
assert repo is not None
id_, params = build_data(self.apikey, 'get_repo', repoid=expected_name)
response = api_call(self.app, params)
body = json.loads(response.body)
@ -66,7 +77,7 @@ class TestCreateRepo(object):
assert body['result']['enable_locking'] is False
assert body['result']['enable_statistics'] is False
fixture.destroy_repo(repo_name)
fixture.destroy_repo(safe_unicode(expected_name))
def test_api_create_restricted_repo_type(self, backend):
repo_name = 'api-repo-type-{0}'.format(backend.alias)
@ -158,6 +169,21 @@ class TestCreateRepo(object):
fixture.destroy_repo(repo_name)
fixture.destroy_repo_group(repo_group_name)
def test_create_repo_in_group_that_doesnt_exist(self, backend, user_util):
repo_group_name = 'fake_group'
repo_name = '%s/api-repo-gr' % (repo_group_name,)
id_, params = build_data(
self.apikey, 'create_repo',
repo_name=repo_name,
owner=TEST_USER_ADMIN_LOGIN,
repo_type=backend.alias,)
response = api_call(self.app, params)
expected = {'repo_group': 'Repository group `{}` does not exist'.format(
repo_group_name)}
assert_error(id_, expected, given=response.body)
def test_api_create_repo_unknown_owner(self, backend):
repo_name = 'api-repo-2'
owner = 'i-dont-exist'
@ -218,10 +244,48 @@ class TestCreateRepo(object):
owner=owner)
response = api_call(self.app, params)
expected = 'Only RhodeCode admin can specify `owner` param'
expected = 'Only RhodeCode super-admin can specify `owner` param'
assert_error(id_, expected, given=response.body)
fixture.destroy_repo(repo_name)
def test_api_create_repo_by_non_admin_no_parent_group_perms(self, backend):
repo_group_name = 'no-access'
fixture.create_repo_group(repo_group_name)
repo_name = 'no-access/api-repo'
id_, params = build_data(
self.apikey_regular, 'create_repo',
repo_name=repo_name,
repo_type=backend.alias)
response = api_call(self.app, params)
expected = {'repo_group': 'Repository group `{}` does not exist'.format(
repo_group_name)}
assert_error(id_, expected, given=response.body)
fixture.destroy_repo_group(repo_group_name)
fixture.destroy_repo(repo_name)
def test_api_create_repo_non_admin_no_permission_to_create_to_root_level(
self, backend, user_util):
regular_user = user_util.create_user()
regular_user_api_key = regular_user.api_key
usr = UserModel().get_by_username(regular_user.username)
usr.inherit_default_permissions = False
Session().add(usr)
repo_name = backend.new_repo_name()
id_, params = build_data(
regular_user_api_key, 'create_repo',
repo_name=repo_name,
repo_type=backend.alias)
response = api_call(self.app, params)
expected = {
"repo_name": "You do not have the permission to "
"store repositories in the root location."}
assert_error(id_, expected, given=response.body)
def test_api_create_repo_exists(self, backend):
repo_name = backend.repo_name
id_, params = build_data(
@ -230,7 +294,9 @@ class TestCreateRepo(object):
owner=TEST_USER_ADMIN_LOGIN,
repo_type=backend.alias,)
response = api_call(self.app, params)
expected = "repo `%s` already exist" % (repo_name,)
expected = {
'unique_repo_name': 'Repository with name `{}` already exists'.format(
repo_name)}
assert_error(id_, expected, given=response.body)
@mock.patch.object(RepoModel, 'create', crash)
@ -245,26 +311,40 @@ class TestCreateRepo(object):
expected = 'failed to create repository `%s`' % (repo_name,)
assert_error(id_, expected, given=response.body)
def test_create_repo_with_extra_slashes_in_name(self, backend, user_util):
existing_repo_group = user_util.create_repo_group()
dirty_repo_name = '//{}/repo_name//'.format(
existing_repo_group.group_name)
cleaned_repo_name = '{}/repo_name'.format(
existing_repo_group.group_name)
@pytest.mark.parametrize('parent_group, dirty_name, expected_name', [
(None, 'foo bar x', 'foo-bar-x'),
('foo', '/foo//bar x', 'foo/bar-x'),
('foo-bar', 'foo-bar //bar x', 'foo-bar/bar-x'),
])
def test_create_repo_with_extra_slashes_in_name(
self, backend, parent_group, dirty_name, expected_name):
if parent_group:
gr = fixture.create_repo_group(parent_group)
assert gr.group_name == parent_group
id_, params = build_data(
self.apikey, 'create_repo',
repo_name=dirty_repo_name,
repo_name=dirty_name,
repo_type=backend.alias,
owner=TEST_USER_ADMIN_LOGIN,)
response = api_call(self.app, params)
repo = RepoModel().get_by_repo_name(cleaned_repo_name)
expected ={
"msg": "Created new repository `{}`".format(expected_name),
"task": None,
"success": True
}
assert_ok(id_, expected, response.body)
repo = RepoModel().get_by_repo_name(expected_name)
assert repo is not None
expected = {
'msg': 'Created new repository `%s`' % (cleaned_repo_name,),
'msg': 'Created new repository `%s`' % (expected_name,),
'success': True,
'task': None,
}
assert_ok(id_, expected, given=response.body)
fixture.destroy_repo(cleaned_repo_name)
fixture.destroy_repo(expected_name)
if parent_group:
fixture.destroy_repo_group(parent_group)

71
rhodecode/api/tests/test_fork_repo.py
View file

@ -24,6 +24,7 @@ import pytest
from rhodecode.model.meta import Session
from rhodecode.model.repo import RepoModel
from rhodecode.model.repo_group import RepoGroupModel
from rhodecode.model.user import UserModel
from rhodecode.tests import TEST_USER_ADMIN_LOGIN
from rhodecode.api.tests.utils import (
@ -99,11 +100,35 @@ class TestApiForkRepo(object):
finally:
fixture.destroy_repo(fork_name)
def test_api_fork_repo_non_admin_into_group_no_permission(self, backend, user_util):
source_name = backend['minimal'].repo_name
repo_group = user_util.create_repo_group()
repo_group_name = repo_group.group_name
fork_name = '%s/api-repo-fork' % repo_group_name
id_, params = build_data(
self.apikey_regular, 'fork_repo',
repoid=source_name,
fork_name=fork_name)
response = api_call(self.app, params)
expected = {
'repo_group': 'Repository group `{}` does not exist'.format(
repo_group_name)}
try:
assert_error(id_, expected, given=response.body)
finally:
fixture.destroy_repo(fork_name)
def test_api_fork_repo_non_admin_into_group(self, backend, user_util):
source_name = backend['minimal'].repo_name
repo_group = user_util.create_repo_group()
fork_name = '%s/api-repo-fork' % repo_group.group_name
RepoGroupModel().grant_user_permission(
repo_group, self.TEST_USER_LOGIN, 'group.admin')
Session().commit()
id_, params = build_data(
self.apikey_regular, 'fork_repo',
repoid=source_name,
@ -129,10 +154,11 @@ class TestApiForkRepo(object):
fork_name=fork_name,
owner=TEST_USER_ADMIN_LOGIN)
response = api_call(self.app, params)
expected = 'Only RhodeCode admin can specify `owner` param'
expected = 'Only RhodeCode super-admin can specify `owner` param'
assert_error(id_, expected, given=response.body)
def test_api_fork_repo_non_admin_no_permission_to_fork(self, backend):
def test_api_fork_repo_non_admin_no_permission_of_source_repo(
self, backend):
source_name = backend['minimal'].repo_name
RepoModel().grant_user_permission(repo=source_name,
user=self.TEST_USER_LOGIN,
@ -147,19 +173,44 @@ class TestApiForkRepo(object):
assert_error(id_, expected, given=response.body)
def test_api_fork_repo_non_admin_no_permission_to_fork_to_root_level(
self, backend):
source_name = backend['minimal'].repo_name
self, backend, user_util):
usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
regular_user = user_util.create_user()
regular_user_api_key = regular_user.api_key
usr = UserModel().get_by_username(regular_user.username)
usr.inherit_default_permissions = False
Session().add(usr)
UserModel().grant_perm(regular_user.username, 'hg.fork.repository')
source_name = backend['minimal'].repo_name
fork_name = backend.new_repo_name()
id_, params = build_data(
self.apikey_regular, 'fork_repo',
regular_user_api_key, 'fork_repo',
repoid=source_name,
fork_name=fork_name)
response = api_call(self.app, params)
expected = {
"repo_name": "You do not have the permission to "
"store repositories in the root location."}
assert_error(id_, expected, given=response.body)
def test_api_fork_repo_non_admin_no_permission_to_fork(
self, backend, user_util):
regular_user = user_util.create_user()
regular_user_api_key = regular_user.api_key
usr = UserModel().get_by_username(regular_user.username)
usr.inherit_default_permissions = False
Session().add(usr)
source_name = backend['minimal'].repo_name
fork_name = backend.new_repo_name()
id_, params = build_data(
regular_user_api_key, 'fork_repo',
repoid=source_name,
fork_name=fork_name)
response = api_call(self.app, params)
expected = "Access was denied to this resource."
assert_error(id_, expected, given=response.body)
@ -189,7 +240,9 @@ class TestApiForkRepo(object):
response = api_call(self.app, params)
try:
expected = "fork `%s` already exist" % (fork_name,)
expected = {
'unique_repo_name': 'Repository with name `{}` already exists'.format(
fork_name)}
assert_error(id_, expected, given=response.body)
finally:
fixture.destroy_repo(fork_repo.repo_name)
@ -205,7 +258,9 @@ class TestApiForkRepo(object):
owner=TEST_USER_ADMIN_LOGIN)
response = api_call(self.app, params)
expected = "repo `%s` already exist" % (fork_name,)
expected = {
'unique_repo_name': 'Repository with name `{}` already exists'.format(
fork_name)}
assert_error(id_, expected, given=response.body)
@mock.patch.object(RepoModel, 'create_fork', crash)

86
rhodecode/api/tests/test_update_repo.py
View file

@ -32,35 +32,60 @@ fixture = Fixture()
UPDATE_REPO_NAME = 'api_update_me'
class SAME_AS_UPDATES(object): """ Constant used for tests below """
class SAME_AS_UPDATES(object):
""" Constant used for tests below """
@pytest.mark.usefixtures("testuser_api", "app")
class TestApiUpdateRepo(object):
@pytest.mark.parametrize("updates, expected", [
({'owner': TEST_USER_REGULAR_LOGIN}, SAME_AS_UPDATES),
({'description': 'new description'}, SAME_AS_UPDATES),
({'clone_uri': 'http://foo.com/repo'}, SAME_AS_UPDATES),
({'clone_uri': None}, {'clone_uri': ''}),
({'clone_uri': ''}, {'clone_uri': ''}),
({'landing_rev': 'branch:master'}, {'landing_rev': ['branch','master']}),
({'enable_statistics': True}, SAME_AS_UPDATES),
({'enable_locking': True}, SAME_AS_UPDATES),
({'enable_downloads': True}, SAME_AS_UPDATES),
({'name': 'new_repo_name'}, {
({'owner': TEST_USER_REGULAR_LOGIN},
SAME_AS_UPDATES),
({'description': 'new description'},
SAME_AS_UPDATES),
({'clone_uri': 'http://foo.com/repo'},
SAME_AS_UPDATES),
({'clone_uri': None},
{'clone_uri': ''}),
({'clone_uri': ''},
{'clone_uri': ''}),
({'landing_rev': 'rev:tip'},
{'landing_rev': ['rev', 'tip']}),
({'enable_statistics': True},
SAME_AS_UPDATES),
({'enable_locking': True},
SAME_AS_UPDATES),
({'enable_downloads': True},
SAME_AS_UPDATES),
({'repo_name': 'new_repo_name'},
{
'repo_name': 'new_repo_name',
'url': 'http://test.example.com:80/new_repo_name',
}),
({'group': 'test_group_for_update'}, {
'repo_name': 'test_group_for_update/%s' % UPDATE_REPO_NAME,
'url': 'http://test.example.com:80/test_group_for_update/%s' % UPDATE_REPO_NAME
}),
'url': 'http://test.example.com:80/new_repo_name'
}),
({'repo_name': 'test_group_for_update/{}'.format(UPDATE_REPO_NAME),
'_group': 'test_group_for_update'},
{
'repo_name': 'test_group_for_update/{}'.format(UPDATE_REPO_NAME),
'url': 'http://test.example.com:80/test_group_for_update/{}'.format(UPDATE_REPO_NAME)
}),
])
def test_api_update_repo(self, updates, expected, backend):
repo_name = UPDATE_REPO_NAME
repo = fixture.create_repo(repo_name, repo_type=backend.alias)
if updates.get('group'):
fixture.create_repo_group(updates['group'])
if updates.get('_group'):
fixture.create_repo_group(updates['_group'])
expected_api_data = repo.get_api_data(include_secrets=True)
if expected is SAME_AS_UPDATES:
@ -68,15 +93,12 @@ class TestApiUpdateRepo(object):
else:
expected_api_data.update(expected)
id_, params = build_data(
self.apikey, 'update_repo', repoid=repo_name, **updates)
response = api_call(self.app, params)
if updates.get('name'):
repo_name = updates['name']
if updates.get('group'):
repo_name = '/'.join([updates['group'], repo_name])
if updates.get('repo_name'):
repo_name = updates['repo_name']
try:
expected = {
@ -86,8 +108,8 @@ class TestApiUpdateRepo(object):
assert_ok(id_, expected, given=response.body)
finally:
fixture.destroy_repo(repo_name)
if updates.get('group'):
fixture.destroy_repo_group(updates['group'])
if updates.get('_group'):
fixture.destroy_repo_group(updates['_group'])
def test_api_update_repo_fork_of_field(self, backend):
master_repo = backend.create_repo()
@ -118,19 +140,23 @@ class TestApiUpdateRepo(object):
id_, params = build_data(
self.apikey, 'update_repo', repoid=repo.repo_name, **updates)
response = api_call(self.app, params)
expected = 'repository `{}` does not exist'.format(master_repo_name)
expected = {
'repo_fork_of': 'Fork with id `{}` does not exists'.format(
master_repo_name)}
assert_error(id_, expected, given=response.body)
def test_api_update_repo_with_repo_group_not_existing(self):
repo_name = 'admin_owned'
fake_repo_group = 'test_group_for_update'
fixture.create_repo(repo_name)
updates = {'group': 'test_group_for_update'}
updates = {'repo_name': '{}/{}'.format(fake_repo_group, repo_name)}
id_, params = build_data(
self.apikey, 'update_repo', repoid=repo_name, **updates)
response = api_call(self.app, params)
try:
expected = 'repository group `%s` does not exist' % (
updates['group'],)
expected = {
'repo_group': 'Repository group `{}` does not exist'.format(fake_repo_group)
}
assert_error(id_, expected, given=response.body)
finally:
fixture.destroy_repo(repo_name)

486
rhodecode/api/views/repo_api.py
View file

@ -21,29 +21,26 @@
import logging
import time
import colander
from rhodecode import BACKENDS
from rhodecode.api import jsonrpc_method, JSONRPCError, JSONRPCForbidden, json
import rhodecode
from rhodecode.api import (
jsonrpc_method, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
from rhodecode.api.utils import (
has_superadmin_permission, Optional, OAttr, get_repo_or_error,
get_user_group_or_error, get_user_or_error, has_repo_permissions,
get_perm_or_error, store_update, get_repo_group_or_error, parse_args,
get_origin, build_commit_data)
from rhodecode.lib.auth import (
HasPermissionAnyApi, HasRepoGroupPermissionAnyApi,
HasUserGroupPermissionAnyApi)
get_user_group_or_error, get_user_or_error, validate_repo_permissions,
get_perm_or_error, parse_args, get_origin, build_commit_data,
validate_set_owner_permissions)
from rhodecode.lib.auth import HasPermissionAnyApi, HasUserGroupPermissionAnyApi
from rhodecode.lib.exceptions import StatusChangeOnClosedPullRequestError
from rhodecode.lib.utils import map_groups
from rhodecode.lib.utils2 import str2bool, time_to_datetime
from rhodecode.lib.ext_json import json
from rhodecode.model.changeset_status import ChangesetStatusModel
from rhodecode.model.comment import ChangesetCommentsModel
from rhodecode.model.db import (
Session, ChangesetStatus, RepositoryField, Repository)
from rhodecode.model.repo import RepoModel
from rhodecode.model.repo_group import RepoGroupModel
from rhodecode.model.scm import ScmModel, RepoList
from rhodecode.model.settings import SettingsModel, VcsSettingsModel
from rhodecode.model import validation_schema
from rhodecode.model.validation_schema.schemas import repo_schema
log = logging.getLogger(__name__)
@ -177,6 +174,7 @@ def get_repo(request, apiuser, repoid, cache=Optional(True)):
repo = get_repo_or_error(repoid)
cache = Optional.extract(cache)
include_secrets = False
if has_superadmin_permission(apiuser):
include_secrets = True
@ -184,7 +182,7 @@ def get_repo(request, apiuser, repoid, cache=Optional(True)):
# check if we have at least read permission for this repo !
_perms = (
'repository.admin', 'repository.write', 'repository.read',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
permissions = []
for _user in repo.permissions():
@ -292,7 +290,7 @@ def get_repo_changeset(request, apiuser, repoid, revision,
if not has_superadmin_permission(apiuser):
_perms = (
'repository.admin', 'repository.write', 'repository.read',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
changes_details = Optional.extract(details)
_changes_details_types = ['basic', 'extended', 'full']
@ -355,7 +353,7 @@ def get_repo_changesets(request, apiuser, repoid, start_rev, limit,
if not has_superadmin_permission(apiuser):
_perms = (
'repository.admin', 'repository.write', 'repository.read',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
changes_details = Optional.extract(details)
_changes_details_types = ['basic', 'extended', 'full']
@ -450,7 +448,7 @@ def get_repo_nodes(request, apiuser, repoid, revision, root_path,
if not has_superadmin_permission(apiuser):
_perms = (
'repository.admin', 'repository.write', 'repository.read',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
ret_type = Optional.extract(ret_type)
details = Optional.extract(details)
@ -523,7 +521,7 @@ def get_repo_refs(request, apiuser, repoid):
repo = get_repo_or_error(repoid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin', 'repository.write', 'repository.read',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
try:
# check if repo is not empty by any chance, skip quicker if it is.
@ -538,26 +536,30 @@ def get_repo_refs(request, apiuser, repoid):
@jsonrpc_method()
def create_repo(request, apiuser, repo_name, repo_type,
owner=Optional(OAttr('apiuser')), description=Optional(''),
private=Optional(False), clone_uri=Optional(None),
landing_rev=Optional('rev:tip'),
enable_statistics=Optional(False),
enable_locking=Optional(False),
enable_downloads=Optional(False),
copy_permissions=Optional(False)):
def create_repo(
request, apiuser, repo_name, repo_type,
owner=Optional(OAttr('apiuser')),
description=Optional(''),
private=Optional(False),
clone_uri=Optional(None),
landing_rev=Optional('rev:tip'),
enable_statistics=Optional(False),
enable_locking=Optional(False),
enable_downloads=Optional(False),
copy_permissions=Optional(False)):
"""
Creates a repository.
* If the repository name contains "/", all the required repository
groups will be created.
* If the repository name contains "/", repository will be created inside
a repository group or nested repository groups
For example "foo/bar/baz" will create |repo| groups "foo" and "bar"
(with "foo" as parent). It will also create the "baz" repository
with "bar" as |repo| group.
For example "foo/bar/repo1" will create |repo| called "repo1" inside
group "foo/bar". You have to have permissions to access and write to
the last repository group ("bar" in this example)
This command can only be run using an |authtoken| with at least
write permissions to the |repo|.
permissions to create repositories, or write permissions to
parent repository groups.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
@ -569,9 +571,9 @@ def create_repo(request, apiuser, repo_name, repo_type,
:type owner: Optional(str)
:param description: Set the repository description.
:type description: Optional(str)
:param private:
:param private: set repository as private
:type private: bool
:param clone_uri:
:param clone_uri: set clone_uri
:type clone_uri: str
:param landing_rev: <rev_type>:<rev>
:type landing_rev: str
@ -610,49 +612,13 @@ def create_repo(request, apiuser, repo_name, repo_type,
}
"""
schema = repo_schema.RepoSchema()
try:
data = schema.deserialize({
'repo_name': repo_name
})
except colander.Invalid as e:
raise JSONRPCError("Validation failed: %s" % (e.asdict(),))
repo_name = data['repo_name']
(repo_name_cleaned,
parent_group_name) = RepoGroupModel()._get_group_name_and_parent(
repo_name)
owner = validate_set_owner_permissions(apiuser, owner)
if not HasPermissionAnyApi(
'hg.admin', 'hg.create.repository')(user=apiuser):
# check if we have admin permission for this repo group if given !
if parent_group_name:
repogroupid = parent_group_name
repo_group = get_repo_group_or_error(parent_group_name)
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
raise JSONRPCError(
'repository group `%s` does not exist' % (
repogroupid,))
else:
raise JSONRPCForbidden()
if not has_superadmin_permission(apiuser):
if not isinstance(owner, Optional):
# forbid setting owner for non-admins
raise JSONRPCError(
'Only RhodeCode admin can specify `owner` param')
if isinstance(owner, Optional):
owner = apiuser.user_id
owner = get_user_or_error(owner)
if RepoModel().get_by_repo_name(repo_name):
raise JSONRPCError("repo `%s` already exist" % repo_name)
description = Optional.extract(description)
copy_permissions = Optional.extract(copy_permissions)
clone_uri = Optional.extract(clone_uri)
landing_commit_ref = Optional.extract(landing_rev)
defs = SettingsModel().get_default_repo_settings(strip_prefix=True)
if isinstance(private, Optional):
@ -666,32 +632,44 @@ def create_repo(request, apiuser, repo_name, repo_type,
if isinstance(enable_downloads, Optional):
enable_downloads = defs.get('repo_enable_downloads')
clone_uri = Optional.extract(clone_uri)
description = Optional.extract(description)
landing_rev = Optional.extract(landing_rev)
copy_permissions = Optional.extract(copy_permissions)
schema = repo_schema.RepoSchema().bind(
repo_type_options=rhodecode.BACKENDS.keys(),
# user caller
user=apiuser)
try:
schema_data = schema.deserialize(dict(
repo_name=repo_name,
repo_type=repo_type,
repo_owner=owner.username,
repo_description=description,
repo_landing_commit_ref=landing_commit_ref,
repo_clone_uri=clone_uri,
repo_private=private,
repo_copy_permissions=copy_permissions,
repo_enable_statistics=enable_statistics,
repo_enable_downloads=enable_downloads,
repo_enable_locking=enable_locking))
except validation_schema.Invalid as err:
raise JSONRPCValidationError(colander_exc=err)
try:
# create structure of groups and return the last group
repo_group = map_groups(repo_name)
data = {
'repo_name': repo_name_cleaned,
'repo_name_full': repo_name,
'repo_type': repo_type,
'repo_description': description,
'owner': owner,
'repo_private': private,
'clone_uri': clone_uri,
'repo_group': repo_group.group_id if repo_group else None,
'repo_landing_rev': landing_rev,
'enable_statistics': enable_statistics,
'enable_locking': enable_locking,
'enable_downloads': enable_downloads,
'repo_copy_permissions': copy_permissions,
'repo_name': schema_data['repo_group']['repo_name_without_group'],
'repo_name_full': schema_data['repo_name'],
'repo_group': schema_data['repo_group']['repo_group_id'],
'repo_type': schema_data['repo_type'],
'repo_description': schema_data['repo_description'],
'repo_private': schema_data['repo_private'],
'clone_uri': schema_data['repo_clone_uri'],
'repo_landing_rev': schema_data['repo_landing_commit_ref'],
'enable_statistics': schema_data['repo_enable_statistics'],
'enable_locking': schema_data['repo_enable_locking'],
'enable_downloads': schema_data['repo_enable_downloads'],
'repo_copy_permissions': schema_data['repo_copy_permissions'],
}
if repo_type not in BACKENDS.keys():
raise Exception("Invalid backend type %s" % repo_type)
task = RepoModel().create(form_data=data, cur_user=owner)
from celery.result import BaseAsyncResult
task_id = None
@ -699,17 +677,17 @@ def create_repo(request, apiuser, repo_name, repo_type,
task_id = task.task_id
# no commit, it's done in RepoModel, or async via celery
return {
'msg': "Created new repository `%s`" % (repo_name,),
'msg': "Created new repository `%s`" % (schema_data['repo_name'],),
'success': True, # cannot return the repo data here since fork
# cann be done async
# can be done async
'task': task_id
}
except Exception:
log.exception(
u"Exception while trying to create the repository %s",
repo_name)
schema_data['repo_name'])
raise JSONRPCError(
'failed to create repository `%s`' % (repo_name,))
'failed to create repository `%s`' % (schema_data['repo_name'],))
@jsonrpc_method()
@ -735,7 +713,7 @@ def add_field_to_repo(request, apiuser, repoid, key, label=Optional(''),
repo = get_repo_or_error(repoid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
label = Optional.extract(label) or key
description = Optional.extract(description)
@ -778,7 +756,7 @@ def remove_field_from_repo(request, apiuser, repoid, key):
repo = get_repo_or_error(repoid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
field = RepositoryField.get_by_key_name(key, repo)
if not field:
@ -800,33 +778,38 @@ def remove_field_from_repo(request, apiuser, repoid, key):
@jsonrpc_method()
def update_repo(request, apiuser, repoid, name=Optional(None),
owner=Optional(OAttr('apiuser')),
group=Optional(None),
fork_of=Optional(None),
description=Optional(''), private=Optional(False),
clone_uri=Optional(None), landing_rev=Optional('rev:tip'),
enable_statistics=Optional(False),
enable_locking=Optional(False),
enable_downloads=Optional(False),
fields=Optional('')):
def update_repo(
request, apiuser, repoid, repo_name=Optional(None),
owner=Optional(OAttr('apiuser')), description=Optional(''),
private=Optional(False), clone_uri=Optional(None),
landing_rev=Optional('rev:tip'), fork_of=Optional(None),
enable_statistics=Optional(False),
enable_locking=Optional(False),
enable_downloads=Optional(False), fields=Optional('')):
"""
Updates a repository with the given information.
This command can only be run using an |authtoken| with at least
write permissions to the |repo|.
admin permissions to the |repo|.
* If the repository name contains "/", repository will be updated
accordingly with a repository group or nested repository groups
For example repoid=repo-test name="foo/bar/repo-test" will update |repo|
called "repo-test" and place it inside group "foo/bar".
You have to have permissions to access and write to the last repository
group ("bar" in this example)
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param repoid: repository name or repository ID.
:type repoid: str or int
:param name: Update the |repo| name.
:type name: str
:param repo_name: Update the |repo| name, including the
repository group it's in.
:type repo_name: str
:param owner: Set the |repo| owner.
:type owner: str
:param group: Set the |repo| group the |repo| belongs to.
:type group: str
:param fork_of: Set the master |repo| name.
:param fork_of: Set the |repo| as fork of another |repo|.
:type fork_of: str
:param description: Update the |repo| description.
:type description: str
@ -834,69 +817,115 @@ def update_repo(request, apiuser, repoid, name=Optional(None),
:type private: bool
:param clone_uri: Update the |repo| clone URI.
:type clone_uri: str
:param landing_rev: Set the |repo| landing revision. Default is
``tip``.
:param landing_rev: Set the |repo| landing revision. Default is ``rev:tip``.
:type landing_rev: str
:param enable_statistics: Enable statistics on the |repo|,
(True | False).
:param enable_statistics: Enable statistics on the |repo|, (True | False).
:type enable_statistics: bool
:param enable_locking: Enable |repo| locking.
:type enable_locking: bool
:param enable_downloads: Enable downloads from the |repo|,
(True | False).
:param enable_downloads: Enable downloads from the |repo|, (True | False).
:type enable_downloads: bool
:param fields: Add extra fields to the |repo|. Use the following
example format: ``field_key=field_val,field_key2=fieldval2``.
Escape ', ' with \,
:type fields: str
"""
repo = get_repo_or_error(repoid)
include_secrets = False
if has_superadmin_permission(apiuser):
include_secrets = True
if not has_superadmin_permission(apiuser):
validate_repo_permissions(apiuser, repoid, repo, ('repository.admin',))
else:
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
include_secrets = True
updates = {
# update function requires this.
'repo_name': repo.just_name
}
repo_group = group
if not isinstance(repo_group, Optional):
repo_group = get_repo_group_or_error(repo_group)
repo_group = repo_group.group_id
updates = dict(
repo_name=repo_name
if not isinstance(repo_name, Optional) else repo.repo_name,
repo_fork_of = fork_of
if not isinstance(repo_fork_of, Optional):
repo_fork_of = get_repo_or_error(repo_fork_of)
repo_fork_of = repo_fork_of.repo_id
fork_id=fork_of
if not isinstance(fork_of, Optional) else repo.fork.repo_name if repo.fork else None,
user=owner
if not isinstance(owner, Optional) else repo.user.username,
repo_description=description
if not isinstance(description, Optional) else repo.description,
repo_private=private
if not isinstance(private, Optional) else repo.private,
clone_uri=clone_uri
if not isinstance(clone_uri, Optional) else repo.clone_uri,
repo_landing_rev=landing_rev
if not isinstance(landing_rev, Optional) else repo._landing_revision,
repo_enable_statistics=enable_statistics
if not isinstance(enable_statistics, Optional) else repo.enable_statistics,
repo_enable_locking=enable_locking
if not isinstance(enable_locking, Optional) else repo.enable_locking,
repo_enable_downloads=enable_downloads
if not isinstance(enable_downloads, Optional) else repo.enable_downloads)
ref_choices, _labels = ScmModel().get_repo_landing_revs(repo=repo)
schema = repo_schema.RepoSchema().bind(
repo_type_options=rhodecode.BACKENDS.keys(),
repo_ref_options=ref_choices,
# user caller
user=apiuser,
old_values=repo.get_api_data())
try:
schema_data = schema.deserialize(dict(
# we save old value, users cannot change type
repo_type=repo.repo_type,
repo_name=updates['repo_name'],
repo_owner=updates['user'],
repo_description=updates['repo_description'],
repo_clone_uri=updates['clone_uri'],
repo_fork_of=updates['fork_id'],
repo_private=updates['repo_private'],
repo_landing_commit_ref=updates['repo_landing_rev'],
repo_enable_statistics=updates['repo_enable_statistics'],
repo_enable_downloads=updates['repo_enable_downloads'],
repo_enable_locking=updates['repo_enable_locking']))
except validation_schema.Invalid as err:
raise JSONRPCValidationError(colander_exc=err)
# save validated data back into the updates dict
validated_updates = dict(
repo_name=schema_data['repo_group']['repo_name_without_group'],
repo_group=schema_data['repo_group']['repo_group_id'],
user=schema_data['repo_owner'],
repo_description=schema_data['repo_description'],
repo_private=schema_data['repo_private'],
clone_uri=schema_data['repo_clone_uri'],
repo_landing_rev=schema_data['repo_landing_commit_ref'],
repo_enable_statistics=schema_data['repo_enable_statistics'],
repo_enable_locking=schema_data['repo_enable_locking'],
repo_enable_downloads=schema_data['repo_enable_downloads'],
)
if schema_data['repo_fork_of']:
fork_repo = get_repo_or_error(schema_data['repo_fork_of'])
validated_updates['fork_id'] = fork_repo.repo_id
# extra fields
fields = parse_args(Optional.extract(fields), key_prefix='ex_')
if fields:
validated_updates.update(fields)
try:
store_update(updates, name, 'repo_name')
store_update(updates, repo_group, 'repo_group')
store_update(updates, repo_fork_of, 'fork_id')
store_update(updates, owner, 'user')
store_update(updates, description, 'repo_description')
store_update(updates, private, 'repo_private')
store_update(updates, clone_uri, 'clone_uri')
store_update(updates, landing_rev, 'repo_landing_rev')
store_update(updates, enable_statistics, 'repo_enable_statistics')
store_update(updates, enable_locking, 'repo_enable_locking')
store_update(updates, enable_downloads, 'repo_enable_downloads')
# extra fields
fields = parse_args(Optional.extract(fields), key_prefix='ex_')
if fields:
updates.update(fields)
RepoModel().update(repo, **updates)
RepoModel().update(repo, **validated_updates)
Session().commit()
return {
'msg': 'updated repo ID:%s %s' % (
repo.repo_id, repo.repo_name),
'repository': repo.get_api_data(
include_secrets=include_secrets)
'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
'repository': repo.get_api_data(include_secrets=include_secrets)
}
except Exception:
log.exception(
@ -908,26 +937,33 @@ def update_repo(request, apiuser, repoid, name=Optional(None),
@jsonrpc_method()
def fork_repo(request, apiuser, repoid, fork_name,
owner=Optional(OAttr('apiuser')),
description=Optional(''), copy_permissions=Optional(False),
private=Optional(False), landing_rev=Optional('rev:tip')):
description=Optional(''),
private=Optional(False),
clone_uri=Optional(None),
landing_rev=Optional('rev:tip'),
copy_permissions=Optional(False)):
"""
Creates a fork of the specified |repo|.
* If using |RCE| with Celery this will immediately return a success
message, even though the fork will be created asynchronously.
* If the fork_name contains "/", fork will be created inside
a repository group or nested repository groups
This command can only be run using an |authtoken| with fork
permissions on the |repo|.
For example "foo/bar/fork-repo" will create fork called "fork-repo"
inside group "foo/bar". You have to have permissions to access and
write to the last repository group ("bar" in this example)
This command can only be run using an |authtoken| with minimum
read permissions of the forked repo, create fork permissions for an user.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param repoid: Set repository name or repository ID.
:type repoid: str or int
:param fork_name: Set the fork name.
:param fork_name: Set the fork name, including it's repository group membership.
:type fork_name: str
:param owner: Set the fork owner.
:type owner: str
:param description: Set the fork descripton.
:param description: Set the fork description.
:type description: str
:param copy_permissions: Copy permissions from parent |repo|. The
default is False.
@ -965,71 +1001,63 @@ def fork_repo(request, apiuser, repoid, fork_name,
error: null
"""
if not has_superadmin_permission(apiuser):
if not HasPermissionAnyApi('hg.fork.repository')(user=apiuser):
raise JSONRPCForbidden()
repo = get_repo_or_error(repoid)
repo_name = repo.repo_name
(fork_name_cleaned,
parent_group_name) = RepoGroupModel()._get_group_name_and_parent(
fork_name)
if not has_superadmin_permission(apiuser):
# check if we have at least read permission for
# this repo that we fork !
_perms = (
'repository.admin', 'repository.write', 'repository.read')
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
if not isinstance(owner, Optional):
# forbid setting owner for non super admins
raise JSONRPCError(
'Only RhodeCode admin can specify `owner` param'
)
# check if we have a create.repo permission if not maybe the parent
# group permission
if not HasPermissionAnyApi('hg.create.repository')(user=apiuser):
if parent_group_name:
repogroupid = parent_group_name
repo_group = get_repo_group_or_error(parent_group_name)
# check if the regular user has at least fork permissions as well
if not HasPermissionAnyApi('hg.fork.repository')(user=apiuser):
raise JSONRPCForbidden()
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
raise JSONRPCError(
'repository group `%s` does not exist' % (
repogroupid,))
else:
raise JSONRPCForbidden()
# check if user can set owner parameter
owner = validate_set_owner_permissions(apiuser, owner)
_repo = RepoModel().get_by_repo_name(fork_name)
if _repo:
type_ = 'fork' if _repo.fork else 'repo'
raise JSONRPCError("%s `%s` already exist" % (type_, fork_name))
description = Optional.extract(description)
copy_permissions = Optional.extract(copy_permissions)
clone_uri = Optional.extract(clone_uri)
landing_commit_ref = Optional.extract(landing_rev)
private = Optional.extract(private)
if isinstance(owner, Optional):
owner = apiuser.user_id
owner = get_user_or_error(owner)
schema = repo_schema.RepoSchema().bind(
repo_type_options=rhodecode.BACKENDS.keys(),
# user caller
user=apiuser)
try:
# create structure of groups and return the last group
repo_group = map_groups(fork_name)
form_data = {
'repo_name': fork_name_cleaned,
'repo_name_full': fork_name,
'repo_group': repo_group.group_id if repo_group else None,
'repo_type': repo.repo_type,
'description': Optional.extract(description),
'private': Optional.extract(private),
'copy_permissions': Optional.extract(copy_permissions),
'landing_rev': Optional.extract(landing_rev),
schema_data = schema.deserialize(dict(
repo_name=fork_name,
repo_type=repo.repo_type,
repo_owner=owner.username,
repo_description=description,
repo_landing_commit_ref=landing_commit_ref,
repo_clone_uri=clone_uri,
repo_private=private,
repo_copy_permissions=copy_permissions))
except validation_schema.Invalid as err:
raise JSONRPCValidationError(colander_exc=err)
try:
data = {
'fork_parent_id': repo.repo_id,
'repo_name': schema_data['repo_group']['repo_name_without_group'],
'repo_name_full': schema_data['repo_name'],
'repo_group': schema_data['repo_group']['repo_group_id'],
'repo_type': schema_data['repo_type'],
'description': schema_data['repo_description'],
'private': schema_data['repo_private'],
'copy_permissions': schema_data['repo_copy_permissions'],
'landing_rev': schema_data['repo_landing_commit_ref'],
}
task = RepoModel().create_fork(form_data, cur_user=owner)
task = RepoModel().create_fork(data, cur_user=owner)
# no commit, it's done in RepoModel, or async via celery
from celery.result import BaseAsyncResult
task_id = None
@ -1037,16 +1065,18 @@ def fork_repo(request, apiuser, repoid, fork_name,
task_id = task.task_id
return {
'msg': 'Created fork of `%s` as `%s`' % (
repo.repo_name, fork_name),
repo.repo_name, schema_data['repo_name']),
'success': True, # cannot return the repo data here since fork
# can be done async
'task': task_id
}
except Exception:
log.exception("Exception occurred while trying to fork a repo")
log.exception(
u"Exception while trying to create fork %s",
schema_data['repo_name'])
raise JSONRPCError(
'failed to fork repository `%s` as `%s`' % (
repo_name, fork_name))
repo_name, schema_data['repo_name']))
@jsonrpc_method()
@ -1082,7 +1112,7 @@ def delete_repo(request, apiuser, repoid, forks=Optional('')):
repo = get_repo_or_error(repoid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
try:
handle_forks = Optional.extract(forks)
@ -1157,7 +1187,7 @@ def invalidate_cache(request, apiuser, repoid, delete_keys=Optional(False)):
repo = get_repo_or_error(repoid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin', 'repository.write',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
delete = Optional.extract(delete_keys)
try:
@ -1236,7 +1266,7 @@ def lock(request, apiuser, repoid, locked=Optional(None),
if not has_superadmin_permission(apiuser):
# check if we have at least write permission for this repo !
_perms = ('repository.admin', 'repository.write',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
# make sure normal user does not pass someone else userid,
# he is not allowed to do that
@ -1347,7 +1377,7 @@ def comment_commit(
repo = get_repo_or_error(repoid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.read', 'repository.write', 'repository.admin')
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
if isinstance(userid, Optional):
userid = apiuser.user_id
@ -1438,7 +1468,7 @@ def grant_user_permission(request, apiuser, repoid, userid, perm):
perm = get_perm_or_error(perm)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
try:
@ -1492,7 +1522,7 @@ def revoke_user_permission(request, apiuser, repoid, userid):
user = get_user_or_error(userid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
try:
RepoModel().revoke_user_permission(repo=repo, user=user)
@ -1560,7 +1590,7 @@ def grant_user_group_permission(request, apiuser, repoid, usergroupid, perm):
perm = get_perm_or_error(perm)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
@ -1625,7 +1655,7 @@ def revoke_user_group_permission(request, apiuser, repoid, usergroupid):
repo = get_repo_or_error(repoid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
@ -1701,7 +1731,7 @@ def pull(request, apiuser, repoid):
repo = get_repo_or_error(repoid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
try:
ScmModel().pull_changes(repo.repo_name, apiuser.username)
@ -1764,7 +1794,7 @@ def strip(request, apiuser, repoid, revision, branch):
repo = get_repo_or_error(repoid)
if not has_superadmin_permission(apiuser):
_perms = ('repository.admin',)
has_repo_permissions(apiuser, repoid, repo, _perms)
validate_repo_permissions(apiuser, repoid, repo, _perms)
try:
ScmModel().strip(repo, revision, branch)

4
rhodecode/model/repo.py
View file

@ -377,11 +377,11 @@ class RepoModel(BaseModel):
log.debug('Updating repo %s with params:%s', cur_repo, kwargs)
update_keys = [
(1, 'repo_enable_downloads'),
(1, 'repo_description'),
(1, 'repo_enable_locking'),
(1, 'repo_landing_rev'),
(1, 'repo_private'),
(1, 'repo_enable_downloads'),
(1, 'repo_enable_locking'),
(1, 'repo_enable_statistics'),
(0, 'clone_uri'),
(0, 'fork_id')

12
rhodecode/model/scm.py
View file

@ -762,11 +762,15 @@ class ScmModel(BaseModel):
:param repo:
"""
hist_l = []
choices = []
repo = self._get_repo(repo)
hist_l.append(['rev:tip', _('latest tip')])
choices.append('rev:tip')
hist_l = [
['rev:tip', _('latest tip')]
]
choices = [
'rev:tip'
]
if not repo:
return choices, hist_l

1
rhodecode/model/validation_schema/preparers.py
View file

@ -21,7 +21,6 @@
import unicodedata
def strip_preparer(value):
"""
strips given values using .strip() function

298
rhodecode/model/validation_schema/schemas/repo_schema.py
View file

@ -20,8 +20,302 @@
import colander
from rhodecode.translation import _
from rhodecode.model.validation_schema import validators, preparers, types
DEFAULT_LANDING_REF = 'rev:tip'
class RepoSchema(colander.Schema):
repo_name = colander.SchemaNode(types.GroupNameType())
def get_group_and_repo(repo_name):
from rhodecode.model.repo_group import RepoGroupModel
return RepoGroupModel()._get_group_name_and_parent(
repo_name, get_object=True)
@colander.deferred
def deferred_repo_type_validator(node, kw):
options = kw.get('repo_type_options', [])
return colander.OneOf([x for x in options])
@colander.deferred
def deferred_repo_owner_validator(node, kw):
def repo_owner_validator(node, value):
from rhodecode.model.db import User
existing = User.get_by_username(value)
if not existing:
msg = _(u'Repo owner with id `{}` does not exists').format(value)
raise colander.Invalid(node, msg)
return repo_owner_validator
@colander.deferred
def deferred_landing_ref_validator(node, kw):
options = kw.get('repo_ref_options', [DEFAULT_LANDING_REF])
return colander.OneOf([x for x in options])
@colander.deferred
def deferred_fork_of_validator(node, kw):
old_values = kw.get('old_values') or {}
def fork_of_validator(node, value):
from rhodecode.model.db import Repository, RepoGroup
existing = Repository.get_by_repo_name(value)
if not existing:
msg = _(u'Fork with id `{}` does not exists').format(value)
raise colander.Invalid(node, msg)
elif old_values['repo_name'] == existing.repo_name:
msg = _(u'Cannot set fork of '
u'parameter of this repository to itself').format(value)
raise colander.Invalid(node, msg)
return fork_of_validator
@colander.deferred
def deferred_can_write_to_group_validator(node, kw):
request_user = kw.get('user')
old_values = kw.get('old_values') or {}
def can_write_to_group_validator(node, value):
"""
Checks if given repo path is writable by user. This includes checks if
user is allowed to create repositories under root path or under
repo group paths
"""
from rhodecode.lib.auth import (
HasPermissionAny, HasRepoGroupPermissionAny)
from rhodecode.model.repo_group import RepoGroupModel
messages = {
'invalid_repo_group':
_(u"Repository group `{}` does not exist"),
# permissions denied we expose as not existing, to prevent
# resource discovery
'permission_denied':
_(u"Repository group `{}` does not exist"),
'permission_denied_root':
_(u"You do not have the permission to store "
u"repositories in the root location.")
}
value = value['repo_group_name']
is_root_location = value is types.RootLocation
# NOT initialized validators, we must call them
can_create_repos_at_root = HasPermissionAny(
'hg.admin', 'hg.create.repository')
# if values is root location, we simply need to check if we can write
# to root location !
if is_root_location:
if can_create_repos_at_root(user=request_user):
# we can create repo group inside tool-level. No more checks
# are required
return
else:
# "fake" node name as repo_name, otherwise we oddly report
# the error as if it was coming form repo_group
# however repo_group is empty when using root location.
node.name = 'repo_name'
raise colander.Invalid(node, messages['permission_denied_root'])
# parent group not exists ? throw an error
repo_group = RepoGroupModel().get_by_group_name(value)
if value and not repo_group:
raise colander.Invalid(
node, messages['invalid_repo_group'].format(value))
gr_name = repo_group.group_name
# create repositories with write permission on group is set to true
create_on_write = HasPermissionAny(
'hg.create.write_on_repogroup.true')(user=request_user)
group_admin = HasRepoGroupPermissionAny('group.admin')(
gr_name, 'can write into group validator', user=request_user)
group_write = HasRepoGroupPermissionAny('group.write')(
gr_name, 'can write into group validator', user=request_user)
forbidden = not (group_admin or (group_write and create_on_write))
# TODO: handling of old values, and detecting no-change in path
# to skip permission checks in such cases. This only needs to be
# implemented if we use this schema in forms as well
# gid = (old_data['repo_group'].get('group_id')
# if (old_data and 'repo_group' in old_data) else None)
# value_changed = gid != safe_int(value)
# new = not old_data
# do check if we changed the value, there's a case that someone got
# revoked write permissions to a repository, he still created, we
# don't need to check permission if he didn't change the value of
# groups in form box
# if value_changed or new:
# # parent group need to be existing
# TODO: ENDS HERE
if repo_group and forbidden:
msg = messages['permission_denied'].format(value)
raise colander.Invalid(node, msg)
return can_write_to_group_validator
@colander.deferred
def deferred_unique_name_validator(node, kw):
request_user = kw.get('user')
old_values = kw.get('old_values') or {}
def unique_name_validator(node, value):
from rhodecode.model.db import Repository, RepoGroup
name_changed = value != old_values.get('repo_name')
existing = Repository.get_by_repo_name(value)
if name_changed and existing:
msg = _(u'Repository with name `{}` already exists').format(value)
raise colander.Invalid(node, msg)
existing_group = RepoGroup.get_by_group_name(value)
if name_changed and existing_group:
msg = _(u'Repository group with name `{}` already exists').format(
value)
raise colander.Invalid(node, msg)
return unique_name_validator
@colander.deferred
def deferred_repo_name_validator(node, kw):
return validators.valid_name_validator
class GroupType(colander.Mapping):
def _validate(self, node, value):
try:
return dict(repo_group_name=value)
except Exception as e:
raise colander.Invalid(
node, '"${val}" is not a mapping type: ${err}'.format(
val=value, err=e))
def deserialize(self, node, cstruct):
if cstruct is colander.null:
return cstruct
appstruct = super(GroupType, self).deserialize(node, cstruct)
validated_name = appstruct['repo_group_name']
# inject group based on once deserialized data
(repo_name_without_group,
parent_group_name,
parent_group) = get_group_and_repo(validated_name)
appstruct['repo_name_without_group'] = repo_name_without_group
appstruct['repo_group_name'] = parent_group_name or types.RootLocation
if parent_group:
appstruct['repo_group_id'] = parent_group.group_id
return appstruct
class GroupSchema(colander.SchemaNode):
schema_type = GroupType
validator = deferred_can_write_to_group_validator
missing = colander.null
class RepoGroup(GroupSchema):
repo_group_name = colander.SchemaNode(
types.GroupNameType())
repo_group_id = colander.SchemaNode(
colander.String(), missing=None)
repo_name_without_group = colander.SchemaNode(
colander.String(), missing=None)
class RepoGroupAccessSchema(colander.MappingSchema):
repo_group = RepoGroup()
class RepoNameUniqueSchema(colander.MappingSchema):
unique_repo_name = colander.SchemaNode(
colander.String(),
validator=deferred_unique_name_validator)
class RepoSchema(colander.MappingSchema):
repo_name = colander.SchemaNode(
types.RepoNameType(),
validator=deferred_repo_name_validator)
repo_type = colander.SchemaNode(
colander.String(),
validator=deferred_repo_type_validator)
repo_owner = colander.SchemaNode(
colander.String(),
validator=deferred_repo_owner_validator)
repo_description = colander.SchemaNode(
colander.String(), missing='')
repo_landing_commit_ref = colander.SchemaNode(
colander.String(),
validator=deferred_landing_ref_validator,
preparers=[preparers.strip_preparer],
missing=DEFAULT_LANDING_REF)
repo_clone_uri = colander.SchemaNode(
colander.String(),
validator=colander.All(colander.Length(min=1)),
preparers=[preparers.strip_preparer],
missing='')
repo_fork_of = colander.SchemaNode(
colander.String(),
validator=deferred_fork_of_validator,
missing=None)
repo_private = colander.SchemaNode(
types.StringBooleanType(),
missing=False)
repo_copy_permissions = colander.SchemaNode(
types.StringBooleanType(),
missing=False)
repo_enable_statistics = colander.SchemaNode(
types.StringBooleanType(),
missing=False)
repo_enable_downloads = colander.SchemaNode(
types.StringBooleanType(),
missing=False)
repo_enable_locking = colander.SchemaNode(
types.StringBooleanType(),
missing=False)
def deserialize(self, cstruct):
"""
Custom deserialize that allows to chain validation, and verify
permissions, and as last step uniqueness
"""
# first pass, to validate given data
appstruct = super(RepoSchema, self).deserialize(cstruct)
validated_name = appstruct['repo_name']
# second pass to validate permissions to repo_group
second = RepoGroupAccessSchema().bind(**self.bindings)
appstruct_second = second.deserialize({'repo_group': validated_name})
# save result
appstruct['repo_group'] = appstruct_second['repo_group']
# thirds to validate uniqueness
third = RepoNameUniqueSchema().bind(**self.bindings)
third.deserialize({'unique_repo_name': validated_name})
return appstruct

3
rhodecode/model/validation_schema/schemas/search_schema.py
View file

@ -33,8 +33,7 @@ class SearchParamsSchema(colander.MappingSchema):
search_sort = colander.SchemaNode(
colander.String(),
missing='newfirst',
validator=colander.OneOf(
['oldfirst', 'newfirst']))
validator=colander.OneOf(['oldfirst', 'newfirst']))
page_limit = colander.SchemaNode(
colander.Integer(),
missing=10,

65
rhodecode/model/validation_schema/types.py
View file

@ -18,22 +18,73 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import colander
import re
import colander
from rhodecode.model.validation_schema import preparers
from rhodecode.model.db import User, UserGroup
class _RootLocation(object):
pass
RootLocation = _RootLocation()
def _normalize(seperator, path):
if not path:
return ''
elif path is colander.null:
return colander.null
parts = path.split(seperator)
def bad_parts(value):
if not value:
return False
if re.match(r'^[.]+$', value):
return False
return True
def slugify(value):
value = preparers.slugify_preparer(value)
value = re.sub(r'[.]{2,}', '.', value)
return value
clean_parts = [slugify(item) for item in parts if item]
path = filter(bad_parts, clean_parts)
return seperator.join(path)
class RepoNameType(colander.String):
SEPARATOR = '/'
def deserialize(self, node, cstruct):
result = super(RepoNameType, self).deserialize(node, cstruct)
if cstruct is colander.null:
return colander.null
return self._normalize(result)
def _normalize(self, path):
return _normalize(self.SEPARATOR, path)
class GroupNameType(colander.String):
SEPARATOR = '/'
def deserialize(self, node, cstruct):
result = super(GroupNameType, self).deserialize(node, cstruct)
return self._replace_extra_slashes(result)
if cstruct is RootLocation:
return cstruct
def _replace_extra_slashes(self, path):
path = path.split(self.SEPARATOR)
path = [item for item in path if item]
return self.SEPARATOR.join(path)
result = super(GroupNameType, self).deserialize(node, cstruct)
if cstruct is colander.null:
return colander.null
return self._normalize(result)
def _normalize(self, path):
return _normalize(self.SEPARATOR, path)
class StringBooleanType(colander.String):

10
rhodecode/model/validation_schema/validators.py
View file

@ -36,3 +36,13 @@ def glob_validator(node, value):
except Exception:
msg = _(u'Invalid glob pattern')
raise colander.Invalid(node, msg)
def valid_name_validator(node, value):
from rhodecode.model.validation_schema import types
if value is types.RootLocation:
return
msg = _('Name must start with a letter or number. Got `{}`').format(value)
if not re.match(r'^[a-zA-z0-9]{1,}', value):
raise colander.Invalid(node, msg)

8
rhodecode/tests/models/schemas/test_integration_schema.py
View file

@ -21,8 +21,6 @@
import colander
import pytest
from rhodecode.model import validation_schema
from rhodecode.integrations import integration_type_registry
from rhodecode.integrations.types.base import IntegrationTypeBase
from rhodecode.model.validation_schema.schemas.integration_schema import (
@ -33,14 +31,12 @@ from rhodecode.model.validation_schema.schemas.integration_schema import (
@pytest.mark.usefixtures('app', 'autologin_user')
class TestIntegrationSchema(object):
def test_deserialize_integration_schema_perms(self, backend_random,
test_repo_group,
StubIntegrationType):
def test_deserialize_integration_schema_perms(
self, backend_random, test_repo_group, StubIntegrationType):
repo = backend_random.repo
repo_group = test_repo_group
empty_perms_dict = {
'global': [],
'repositories': {},

128
rhodecode/tests/models/schemas/test_repo_schema.py Normal file
View file

@ -0,0 +1,128 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 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 colander
import pytest
from rhodecode.model.validation_schema import types
from rhodecode.model.validation_schema.schemas import repo_schema
class TestRepoSchema(object):
#TODO:
# test nested groups
@pytest.mark.parametrize('given, expected', [
('my repo', 'my-repo'),
(' hello world mike ', 'hello-world-mike'),
('//group1/group2//', 'group1/group2'),
('//group1///group2//', 'group1/group2'),
('///group1/group2///group3', 'group1/group2/group3'),
('word g1/group2///group3', 'word-g1/group2/group3'),
('grou p1/gro;,,##up2//.../group3', 'grou-p1/group2/group3'),
('group,,,/,,,/1/2/3', 'group/1/2/3'),
('grou[]p1/gro;up2///gro up3', 'group1/group2/gro-up3'),
(u'grou[]p1/gro;up2///gro up3/ąć', u'group1/group2/gro-up3/ąć'),
])
def test_deserialize_repo_name(self, app, user_admin, given, expected):
schema = repo_schema.RepoSchema().bind()
assert expected == schema.get('repo_name').deserialize(given)
def test_deserialize(self, app, user_admin):
schema = repo_schema.RepoSchema().bind(
repo_type_options=['hg'],
user=user_admin
)
schema_data = schema.deserialize(dict(
repo_name='dupa',
repo_type='hg',
repo_owner=user_admin.username
))
assert schema_data['repo_name'] == 'dupa'
assert schema_data['repo_group'] == {
'repo_group_id': None,
'repo_group_name': types.RootLocation,
'repo_name_without_group': 'dupa'}
@pytest.mark.parametrize('given, err_key, expected_exc', [
('xxx/dupa','repo_group', 'Repository group `xxx` does not exist'),
('', 'repo_name', 'Name must start with a letter or number. Got ``'),
])
def test_deserialize_with_bad_group_name(
self, app, user_admin, given, err_key, expected_exc):
schema = repo_schema.RepoSchema().bind(
repo_type_options=['hg'],
user=user_admin
)
with pytest.raises(colander.Invalid) as excinfo:
schema.deserialize(dict(
repo_name=given,
repo_type='hg',
repo_owner=user_admin.username
))
assert excinfo.value.asdict()[err_key] == expected_exc
def test_deserialize_with_group_name(self, app, user_admin, test_repo_group):
schema = repo_schema.RepoSchema().bind(
repo_type_options=['hg'],
user=user_admin
)
full_name = test_repo_group.group_name + '/dupa'
schema_data = schema.deserialize(dict(
repo_name=full_name,
repo_type='hg',
repo_owner=user_admin.username
))
assert schema_data['repo_name'] == full_name
assert schema_data['repo_group'] == {
'repo_group_id': test_repo_group.group_id,
'repo_group_name': test_repo_group.group_name,
'repo_name_without_group': 'dupa'}
def test_deserialize_with_group_name_regular_user_no_perms(
self, app, user_regular, test_repo_group):
schema = repo_schema.RepoSchema().bind(
repo_type_options=['hg'],
user=user_regular
)
full_name = test_repo_group.group_name + '/dupa'
with pytest.raises(colander.Invalid) as excinfo:
schema.deserialize(dict(
repo_name=full_name,
repo_type='hg',
repo_owner=user_regular.username
))
expected = 'Repository group `{}` does not exist'.format(
test_repo_group.group_name)
assert excinfo.value.asdict()['repo_group'] == expected

78
rhodecode/tests/models/schemas/test_schema_types.py
View file

@ -21,26 +21,82 @@
import colander
import pytest
from rhodecode.model.validation_schema.types import GroupNameType
from rhodecode.model.validation_schema.types import (
GroupNameType, RepoNameType, StringBooleanType)
class TestGroupNameType(object):
@pytest.mark.parametrize('given, expected', [
('//group1/group2//', 'group1/group2'),
('//group1///group2//', 'group1/group2'),
('group1/group2///group3', 'group1/group2/group3')
('group1/group2///group3', 'group1/group2/group3'),
])
def test_replace_extra_slashes_cleans_up_extra_slashes(
self, given, expected):
type_ = GroupNameType()
result = type_._replace_extra_slashes(given)
def test_normalize_path(self, given, expected):
result = GroupNameType()._normalize(given)
assert result == expected
def test_deserialize_cleans_up_extra_slashes(self):
@pytest.mark.parametrize('given, expected', [
('//group1/group2//', 'group1/group2'),
('//group1///group2//', 'group1/group2'),
('group1/group2///group3', 'group1/group2/group3'),
('v1.2', 'v1.2'),
('/v1.2', 'v1.2'),
('.dirs', '.dirs'),
('..dirs', '.dirs'),
('./..dirs', '.dirs'),
('dir/;name;/;[];/sub', 'dir/name/sub'),
(',/,/,d,,,', 'd'),
('/;/#/,d,,,', 'd'),
('long../../..name', 'long./.name'),
('long../..name', 'long./.name'),
('../', ''),
('\'../"../', ''),
('c,/,/..//./,c,,,/.d/../.........c', 'c/c/.d/.c'),
('c,/,/..//./,c,,,', 'c/c'),
('d../..d', 'd./.d'),
('d../../d', 'd./d'),
('d\;\./\,\./d', 'd./d'),
('d\.\./\.\./d', 'd./d'),
('d\.\./\..\../d', 'd./d'),
])
def test_deserialize_clean_up_name(self, given, expected):
class TestSchema(colander.Schema):
field = colander.SchemaNode(GroupNameType())
field_group = colander.SchemaNode(GroupNameType())
field_repo = colander.SchemaNode(RepoNameType())
schema = TestSchema()
cleaned_data = schema.deserialize(
{'field': '//group1/group2///group3//'})
assert cleaned_data['field'] == 'group1/group2/group3'
cleaned_data = schema.deserialize({
'field_group': given,
'field_repo': given
})
assert cleaned_data['field_group'] == expected
assert cleaned_data['field_repo'] == expected
class TestStringBooleanType(object):
def _get_schema(self):
class Schema(colander.MappingSchema):
bools = colander.SchemaNode(StringBooleanType())
return Schema()
@pytest.mark.parametrize('given, expected', [
('1', True),
('yEs', True),
('true', True),
('0', False),
('NO', False),
('FALSE', False),
])
def test_convert_type(self, given, expected):
schema = self._get_schema()
result = schema.deserialize({'bools':given})
assert result['bools'] == expected
def test_try_convert_bad_type(self):
schema = self._get_schema()
with pytest.raises(colander.Invalid):
result = schema.deserialize({'bools': 'boom'})