# Copyright (C) 2016-2024 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 os import deform import colander from rhodecode.translation import _ from rhodecode.model.db import Repository, RepoGroup from rhodecode.model.validation_schema import validators, preparers def integration_scope_choices(permissions): """ Return list of (value, label) choices for integration scopes depending on the permissions """ result = [('', _('Pick a scope:'))] if 'hg.admin' in permissions['global']: result.extend([ ('global', _('Global (all repositories)')), ('root-repos', _('Top level repositories only')), ]) repo_choices = [ ('repo:%s' % repo_name, '/' + repo_name) for repo_name, repo_perm in list(permissions['repositories'].items()) if repo_perm == 'repository.admin' ] repogroup_choices = [ ('repogroup:%s' % repo_group_name, '/' + repo_group_name + '/ (child repos only)') for repo_group_name, repo_group_perm in list(permissions['repositories_groups'].items()) if repo_group_perm == 'group.admin' ] repogroup_recursive_choices = [ ('repogroup-recursive:%s' % repo_group_name, '/' + repo_group_name + '/ (recursive)') for repo_group_name, repo_group_perm in list(permissions['repositories_groups'].items()) if repo_group_perm == 'group.admin' ] result.extend( sorted(repogroup_recursive_choices + repogroup_choices + repo_choices, key=lambda choice_label: choice_label[0].split(':', 1)[1] ) ) return result @colander.deferred def deferred_integration_scopes_validator(node, kw): perms = kw.get('permissions') def _scope_validator(_node, scope): is_super_admin = 'hg.admin' in perms['global'] if scope.get('repo'): if (is_super_admin or perms['repositories'].get( scope['repo'].repo_name) == 'repository.admin'): return True msg = _('Only repo admins can create integrations') raise colander.Invalid(_node, msg) elif scope.get('repo_group'): if (is_super_admin or perms['repositories_groups'].get( scope['repo_group'].group_name) == 'group.admin'): return True msg = _('Only repogroup admins can create integrations') raise colander.Invalid(_node, msg) else: if is_super_admin: return True msg = _('Only superadmins can create global integrations') raise colander.Invalid(_node, msg) return _scope_validator @colander.deferred def deferred_integration_scopes_widget(node, kw): if kw.get('no_scope'): return deform.widget.TextInputWidget(readonly=True) choices = integration_scope_choices(kw.get('permissions')) widget = deform.widget.Select2Widget(values=choices) return widget class IntegrationScopeType(colander.SchemaType): def serialize(self, node, appstruct): if appstruct is colander.null: return colander.null if appstruct.get('repo'): return 'repo:%s' % appstruct['repo'].repo_name elif appstruct.get('repo_group'): if appstruct.get('child_repos_only'): return 'repogroup:%s' % appstruct['repo_group'].group_name else: return 'repogroup-recursive:%s' % ( appstruct['repo_group'].group_name) else: if appstruct.get('child_repos_only'): return 'root-repos' else: return 'global' def deserialize(self, node, cstruct): if cstruct is colander.null: return colander.null if cstruct.startswith('repo:'): repo = Repository.get_by_repo_name(cstruct.split(':')[1]) if repo: return { 'repo': repo, 'repo_group': None, 'child_repos_only': False, } elif cstruct.startswith('repogroup-recursive:'): repo_group = RepoGroup.get_by_group_name(cstruct.split(':')[1]) if repo_group: return { 'repo': None, 'repo_group': repo_group, 'child_repos_only': False } elif cstruct.startswith('repogroup:'): repo_group = RepoGroup.get_by_group_name(cstruct.split(':')[1]) if repo_group: return { 'repo': None, 'repo_group': repo_group, 'child_repos_only': True } elif cstruct == 'global': return { 'repo': None, 'repo_group': None, 'child_repos_only': False } elif cstruct == 'root-repos': return { 'repo': None, 'repo_group': None, 'child_repos_only': True } raise colander.Invalid(node, '%r is not a valid scope' % cstruct) class IntegrationOptionsSchemaBase(colander.MappingSchema): name = colander.SchemaNode( colander.String(), description=_('Short name for this integration.'), missing=colander.required, title=_('Integration name'), ) scope = colander.SchemaNode( IntegrationScopeType(), description=_( 'Scope of the integration. Recursive means the integration ' ' runs on all repos of that group and children recursively.'), title=_('Integration scope'), validator=deferred_integration_scopes_validator, widget=deferred_integration_scopes_widget, missing=colander.required, ) enabled = colander.SchemaNode( colander.Bool(), default=True, description=_('Enable or disable this integration.'), missing=False, title=_('Enabled'), ) def make_integration_schema(IntegrationType, settings=None): """ Return a colander schema for an integration type :param IntegrationType: the integration type class :param settings: existing integration settings dict (optional) """ settings = settings or {} settings_schema = IntegrationType(settings=settings).settings_schema() class IntegrationSchema(colander.Schema): options = IntegrationOptionsSchemaBase() schema = IntegrationSchema() schema['options'].title = _('General integration options') settings_schema.name = 'settings' settings_schema.title = _('{integration_type} settings').format( integration_type=IntegrationType.display_name) schema.add(settings_schema) return schema