# Copyright (C) 2010-2023 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 . # # 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 logging import collections import datetime import formencode import formencode.htmlfill import rhodecode from pyramid.httpexceptions import HTTPFound, HTTPNotFound from pyramid.renderers import render from pyramid.response import Response from rhodecode.apps._base import BaseAppView from rhodecode.apps._base.navigation import navigation_list from rhodecode.apps.svn_support import config_keys from rhodecode.lib import helpers as h from rhodecode.lib.auth import ( LoginRequired, HasPermissionAllDecorator, CSRFRequired) from rhodecode.lib.celerylib import tasks, run_task from rhodecode.lib.str_utils import safe_str from rhodecode.lib.utils import repo2db_mapper, get_rhodecode_repo_store_path from rhodecode.lib.utils2 import str2bool, AttributeDict from rhodecode.lib.index import searcher_from_config from rhodecode.model.db import RhodeCodeUi, Repository from rhodecode.model.forms import (ApplicationSettingsForm, ApplicationUiSettingsForm, ApplicationVisualisationForm, LabsSettingsForm, IssueTrackerPatternsForm) from rhodecode.model.permission import PermissionModel from rhodecode.model.repo_group import RepoGroupModel from rhodecode.model.scm import ScmModel from rhodecode.model.notification import EmailNotificationModel from rhodecode.model.meta import Session from rhodecode.model.settings import ( IssueTrackerSettingsModel, VcsSettingsModel, SettingNotFound, SettingsModel) log = logging.getLogger(__name__) class AdminSettingsView(BaseAppView): def load_default_context(self): c = self._get_local_tmpl_context() c.labs_active = str2bool( rhodecode.CONFIG.get('labs_settings_active', 'true')) c.navlist = navigation_list(self.request) return c @classmethod def _get_ui_settings(cls): ret = RhodeCodeUi.query().all() if not ret: raise Exception('Could not get application ui settings !') settings = { # legacy param that needs to be kept 'web_push_ssl': False } for each in ret: k = each.ui_key v = each.ui_value # skip some options if they are defined if k in ['push_ssl']: continue if k == '/': k = 'root_path' if k in ['publish', 'enabled']: v = str2bool(v) if k.find('.') != -1: k = k.replace('.', '_') if each.ui_section in ['hooks', 'extensions']: v = each.ui_active settings[each.ui_section + '_' + k] = v return settings @classmethod def _form_defaults(cls): defaults = SettingsModel().get_all_settings() defaults.update(cls._get_ui_settings()) defaults.update({ 'new_svn_branch': '', 'new_svn_tag': '', }) return defaults @LoginRequired() @HasPermissionAllDecorator('hg.admin') def settings_vcs(self): c = self.load_default_context() c.active = 'vcs' model = VcsSettingsModel() c.svn_branch_patterns = model.get_global_svn_branch_patterns() c.svn_tag_patterns = model.get_global_svn_tag_patterns() c.svn_generate_config = rhodecode.ConfigGet().get_bool(config_keys.generate_config) c.svn_config_path = rhodecode.ConfigGet().get_str(config_keys.config_file_path) defaults = self._form_defaults() model.create_largeobjects_dirs_if_needed(defaults['paths_root_path']) data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=defaults, encoding="UTF-8", force_defaults=False ) return Response(html) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_vcs_update(self): _ = self.request.translate c = self.load_default_context() c.active = 'vcs' model = VcsSettingsModel() c.svn_branch_patterns = model.get_global_svn_branch_patterns() c.svn_tag_patterns = model.get_global_svn_tag_patterns() c.svn_generate_config = rhodecode.ConfigGet().get_bool(config_keys.generate_config) c.svn_config_path = rhodecode.ConfigGet().get_str(config_keys.config_file_path) application_form = ApplicationUiSettingsForm(self.request.translate)() try: form_result = application_form.to_python(dict(self.request.POST)) except formencode.Invalid as errors: h.flash( _("Some form inputs contain invalid data."), category='error') data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=errors.value, errors=errors.unpack_errors() or {}, prefix_error=False, encoding="UTF-8", force_defaults=False ) return Response(html) try: model.update_global_hook_settings(form_result) model.create_or_update_global_svn_settings(form_result) model.create_or_update_global_hg_settings(form_result) model.create_or_update_global_git_settings(form_result) model.create_or_update_global_pr_settings(form_result) except Exception: log.exception("Exception while updating settings") h.flash(_('Error occurred during updating ' 'application settings'), category='error') else: Session().commit() h.flash(_('Updated VCS settings'), category='success') raise HTTPFound(h.route_path('admin_settings_vcs')) data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=self._form_defaults(), encoding="UTF-8", force_defaults=False ) return Response(html) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_vcs_delete_svn_pattern(self): delete_pattern_id = self.request.POST.get('delete_svn_pattern') model = VcsSettingsModel() try: model.delete_global_svn_pattern(delete_pattern_id) except SettingNotFound: log.exception( 'Failed to delete svn_pattern with id %s', delete_pattern_id) raise HTTPNotFound() Session().commit() return True @LoginRequired() @HasPermissionAllDecorator('hg.admin') def settings_mapping(self): c = self.load_default_context() c.active = 'mapping' c.storage_path = get_rhodecode_repo_store_path() data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=self._form_defaults(), encoding="UTF-8", force_defaults=False ) return Response(html) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_mapping_update(self): _ = self.request.translate c = self.load_default_context() c.active = 'mapping' rm_obsolete = self.request.POST.get('destroy', False) invalidate_cache = self.request.POST.get('invalidate', False) log.debug('rescanning repo location with destroy obsolete=%s', rm_obsolete) if invalidate_cache: log.debug('invalidating all repositories cache') for repo in Repository.get_all(): ScmModel().mark_for_invalidation(repo.repo_name, delete=True) filesystem_repos = ScmModel().repo_scan() added, removed = repo2db_mapper(filesystem_repos, rm_obsolete, force_hooks_rebuild=True) PermissionModel().trigger_permission_flush() def _repr(rm_repo): return ', '.join(map(safe_str, rm_repo)) or '-' h.flash(_('Repositories successfully ' 'rescanned added: %s ; removed: %s') % (_repr(added), _repr(removed)), category='success') raise HTTPFound(h.route_path('admin_settings_mapping')) @LoginRequired() @HasPermissionAllDecorator('hg.admin') def settings_global(self): c = self.load_default_context() c.active = 'global' c.personal_repo_group_default_pattern = RepoGroupModel()\ .get_personal_group_name_pattern() data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=self._form_defaults(), encoding="UTF-8", force_defaults=False ) return Response(html) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_global_update(self): _ = self.request.translate c = self.load_default_context() c.active = 'global' c.personal_repo_group_default_pattern = RepoGroupModel()\ .get_personal_group_name_pattern() application_form = ApplicationSettingsForm(self.request.translate)() try: form_result = application_form.to_python(dict(self.request.POST)) except formencode.Invalid as errors: h.flash( _("Some form inputs contain invalid data."), category='error') data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=errors.value, errors=errors.unpack_errors() or {}, prefix_error=False, encoding="UTF-8", force_defaults=False ) return Response(html) settings = [ ('title', 'rhodecode_title', 'unicode'), ('realm', 'rhodecode_realm', 'unicode'), ('pre_code', 'rhodecode_pre_code', 'unicode'), ('post_code', 'rhodecode_post_code', 'unicode'), ('captcha_public_key', 'rhodecode_captcha_public_key', 'unicode'), ('captcha_private_key', 'rhodecode_captcha_private_key', 'unicode'), ('create_personal_repo_group', 'rhodecode_create_personal_repo_group', 'bool'), ('personal_repo_group_pattern', 'rhodecode_personal_repo_group_pattern', 'unicode'), ] try: for setting, form_key, type_ in settings: sett = SettingsModel().create_or_update_setting( setting, form_result[form_key], type_) Session().add(sett) Session().commit() SettingsModel().invalidate_settings_cache() h.flash(_('Updated application settings'), category='success') except Exception: log.exception("Exception while updating application settings") h.flash( _('Error occurred during updating application settings'), category='error') raise HTTPFound(h.route_path('admin_settings_global')) @LoginRequired() @HasPermissionAllDecorator('hg.admin') def settings_visual(self): c = self.load_default_context() c.active = 'visual' data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=self._form_defaults(), encoding="UTF-8", force_defaults=False ) return Response(html) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_visual_update(self): _ = self.request.translate c = self.load_default_context() c.active = 'visual' application_form = ApplicationVisualisationForm(self.request.translate)() try: form_result = application_form.to_python(dict(self.request.POST)) except formencode.Invalid as errors: h.flash( _("Some form inputs contain invalid data."), category='error') data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=errors.value, errors=errors.unpack_errors() or {}, prefix_error=False, encoding="UTF-8", force_defaults=False ) return Response(html) try: settings = [ ('show_public_icon', 'rhodecode_show_public_icon', 'bool'), ('show_private_icon', 'rhodecode_show_private_icon', 'bool'), ('stylify_metatags', 'rhodecode_stylify_metatags', 'bool'), ('repository_fields', 'rhodecode_repository_fields', 'bool'), ('dashboard_items', 'rhodecode_dashboard_items', 'int'), ('admin_grid_items', 'rhodecode_admin_grid_items', 'int'), ('show_version', 'rhodecode_show_version', 'bool'), ('use_gravatar', 'rhodecode_use_gravatar', 'bool'), ('markup_renderer', 'rhodecode_markup_renderer', 'unicode'), ('gravatar_url', 'rhodecode_gravatar_url', 'unicode'), ('clone_uri_tmpl', 'rhodecode_clone_uri_tmpl', 'unicode'), ('clone_uri_id_tmpl', 'rhodecode_clone_uri_id_tmpl', 'unicode'), ('clone_uri_ssh_tmpl', 'rhodecode_clone_uri_ssh_tmpl', 'unicode'), ('support_url', 'rhodecode_support_url', 'unicode'), ('show_revision_number', 'rhodecode_show_revision_number', 'bool'), ('show_sha_length', 'rhodecode_show_sha_length', 'int'), ] for setting, form_key, type_ in settings: sett = SettingsModel().create_or_update_setting( setting, form_result[form_key], type_) Session().add(sett) Session().commit() SettingsModel().invalidate_settings_cache() h.flash(_('Updated visualisation settings'), category='success') except Exception: log.exception("Exception updating visualization settings") h.flash(_('Error occurred during updating ' 'visualisation settings'), category='error') raise HTTPFound(h.route_path('admin_settings_visual')) @LoginRequired() @HasPermissionAllDecorator('hg.admin') def settings_issuetracker(self): c = self.load_default_context() c.active = 'issuetracker' defaults = c.rc_config entry_key = 'rhodecode_issuetracker_pat_' c.issuetracker_entries = {} for k, v in defaults.items(): if k.startswith(entry_key): uid = k[len(entry_key):] c.issuetracker_entries[uid] = None for uid in c.issuetracker_entries: c.issuetracker_entries[uid] = AttributeDict({ 'pat': defaults.get('rhodecode_issuetracker_pat_' + uid), 'url': defaults.get('rhodecode_issuetracker_url_' + uid), 'pref': defaults.get('rhodecode_issuetracker_pref_' + uid), 'desc': defaults.get('rhodecode_issuetracker_desc_' + uid), }) return self._get_template_context(c) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_issuetracker_test(self): error_container = [] urlified_commit = h.urlify_commit_message( self.request.POST.get('test_text', ''), 'repo_group/test_repo1', error_container=error_container) if error_container: def converter(inp): return h.html_escape(inp) return 'ERRORS: ' + '\n'.join(map(converter, error_container)) return urlified_commit @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_issuetracker_update(self): _ = self.request.translate self.load_default_context() settings_model = IssueTrackerSettingsModel() try: form = IssueTrackerPatternsForm(self.request.translate)() data = form.to_python(self.request.POST) except formencode.Invalid as errors: log.exception('Failed to add new pattern') error = errors h.flash(_(f'Invalid issue tracker pattern: {error}'), category='error') raise HTTPFound(h.route_path('admin_settings_issuetracker')) if data: for uid in data.get('delete_patterns', []): settings_model.delete_entries(uid) for pattern in data.get('patterns', []): for setting, value, type_ in pattern: sett = settings_model.create_or_update_setting( setting, value, type_) Session().add(sett) Session().commit() SettingsModel().invalidate_settings_cache() h.flash(_('Updated issue tracker entries'), category='success') raise HTTPFound(h.route_path('admin_settings_issuetracker')) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_issuetracker_delete(self): _ = self.request.translate self.load_default_context() uid = self.request.POST.get('uid') try: IssueTrackerSettingsModel().delete_entries(uid) except Exception: log.exception('Failed to delete issue tracker setting %s', uid) raise HTTPNotFound() SettingsModel().invalidate_settings_cache() h.flash(_('Removed issue tracker entry.'), category='success') return {'deleted': uid} @LoginRequired() @HasPermissionAllDecorator('hg.admin') def settings_email(self): c = self.load_default_context() c.active = 'email' c.rhodecode_ini = rhodecode.CONFIG data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=self._form_defaults(), encoding="UTF-8", force_defaults=False ) return Response(html) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_email_update(self): _ = self.request.translate c = self.load_default_context() c.active = 'email' test_email = self.request.POST.get('test_email') if not test_email: h.flash(_('Please enter email address'), category='error') raise HTTPFound(h.route_path('admin_settings_email')) email_kwargs = { 'date': datetime.datetime.now(), 'user': self._rhodecode_db_user } (subject, email_body, email_body_plaintext) = EmailNotificationModel().render_email( EmailNotificationModel.TYPE_EMAIL_TEST, **email_kwargs) recipients = [test_email] if test_email else None run_task(tasks.send_email, recipients, subject, email_body_plaintext, email_body) h.flash(_('Send email task created'), category='success') raise HTTPFound(h.route_path('admin_settings_email')) @LoginRequired() @HasPermissionAllDecorator('hg.admin') def settings_hooks(self): c = self.load_default_context() c.active = 'hooks' model = SettingsModel() c.hooks = model.get_builtin_hooks() c.custom_hooks = model.get_custom_hooks() data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=self._form_defaults(), encoding="UTF-8", force_defaults=False ) return Response(html) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_hooks_update(self): _ = self.request.translate c = self.load_default_context() c.active = 'hooks' if c.visual.allow_custom_hooks_settings: ui_key = self.request.POST.get('new_hook_ui_key') ui_value = self.request.POST.get('new_hook_ui_value') hook_id = self.request.POST.get('hook_id') new_hook = False model = SettingsModel() try: if ui_value and ui_key: model.create_or_update_hook(ui_key, ui_value) h.flash(_('Added new hook'), category='success') new_hook = True elif hook_id: RhodeCodeUi.delete(hook_id) Session().commit() # check for edits update = False _d = self.request.POST.dict_of_lists() for k, v in zip(_d.get('hook_ui_key', []), _d.get('hook_ui_value_new', [])): model.create_or_update_hook(k, v) update = True if update and not new_hook: h.flash(_('Updated hooks'), category='success') Session().commit() except Exception: log.exception("Exception during hook creation") h.flash(_('Error occurred during hook creation'), category='error') raise HTTPFound(h.route_path('admin_settings_hooks')) @LoginRequired() @HasPermissionAllDecorator('hg.admin') def settings_search(self): c = self.load_default_context() c.active = 'search' c.searcher = searcher_from_config(self.request.registry.settings) c.statistics = c.searcher.statistics(self.request.translate) return self._get_template_context(c) @LoginRequired() @HasPermissionAllDecorator('hg.admin') def settings_labs(self): c = self.load_default_context() if not c.labs_active: raise HTTPFound(h.route_path('admin_settings')) c.active = 'labs' c.lab_settings = _LAB_SETTINGS data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=self._form_defaults(), encoding="UTF-8", force_defaults=False ) return Response(html) @LoginRequired() @HasPermissionAllDecorator('hg.admin') @CSRFRequired() def settings_labs_update(self): _ = self.request.translate c = self.load_default_context() c.active = 'labs' application_form = LabsSettingsForm(self.request.translate)() try: form_result = application_form.to_python(dict(self.request.POST)) except formencode.Invalid as errors: h.flash( _("Some form inputs contain invalid data."), category='error') data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=errors.value, errors=errors.unpack_errors() or {}, prefix_error=False, encoding="UTF-8", force_defaults=False ) return Response(html) try: session = Session() for setting in _LAB_SETTINGS: setting_name = setting.key[len('rhodecode_'):] sett = SettingsModel().create_or_update_setting( setting_name, form_result[setting.key], setting.type) session.add(sett) except Exception: log.exception('Exception while updating lab settings') h.flash(_('Error occurred during updating labs settings'), category='error') else: Session().commit() SettingsModel().invalidate_settings_cache() h.flash(_('Updated Labs settings'), category='success') raise HTTPFound(h.route_path('admin_settings_labs')) data = render('rhodecode:templates/admin/settings/settings.mako', self._get_template_context(c), self.request) html = formencode.htmlfill.render( data, defaults=self._form_defaults(), encoding="UTF-8", force_defaults=False ) return Response(html) # :param key: name of the setting including the 'rhodecode_' prefix # :param type: the RhodeCodeSetting type to use. # :param group: the i18ned group in which we should dispaly this setting # :param label: the i18ned label we should display for this setting # :param help: the i18ned help we should dispaly for this setting LabSetting = collections.namedtuple( 'LabSetting', ('key', 'type', 'group', 'label', 'help')) # This list has to be kept in sync with the form # rhodecode.model.forms.LabsSettingsForm. _LAB_SETTINGS = [ ]