Changeset - 40fea9b37a32
[Not reviewed]
default
0 2 0
Thomas De Schampheleire - 7 years ago 2018-05-20 22:29:40
thomas.de_schampheleire@nokia.com
admin: hooks: prevent editing of builtin hooks (issue #226)

Builtin hooks are supposed to be read-only, but it was still possible to
'add' a new hook with the same name as an existing built-in one, changing
its value.
2 files changed with 14 insertions and 0 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/admin/settings.py
Show inline comments
 
@@ -260,192 +260,194 @@ class SettingsController(BaseController)
 
                    errors=errors.error_dict or {},
 
                    prefix_error=False,
 
                    encoding="UTF-8",
 
                    force_defaults=False)
 

	
 
            try:
 
                settings = [
 
                    ('show_public_icon', 'show_public_icon', 'bool'),
 
                    ('show_private_icon', 'show_private_icon', 'bool'),
 
                    ('stylify_metalabels', 'stylify_metalabels', 'bool'),
 
                    ('repository_fields', 'repository_fields', 'bool'),
 
                    ('dashboard_items', 'dashboard_items', 'int'),
 
                    ('admin_grid_items', 'admin_grid_items', 'int'),
 
                    ('show_version', 'show_version', 'bool'),
 
                    ('use_gravatar', 'use_gravatar', 'bool'),
 
                    ('gravatar_url', 'gravatar_url', 'unicode'),
 
                    ('clone_uri_tmpl', 'clone_uri_tmpl', 'unicode'),
 
                ]
 
                for setting, form_key, type_ in settings:
 
                    Setting.create_or_update(setting, form_result[form_key], type_)
 

	
 
                Session().commit()
 
                set_app_settings(config)
 
                h.flash(_('Updated visualisation settings'),
 
                        category='success')
 

	
 
            except Exception:
 
                log.error(traceback.format_exc())
 
                h.flash(_('Error occurred during updating '
 
                          'visualisation settings'),
 
                        category='error')
 

	
 
            raise HTTPFound(location=url('admin_settings_visual'))
 

	
 
        defaults = Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        return htmlfill.render(
 
            render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_email(self):
 
        c.active = 'email'
 
        if request.POST:
 
            test_email = request.POST.get('test_email')
 
            test_email_subj = 'Kallithea test email'
 
            test_body = ('Kallithea Email test, '
 
                               'Kallithea version: %s' % c.kallithea_version)
 
            if not test_email:
 
                h.flash(_('Please enter email address'), category='error')
 
                raise HTTPFound(location=url('admin_settings_email'))
 

	
 
            test_email_txt_body = EmailNotificationModel() \
 
                .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
 
                                'txt', body=test_body)
 
            test_email_html_body = EmailNotificationModel() \
 
                .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
 
                                'html', body=test_body)
 

	
 
            recipients = [test_email] if test_email else None
 

	
 
            tasks.send_email(recipients, test_email_subj,
 
                             test_email_txt_body, test_email_html_body)
 

	
 
            h.flash(_('Send email task created'), category='success')
 
            raise HTTPFound(location=url('admin_settings_email'))
 

	
 
        defaults = Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        import kallithea
 
        c.ini = kallithea.CONFIG
 

	
 
        return htmlfill.render(
 
            render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_hooks(self):
 
        c.active = 'hooks'
 
        if request.POST:
 
            if c.visual.allow_custom_hooks_settings:
 
                ui_key = request.POST.get('new_hook_ui_key')
 
                ui_value = request.POST.get('new_hook_ui_value')
 

	
 
                hook_id = request.POST.get('hook_id')
 

	
 
                try:
 
                    ui_key = ui_key and ui_key.strip()
 
                    if ui_key in (x.ui_key for x in Ui.get_custom_hooks()):
 
                        h.flash(_('Hook already exists'), category='error')
 
                    elif ui_key in (x.ui_key for x in Ui.get_builtin_hooks()):
 
                        h.flash(_('Builtin hooks are read-only. Please use another hook name.'), category='error')
 
                    elif ui_value and ui_key:
 
                        Ui.create_or_update_hook(ui_key, ui_value)
 
                        h.flash(_('Added new hook'), category='success')
 
                    elif hook_id:
 
                        Ui.delete(hook_id)
 
                        Session().commit()
 

	
 
                    # check for edits
 
                    update = False
 
                    _d = request.POST.dict_of_lists()
 
                    for k, v in zip(_d.get('hook_ui_key', []),
 
                                    _d.get('hook_ui_value_new', [])):
 
                        Ui.create_or_update_hook(k, v)
 
                        update = True
 

	
 
                    if update:
 
                        h.flash(_('Updated hooks'), category='success')
 
                    Session().commit()
 
                except Exception:
 
                    log.error(traceback.format_exc())
 
                    h.flash(_('Error occurred during hook creation'),
 
                            category='error')
 

	
 
                raise HTTPFound(location=url('admin_settings_hooks'))
 

	
 
        defaults = Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        c.hooks = Ui.get_builtin_hooks()
 
        c.custom_hooks = Ui.get_custom_hooks()
 

	
 
        return htmlfill.render(
 
            render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_search(self):
 
        c.active = 'search'
 
        if request.POST:
 
            repo_location = self._get_hg_ui_settings()['paths_root_path']
 
            full_index = request.POST.get('full_index', False)
 
            tasks.whoosh_index(repo_location, full_index)
 
            h.flash(_('Whoosh reindex task scheduled'), category='success')
 
            raise HTTPFound(location=url('admin_settings_search'))
 

	
 
        defaults = Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        return htmlfill.render(
 
            render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_system(self):
 
        c.active = 'system'
 

	
 
        defaults = Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        import kallithea
 
        c.ini = kallithea.CONFIG
 
        c.update_url = defaults.get('update_url')
 
        server_info = Setting.get_server_info()
 
        for key, val in server_info.iteritems():
 
            setattr(c, key, val)
 

	
 
        return htmlfill.render(
 
            render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_system_update(self):
 
        import json
 
        import urllib2
 
        from kallithea.lib.verlib import NormalizedVersion
 
        from kallithea import __version__
 

	
 
        defaults = Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 
        _update_url = defaults.get('update_url', '')
 
        _update_url = "" # FIXME: disabled
 

	
 
        _err = lambda s: '<div class="alert alert-danger">%s</div>' % (s)
 
        try:
 
            import kallithea
 
            ver = kallithea.__version__
 
            log.debug('Checking for upgrade on `%s` server', _update_url)
 
            opener = urllib2.build_opener()
 
            opener.addheaders = [('User-agent', 'Kallithea-SCM/%s' % ver)]
 
            response = opener.open(_update_url)
kallithea/tests/functional/test_admin_settings.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
from kallithea.model.db import Setting, Ui
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestAdminSettingsController(TestController):
 

	
 
    def test_index_main(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings'))
 

	
 
    def test_index_mapping(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_mapping'))
 

	
 
    def test_index_global(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_global'))
 

	
 
    def test_index_visual(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_visual'))
 

	
 
    def test_index_email(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_email'))
 

	
 
    def test_index_hooks(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_hooks'))
 

	
 
    def test_create_custom_hook(self):
 
        self.log_user()
 
        response = self.app.post(url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='test_hooks_1',
 
                                            new_hook_ui_value='cd %s' % TESTS_TMP_PATH,
 
                                            _authentication_token=self.authentication_token()))
 

	
 
        self.checkSessionFlash(response, 'Added new hook')
 
        response = response.follow()
 
        response.mustcontain('test_hooks_1')
 
        response.mustcontain('cd %s' % TESTS_TMP_PATH)
 

	
 
    def test_edit_custom_hook(self):
 
        self.log_user()
 
        response = self.app.post(url('admin_settings_hooks'),
 
                                params=dict(hook_ui_key='test_hooks_1',
 
                                            hook_ui_value_new='new_value_of_hook_1',
 
                                            _authentication_token=self.authentication_token()))
 

	
 
        response = response.follow()
 
        response.mustcontain('test_hooks_1')
 
        response.mustcontain('new_value_of_hook_1')
 

	
 
    def test_add_existing_custom_hook(self):
 
        self.log_user()
 
        response = self.app.post(url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='test_hooks_1',
 
                                            new_hook_ui_value='attempted_new_value',
 
                                            _authentication_token=self.authentication_token()))
 

	
 
        self.checkSessionFlash(response, 'Hook already exists')
 
        response = response.follow()
 
        response.mustcontain('test_hooks_1')
 
        response.mustcontain('new_value_of_hook_1')
 

	
 
    def test_create_custom_hook_delete(self):
 
        self.log_user()
 
        response = self.app.post(url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='test_hooks_2',
 
                                            new_hook_ui_value='cd %s2' % TESTS_TMP_PATH,
 
                                            _authentication_token=self.authentication_token()))
 

	
 
        self.checkSessionFlash(response, 'Added new hook')
 
        response = response.follow()
 
        response.mustcontain('test_hooks_2')
 
        response.mustcontain('cd %s2' % TESTS_TMP_PATH)
 

	
 
        hook_id = Ui.get_by_key('hooks', 'test_hooks_2').ui_id
 
        ## delete
 
        self.app.post(url('admin_settings_hooks'),
 
                        params=dict(hook_id=hook_id, _authentication_token=self.authentication_token()))
 
        response = self.app.get(url('admin_settings_hooks'))
 
        response.mustcontain(no=['test_hooks_2'])
 
        response.mustcontain(no=['cd %s2' % TESTS_TMP_PATH])
 

	
 
    def test_add_existing_builtin_hook(self):
 
        self.log_user()
 
        response = self.app.post(url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='changegroup.update',
 
                                            new_hook_ui_value='attempted_new_value',
 
                                            _authentication_token=self.authentication_token()))
 

	
 
        self.checkSessionFlash(response, 'Builtin hooks are read-only')
 
        response = response.follow()
 
        response.mustcontain('changegroup.update')
 
        response.mustcontain('hg update &gt;&amp;2')
 

	
 
    def test_index_search(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_search'))
 

	
 
    def test_index_system(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_system'))
 

	
 
    def test_ga_code_active(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = 'ga-test-123456789'
 
        response = self.app.post(url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='',
 
                                 captcha_public_key='',
 
                                 _authentication_token=self.authentication_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 

	
 
        assert Setting.get_app_settings()['ga_code'] == new_ga_code
 

	
 
        response = response.follow()
 
        response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)
 

	
 
    def test_ga_code_inactive(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = ''
 
        response = self.app.post(url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='',
 
                                 captcha_public_key='',
 
                                 _authentication_token=self.authentication_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 
        assert Setting.get_app_settings()['ga_code'] == new_ga_code
 

	
 
        response = response.follow()
 
        response.mustcontain(no=["_gaq.push(['_setAccount', '%s']);" % new_ga_code])
 

	
 
    def test_captcha_activate(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = ''
 
        response = self.app.post(url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='1234567890',
 
                                 captcha_public_key='1234567890',
 
                                 _authentication_token=self.authentication_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 
        assert Setting.get_app_settings()['captcha_private_key'] == '1234567890'
 

	
 
        response = self.app.get(url('register'))
 
        response.mustcontain('captcha')
 

	
 
    def test_captcha_deactivate(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = ''
 
        response = self.app.post(url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='',
 
                                 captcha_public_key='1234567890',
 
                                 _authentication_token=self.authentication_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 
        assert Setting.get_app_settings()['captcha_private_key'] == ''
 

	
 
        response = self.app.get(url('register'))
 
        response.mustcontain(no=['captcha'])
 

	
 
    def test_title_change(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        new_title = old_title + '_changed'
 
        old_realm = 'Kallithea authentication'
 

	
 
        for new_title in ['Changed', 'Żółwik', old_title]:
0 comments (0 inline, 0 general)