auth: add scope and login restrictions to rhodecode plugin, and scope restriction to token plugin.

- allows limiting the usage of builtin auth to HTTP only (so force usage of tokens)
- allows migration to something like saml keeping only super-admin for login.

rhodecode/apps/login/tests/test_login.py

@@ -1,569 +1,579 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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 urlparse
 
 import mock
 import pytest
 
 from rhodecode.tests import (
     assert_session_flash, HG_REPO, TEST_USER_ADMIN_LOGIN,
     no_newline_id_generator)
 from rhodecode.tests.fixture import Fixture
 from rhodecode.lib.auth import check_password
 from rhodecode.lib import helpers as h
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model.db import User, Notification, UserApiKeys
 from rhodecode.model.meta import Session
 
 fixture = Fixture()
 
 whitelist_view = ['RepoCommitsView:repo_commit_raw']
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
     from rhodecode.apps._base import ADMIN_PREFIX
 
     base_url = {
         'login': ADMIN_PREFIX + '/login',
         'logout': ADMIN_PREFIX + '/logout',
         'register': ADMIN_PREFIX + '/register',
         'reset_password':
             ADMIN_PREFIX + '/password_reset',
         'reset_password_confirmation':
             ADMIN_PREFIX + '/password_reset_confirmation',
 
         'admin_permissions_application':
             ADMIN_PREFIX + '/permissions/application',
         'admin_permissions_application_update':
             ADMIN_PREFIX + '/permissions/application/update',
 
         'repo_commit_raw': '/{repo_name}/raw-changeset/{commit_id}'
 
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
     return base_url
 
 
 @pytest.mark.usefixtures('app')
 class TestLoginController(object):
     destroy_users = set()
 
     @classmethod
     def teardown_class(cls):
         fixture.destroy_users(cls.destroy_users)
 
     def teardown_method(self, method):
         for n in Notification.query().all():
             Session().delete(n)
 
         Session().commit()
         assert Notification.query().all() == []
 
     def test_index(self):
         response = self.app.get(route_path('login'))
         assert response.status == '200 OK'
         # Test response...
 
     def test_login_admin_ok(self):
         response = self.app.post(route_path('login'),
                                  {'username': 'test_admin',
                                   'password': 'test12'}, status=302)
         response = response.follow()
         session = response.get_session_from_response()
         username = session['rhodecode_user'].get('username')
         assert username == 'test_admin'
         response.mustcontain('/%s' % HG_REPO)
 
     def test_login_regular_ok(self):
         response = self.app.post(route_path('login'),
                                  {'username': 'test_regular',
                                   'password': 'test12'}, status=302)
 
         response = response.follow()
         session = response.get_session_from_response()
         username = session['rhodecode_user'].get('username')
         assert username == 'test_regular'
 
         response.mustcontain('/%s' % HG_REPO)
 
     def test_login_regular_forbidden_when_super_admin_restriction(self):
         from rhodecode.authentication.plugins.auth_rhodecode import RhodeCodeAuthPlugin
-        with fixture.login_restriction(RhodeCodeAuthPlugin.LOGIN_RESTRICTION_SUPER_ADMIN):
+        with fixture.auth_restriction(RhodeCodeAuthPlugin.AUTH_RESTRICTION_SUPER_ADMIN):
+            response = self.app.post(route_path('login'),
+                                     {'username': 'test_regular',
+                                      'password': 'test12'})
+
+            response.mustcontain('invalid user name')
+            response.mustcontain('invalid password')
+
+    def test_login_regular_forbidden_when_scope_restriction(self):
+        from rhodecode.authentication.plugins.auth_rhodecode import RhodeCodeAuthPlugin
+        with fixture.scope_restriction(RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_VCS):
             response = self.app.post(route_path('login'),
                                      {'username': 'test_regular',
                                       'password': 'test12'})
 
             response.mustcontain('invalid user name')
             response.mustcontain('invalid password')
 
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users?branch=stable'
         _url = '{}?came_from={}'.format(route_path('login'), test_came_from)
         response = self.app.post(
             _url, {'username': 'test_admin', 'password': 'test12'}, status=302)
 
         assert 'branch=stable' in response.location
         response = response.follow()
 
         assert response.status == '200 OK'
         response.mustcontain('Users administration')
 
     def test_redirect_to_login_with_get_args(self):
         with fixture.anon_access(False):
             kwargs = {'branch': 'stable'}
             response = self.app.get(
                 h.route_path('repo_summary', repo_name=HG_REPO, _query=kwargs),
                 status=302)
 
             response_query = urlparse.parse_qsl(response.location)
             assert 'branch=stable' in response_query[0][1]
 
     def test_login_form_with_get_args(self):
         _url = '{}?came_from=/_admin/users,branch=stable'.format(route_path('login'))
         response = self.app.get(_url)
         assert 'branch%3Dstable' in response.form.action
 
     @pytest.mark.parametrize("url_came_from", [
         'data:text/html,<script>window.alert("xss")</script>',
         'mailto:[e-mail]',
         'file:///etc/passwd',
         'ftp://some.ftp.server',
         'http://other.domain',
         '/\r\nX-Forwarded-Host: http://example.org',
     ], ids=no_newline_id_generator)
     def test_login_bad_came_froms(self, url_came_from):
         _url = '{}?came_from={}'.format(route_path('login'), url_came_from)
         response = self.app.post(
             _url,
             {'username': 'test_admin', 'password': 'test12'})
         assert response.status == '302 Found'
         response = response.follow()
         assert response.status == '200 OK'
         assert response.request.path == '/'
 
     def test_login_short_password(self):
         response = self.app.post(route_path('login'),
                                  {'username': 'test_admin',
                                   'password': 'as'})
         assert response.status == '200 OK'
 
         response.mustcontain('Enter 3 characters or more')
 
     def test_login_wrong_non_ascii_password(self, user_regular):
         response = self.app.post(
             route_path('login'),
             {'username': user_regular.username,
              'password': u'invalid-non-asci\xe4'.encode('utf8')})
 
         response.mustcontain('invalid user name')
         response.mustcontain('invalid password')
 
     def test_login_with_non_ascii_password(self, user_util):
         password = u'valid-non-ascii\xe4'
         user = user_util.create_user(password=password)
         response = self.app.post(
             route_path('login'),
             {'username': user.username,
              'password': password.encode('utf-8')})
         assert response.status_code == 302
 
     def test_login_wrong_username_password(self):
         response = self.app.post(route_path('login'),
                                  {'username': 'error',
                                   'password': 'test12'})
 
         response.mustcontain('invalid user name')
         response.mustcontain('invalid password')
 
     def test_login_admin_ok_password_migration(self, real_crypto_backend):
         from rhodecode.lib import auth
 
         # create new user, with sha256 password
         temp_user = 'test_admin_sha256'
         user = fixture.create_user(temp_user)
         user.password = auth._RhodeCodeCryptoSha256().hash_create(
             b'test123')
         Session().add(user)
         Session().commit()
         self.destroy_users.add(temp_user)
         response = self.app.post(route_path('login'),
                                  {'username': temp_user,
                                   'password': 'test123'}, status=302)
 
         response = response.follow()
         session = response.get_session_from_response()
         username = session['rhodecode_user'].get('username')
         assert username == temp_user
         response.mustcontain('/%s' % HG_REPO)
 
         # new password should be bcrypted, after log-in and transfer
         user = User.get_by_username(temp_user)
         assert user.password.startswith('$')
 
     # REGISTRATIONS
     def test_register(self):
         response = self.app.get(route_path('register'))
         response.mustcontain('Create an Account')
 
     def test_register_err_same_username(self):
         uname = 'test_admin'
         response = self.app.post(
             route_path('register'),
             {
                 'username': uname,
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         assertr = response.assert_response()
         msg = 'Username "%(username)s" already exists'
         msg = msg % {'username': uname}
         assertr.element_contains('#username+.error-message', msg)
 
     def test_register_err_same_email(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'test_admin_0',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         assertr = response.assert_response()
         msg = u'This e-mail address is already taken'
         assertr.element_contains('#email+.error-message', msg)
 
     def test_register_err_same_email_case_sensitive(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'test_admin_1',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         assertr = response.assert_response()
         msg = u'This e-mail address is already taken'
         assertr.element_contains('#email+.error-message', msg)
 
     def test_register_err_wrong_data(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'xs',
                 'password': 'test',
                 'password_confirmation': 'test',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         assert response.status == '200 OK'
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
 
     def test_register_err_username(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'error user',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         response.mustcontain('An email address must contain a single @')
         response.mustcontain(
             'Username may only contain '
             'alphanumeric characters underscores, '
             'periods or dashes and must begin with '
             'alphanumeric character')
 
     def test_register_err_case_sensitive(self):
         usr = 'Test_Admin'
         response = self.app.post(
             route_path('register'),
             {
                 'username': usr,
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         assertr = response.assert_response()
         msg = u'Username "%(username)s" already exists'
         msg = msg % {'username': usr}
         assertr.element_contains('#username+.error-message', msg)
 
     def test_register_special_chars(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'xxxaxn',
                 'password': 'ąćźżąśśśś',
                 'password_confirmation': 'ąćźżąśśśś',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         msg = u'Invalid characters (non-ascii) in password'
         response.mustcontain(msg)
 
     def test_register_password_mismatch(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'xs',
                 'password': '123qwe',
                 'password_confirmation': 'qwe123',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         msg = u'Passwords do not match'
         response.mustcontain(msg)
 
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = '[e-mail]'
         name = 'testname'
         lastname = 'testlastname'
 
         # this initializes a session
         response = self.app.get(route_path('register'))
         response.mustcontain('Create an Account')
 
 
         response = self.app.post(
             route_path('register'),
             {
                 'username': username,
                 'password': password,
                 'password_confirmation': password,
                 'email': email,
                 'firstname': name,
                 'lastname': lastname,
                 'admin': True
             },
             status=302
         )  # This should be overridden
 
         assert_session_flash(
             response, 'You have successfully registered with RhodeCode')
 
         ret = Session().query(User).filter(
             User.username == 'test_regular4').one()
         assert ret.username == username
         assert check_password(password, ret.password)
         assert ret.email == email
         assert ret.name == name
         assert ret.lastname == lastname
         assert ret.auth_tokens is not None
         assert not ret.admin
 
     def test_forgot_password_wrong_mail(self):
         bad_email = '[e-mail]'
         # this initializes a session
         self.app.get(route_path('reset_password'))
 
         response = self.app.post(
             route_path('reset_password'), {'email': bad_email, }
         )
         assert_session_flash(response,
             'If such email exists, a password reset link was sent to it.')
 
     def test_forgot_password(self, user_util):
         # this initializes a session
         self.app.get(route_path('reset_password'))
 
         user = user_util.create_user()
         user_id = user.user_id
         email = user.email
 
         response = self.app.post(route_path('reset_password'), {'email': email, })
 
         assert_session_flash(response,
             'If such email exists, a password reset link was sent to it.')
 
         # BAD KEY
         confirm_url = '{}?key={}'.format(route_path('reset_password_confirmation'), 'badkey')
         response = self.app.get(confirm_url, status=302)
         assert response.location.endswith(route_path('reset_password'))
         assert_session_flash(response, 'Given reset token is invalid')
 
         response.follow()  # cleanup flash
 
         # GOOD KEY
         key = UserApiKeys.query()\
             .filter(UserApiKeys.user_id == user_id)\
             .filter(UserApiKeys.role == UserApiKeys.ROLE_PASSWORD_RESET)\
             .first()
 
         assert key
 
         confirm_url = '{}?key={}'.format(route_path('reset_password_confirmation'), key.api_key)
         response = self.app.get(confirm_url)
         assert response.status == '302 Found'
         assert response.location.endswith(route_path('login'))
 
         assert_session_flash(
             response,
             'Your password reset was successful, '
             'a new password has been sent to your email')
 
         response.follow()
 
     def _get_api_whitelist(self, values=None):
         config = {'api_access_controllers_whitelist': values or []}
         return config
 
     @pytest.mark.parametrize("test_name, auth_token", [
         ('none', None),
         ('empty_string', ''),
         ('fake_number', '123456'),
         ('proper_auth_token', None)
     ])
     def test_access_not_whitelisted_page_via_auth_token(
             self, test_name, auth_token, user_admin):
 
         whitelist = self._get_api_whitelist([])
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert [] == whitelist['api_access_controllers_whitelist']
             if test_name == 'proper_auth_token':
                 # use builtin if api_key is None
                 auth_token = user_admin.api_key
 
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=auth_token)),
                     status=302)
 
     @pytest.mark.parametrize("test_name, auth_token, code", [
         ('none', None, 302),
         ('empty_string', '', 302),
         ('fake_number', '123456', 302),
         ('proper_auth_token', None, 200)
     ])
     def test_access_whitelisted_page_via_auth_token(
             self, test_name, auth_token, code, user_admin):
 
         whitelist = self._get_api_whitelist(whitelist_view)
 
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert whitelist_view == whitelist['api_access_controllers_whitelist']
 
             if test_name == 'proper_auth_token':
                 auth_token = user_admin.api_key
                 assert auth_token
 
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=auth_token)),
                     status=code)
 
     @pytest.mark.parametrize("test_name, auth_token, code", [
         ('proper_auth_token', None, 200),
         ('wrong_auth_token', '123456', 302),
     ])
     def test_access_whitelisted_page_via_auth_token_bound_to_token(
             self, test_name, auth_token, code, user_admin):
 
         expected_token = auth_token
         if test_name == 'proper_auth_token':
             auth_token = user_admin.api_key
             expected_token = auth_token
             assert auth_token
 
         whitelist = self._get_api_whitelist([
             'RepoCommitsView:repo_commit_raw@{}'.format(expected_token)])
 
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
 
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=auth_token)),
                     status=code)
 
     def test_access_page_via_extra_auth_token(self):
         whitelist = self._get_api_whitelist(whitelist_view)
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert whitelist_view == \
                 whitelist['api_access_controllers_whitelist']
 
             new_auth_token = AuthTokenModel().create(
                 TEST_USER_ADMIN_LOGIN, 'test')
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=new_auth_token.api_key)),
                     status=200)
 
     def test_access_page_via_expired_auth_token(self):
         whitelist = self._get_api_whitelist(whitelist_view)
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert whitelist_view == \
                 whitelist['api_access_controllers_whitelist']
 
             new_auth_token = AuthTokenModel().create(
                 TEST_USER_ADMIN_LOGIN, 'test')
             Session().commit()
             # patch the api key and make it expired
             new_auth_token.expires = 0
             Session().add(new_auth_token)
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=new_auth_token.api_key)),
                     status=302)

rhodecode/authentication/plugins/auth_rhodecode.py

@@ -1,186 +1,220 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2012-2019 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/
 
 """
 RhodeCode authentication plugin for built in internal auth
 """
 
 import logging
 
 import colander
 
-from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
 from rhodecode.translation import _
-
-from rhodecode.authentication.base import RhodeCodeAuthPluginBase, hybrid_property
-from rhodecode.authentication.routes import AuthnPluginResourceBase
 from rhodecode.lib.utils2 import safe_str
 from rhodecode.model.db import User
+from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
+from rhodecode.authentication.base import (
+    RhodeCodeAuthPluginBase, hybrid_property, HTTP_TYPE, VCS_TYPE)
+from rhodecode.authentication.routes import AuthnPluginResourceBase
 
 log = logging.getLogger(__name__)
 
 
 def plugin_factory(plugin_id, *args, **kwargs):
     plugin = RhodeCodeAuthPlugin(plugin_id)
     return plugin
 
 
 class RhodecodeAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class RhodeCodeAuthPlugin(RhodeCodeAuthPluginBase):
     uid = 'rhodecode'
-    LOGIN_RESTRICTION_NONE = 'none'
-    LOGIN_RESTRICTION_SUPER_ADMIN = 'super_admin'
+    AUTH_RESTRICTION_NONE = 'user_all'
+    AUTH_RESTRICTION_SUPER_ADMIN = 'user_super_admin'
+    AUTH_RESTRICTION_SCOPE_ALL = 'scope_all'
+    AUTH_RESTRICTION_SCOPE_HTTP = 'scope_http'
+    AUTH_RESTRICTION_SCOPE_VCS = 'scope_vcs'
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), RhodecodeAuthnResource(self))
         config.add_view(
             'rhodecode.authentication.views.AuthnPluginViewBase',
             attr='settings_get',
             renderer='rhodecode:templates/admin/auth/plugin_settings.mako',
             request_method='GET',
             route_name='auth_home',
             context=RhodecodeAuthnResource)
         config.add_view(
             'rhodecode.authentication.views.AuthnPluginViewBase',
             attr='settings_post',
             renderer='rhodecode:templates/admin/auth/plugin_settings.mako',
             request_method='POST',
             route_name='auth_home',
             context=RhodecodeAuthnResource)
 
     def get_settings_schema(self):
         return RhodeCodeSettingsSchema()
 
     def get_display_name(self):
         return _('RhodeCode Internal')
 
     @classmethod
     def docs(cls):
         return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth.html"
 
     @hybrid_property
     def name(self):
         return u"rhodecode"
 
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser().permissions['global']
         return 'hg.register.auto_activate' in def_user_perms
 
     def allows_authentication_from(
             self, user, allows_non_existing_user=True,
             allowed_auth_plugins=None, allowed_auth_sources=None):
         """
         Custom method for this auth that doesn't accept non existing users.
         We know that user exists in our database.
         """
         allows_non_existing_user = False
         return super(RhodeCodeAuthPlugin, self).allows_authentication_from(
             user, allows_non_existing_user=allows_non_existing_user)
 
     def auth(self, userobj, username, password, settings, **kwargs):
         if not userobj:
             log.debug('userobj was:%s skipping', userobj)
             return None
 
         if userobj.extern_type != self.name:
-            log.warning(
-                "userobj:%s extern_type mismatch got:`%s` expected:`%s`",
-                userobj, userobj.extern_type, self.name)
+            log.warning("userobj:%s extern_type mismatch got:`%s` expected:`%s`",
+                        userobj, userobj.extern_type, self.name)
+            return None
+
+        # check scope of auth
+        scope_restriction = settings.get('scope_restriction', '')
+
+        if scope_restriction == self.AUTH_RESTRICTION_SCOPE_HTTP \
+                and self.auth_type != HTTP_TYPE:
+            log.warning("userobj:%s tried scope type %s and scope restriction is set to %s",
+                        userobj, self.auth_type, scope_restriction)
             return None
 
-        login_restriction = settings.get('login_restriction', '')
-        if login_restriction == self.LOGIN_RESTRICTION_SUPER_ADMIN and userobj.admin is False:
-            log.info(
-                "userobj:%s is not super-admin and login restriction is set to %s",
-                userobj, login_restriction)
+        if scope_restriction == self.AUTH_RESTRICTION_SCOPE_VCS \
+                and self.auth_type != VCS_TYPE:
+            log.warning("userobj:%s tried scope type %s and scope restriction is set to %s",
+                        userobj, self.auth_type, scope_restriction)
+            return None
+
+        # check super-admin restriction
+        auth_restriction = settings.get('auth_restriction', '')
+
+        if auth_restriction == self.AUTH_RESTRICTION_SUPER_ADMIN \
+                and userobj.admin is False:
+            log.warning("userobj:%s is not super-admin and auth restriction is set to %s",
+                        userobj, auth_restriction)
             return None
 
         user_attrs = {
             "username": userobj.username,
             "firstname": userobj.firstname,
             "lastname": userobj.lastname,
             "groups": [],
             'user_group_sync': False,
             "email": userobj.email,
             "admin": userobj.admin,
             "active": userobj.active,
             "active_from_extern": userobj.active,
             "extern_name": userobj.user_id,
             "extern_type": userobj.extern_type,
         }
 
         log.debug("User attributes:%s", user_attrs)
         if userobj.active:
             from rhodecode.lib import auth
             crypto_backend = auth.crypto_backend()
             password_encoded = safe_str(password)
             password_match, new_hash = crypto_backend.hash_check_with_upgrade(
                 password_encoded, userobj.password or '')
 
             if password_match and new_hash:
                 log.debug('user %s properly authenticated, but '
                           'requires hash change to bcrypt', userobj)
                 # if password match, and we use OLD deprecated hash,
                 # we should migrate this user hash password to the new hash
                 # we store the new returned by hash_check_with_upgrade function
                 user_attrs['_hash_migrate'] = new_hash
 
             if userobj.username == User.DEFAULT_USER and userobj.active:
                 log.info('user `%s` authenticated correctly as anonymous user',
                          userobj.username)
                 return user_attrs
 
             elif userobj.username == username and password_match:
                 log.info('user `%s` authenticated correctly', userobj.username)
                 return user_attrs
-            log.warn("user `%s` used a wrong password when "
-                     "authenticating on this plugin", userobj.username)
+            log.warning("user `%s` used a wrong password when "
+                        "authenticating on this plugin", userobj.username)
             return None
         else:
-            log.warning(
-                'user `%s` failed to authenticate via %s, reason: account not '
-                'active.', username, self.name)
+            log.warning('user `%s` failed to authenticate via %s, reason: account not '
+                        'active.', username, self.name)
             return None
 
 
 class RhodeCodeSettingsSchema(AuthnPluginSettingsSchemaBase):
-    login_restriction_choices = [
-        (RhodeCodeAuthPlugin.LOGIN_RESTRICTION_NONE, 'All users'),
-        (RhodeCodeAuthPlugin.LOGIN_RESTRICTION_SUPER_ADMIN, 'Super admins only')
+
+    auth_restriction_choices = [
+        (RhodeCodeAuthPlugin.AUTH_RESTRICTION_NONE, 'All users'),
+        (RhodeCodeAuthPlugin.AUTH_RESTRICTION_SUPER_ADMIN, 'Super admins only'),
+    ]
+
+    auth_scope_choices = [
+        (RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_ALL, 'HTTP and VCS'),
+        (RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_HTTP, 'HTTP only'),
     ]
 
-    login_restriction = colander.SchemaNode(
+    auth_restriction = colander.SchemaNode(
         colander.String(),
-        default=login_restriction_choices[0],
-        description=_('Choose login restrition for users.'),
-        title=_('Login restriction'),
-        validator=colander.OneOf([x[0] for x in login_restriction_choices]),
+        default=auth_restriction_choices[0],
+        description=_('Allowed user types for authentication using this plugin.'),
+        title=_('User restriction'),
+        validator=colander.OneOf([x[0] for x in auth_restriction_choices]),
         widget='select_with_labels',
-        choices=login_restriction_choices
+        choices=auth_restriction_choices
+    )
+    scope_restriction = colander.SchemaNode(
+        colander.String(),
+        default=auth_scope_choices[0],
+        description=_('Allowed protocols for authentication using this plugin. '
+                      'VCS means GIT/HG/SVN. HTTP is web based login.'),
+        title=_('Scope restriction'),
+        validator=colander.OneOf([x[0] for x in auth_scope_choices]),
+        widget='select_with_labels',
+        choices=auth_scope_choices
     )
 
 
 def includeme(config):
     plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
     plugin_factory(plugin_id).includeme(config)

rhodecode/authentication/plugins/auth_token.py

@@ -1,157 +1,177 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2019 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/
 
 """
 RhodeCode authentication token plugin for built in internal auth
 """
 
 import logging
+import colander
 
+from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
 from rhodecode.translation import _
 from rhodecode.authentication.base import (
     RhodeCodeAuthPluginBase, VCS_TYPE, hybrid_property)
 from rhodecode.authentication.routes import AuthnPluginResourceBase
 from rhodecode.model.db import User, UserApiKeys, Repository
 
 
 log = logging.getLogger(__name__)
 
 
 def plugin_factory(plugin_id, *args, **kwargs):
     plugin = RhodeCodeAuthPlugin(plugin_id)
     return plugin
 
 
 class RhodecodeAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class RhodeCodeAuthPlugin(RhodeCodeAuthPluginBase):
     """
     Enables usage of authentication tokens for vcs operations.
     """
     uid = 'token'
+    AUTH_RESTRICTION_SCOPE_VCS = 'scope_vcs'
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), RhodecodeAuthnResource(self))
         config.add_view(
             'rhodecode.authentication.views.AuthnPluginViewBase',
             attr='settings_get',
             renderer='rhodecode:templates/admin/auth/plugin_settings.mako',
             request_method='GET',
             route_name='auth_home',
             context=RhodecodeAuthnResource)
         config.add_view(
             'rhodecode.authentication.views.AuthnPluginViewBase',
             attr='settings_post',
             renderer='rhodecode:templates/admin/auth/plugin_settings.mako',
             request_method='POST',
             route_name='auth_home',
             context=RhodecodeAuthnResource)
 
+    def get_settings_schema(self):
+        return RhodeCodeSettingsSchema()
+
     def get_display_name(self):
         return _('Rhodecode Token')
 
     @classmethod
     def docs(cls):
         return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth-token.html"
 
     @hybrid_property
     def name(self):
         return u"authtoken"
 
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser().permissions['global']
         return 'hg.register.auto_activate' in def_user_perms
 
     def allows_authentication_from(
             self, user, allows_non_existing_user=True,
             allowed_auth_plugins=None, allowed_auth_sources=None):
         """
         Custom method for this auth that doesn't accept empty users. And also
         allows users from all other active plugins to use it and also
         authenticate against it. But only via vcs mode
         """
         from rhodecode.authentication.base import get_authn_registry
         authn_registry = get_authn_registry()
 
         active_plugins = set(
             [x.name for x in authn_registry.get_plugins_for_authentication()])
         active_plugins.discard(self.name)
 
         allowed_auth_plugins = [self.name] + list(active_plugins)
         # only for vcs operations
         allowed_auth_sources = [VCS_TYPE]
 
         return super(RhodeCodeAuthPlugin, self).allows_authentication_from(
             user, allows_non_existing_user=False,
             allowed_auth_plugins=allowed_auth_plugins,
             allowed_auth_sources=allowed_auth_sources)
 
     def auth(self, userobj, username, password, settings, **kwargs):
         if not userobj:
             log.debug('userobj was:%s skipping', userobj)
             return None
 
         user_attrs = {
             "username": userobj.username,
             "firstname": userobj.firstname,
             "lastname": userobj.lastname,
             "groups": [],
             'user_group_sync': False,
             "email": userobj.email,
             "admin": userobj.admin,
             "active": userobj.active,
             "active_from_extern": userobj.active,
             "extern_name": userobj.user_id,
             "extern_type": userobj.extern_type,
         }
 
         log.debug('Authenticating user with args %s', user_attrs)
         if userobj.active:
             # calling context repo for token scopes
             scope_repo_id = None
             if self.acl_repo_name:
                 repo = Repository.get_by_repo_name(self.acl_repo_name)
                 scope_repo_id = repo.repo_id if repo else None
 
             token_match = userobj.authenticate_by_token(
                 password, roles=[UserApiKeys.ROLE_VCS],
                 scope_repo_id=scope_repo_id)
 
             if userobj.username == username and token_match:
                 log.info(
                     'user `%s` successfully authenticated via %s',
                     user_attrs['username'], self.name)
                 return user_attrs
-            log.warn(
-                'user `%s` failed to authenticate via %s, reason: bad or '
-                'inactive token.', username, self.name)
+            log.warning('user `%s` failed to authenticate via %s, reason: bad or '
+                        'inactive token.', username, self.name)
         else:
-            log.warning(
-                'user `%s` failed to authenticate via %s, reason: account not '
-                'active.', username, self.name)
+            log.warning('user `%s` failed to authenticate via %s, reason: account not '
+                        'active.', username, self.name)
         return None
 
 
 def includeme(config):
     plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
     plugin_factory(plugin_id).includeme(config)
+
+
+class RhodeCodeSettingsSchema(AuthnPluginSettingsSchemaBase):
+    auth_scope_choices = [
+        (RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_VCS, 'VCS only'),
+    ]
+
+    scope_restriction = colander.SchemaNode(
+        colander.String(),
+        default=auth_scope_choices[0],
+        description=_('Choose operation scope restriction when authenticating.'),
+        title=_('Scope restriction'),
+        validator=colander.OneOf([x[0] for x in auth_scope_choices]),
+        widget='select_with_labels',
+        choices=auth_scope_choices
+    )

rhodecode/tests/fixture.py

@@ -1,383 +1,415 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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/
 
 """
 Helpers for fixture generation
 """
 
 import os
 import time
 import tempfile
 import shutil
 
 import configobj
 
 from rhodecode.tests import *
 from rhodecode.model.db import Repository, User, RepoGroup, UserGroup, Gist, UserEmailMap
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.model.gist import GistModel
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.authentication.plugins.auth_rhodecode import \
     RhodeCodeAuthPlugin
 
 dn = os.path.dirname
 FIXTURES = os.path.join(dn(dn(os.path.abspath(__file__))), 'tests', 'fixtures')
 
 
 def error_function(*args, **kwargs):
     raise Exception('Total Crash !')
 
 
 class TestINI(object):
     """
     Allows to create a new test.ini file as a copy of existing one with edited
     data. Example usage::
 
         with TestINI('test.ini', [{'section':{'key':val'}]) as new_test_ini_path:
             print('paster server %s' % new_test_ini)
     """
 
     def __init__(self, ini_file_path, ini_params, new_file_prefix='DEFAULT',
                  destroy=True, dir=None):
         self.ini_file_path = ini_file_path
         self.ini_params = ini_params
         self.new_path = None
         self.new_path_prefix = new_file_prefix
         self._destroy = destroy
         self._dir = dir
 
     def __enter__(self):
         return self.create()
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.destroy()
 
     def create(self):
         config = configobj.ConfigObj(
             self.ini_file_path, file_error=True, write_empty_values=True)
 
         for data in self.ini_params:
             section, ini_params = data.items()[0]
             for key, val in ini_params.items():
                 config[section][key] = val
         with tempfile.NamedTemporaryFile(
                 prefix=self.new_path_prefix, suffix='.ini', dir=self._dir,
                 delete=False) as new_ini_file:
             config.write(new_ini_file)
             self.new_path = new_ini_file.name
 
         return self.new_path
 
     def destroy(self):
         if self._destroy:
             os.remove(self.new_path)
 
 
 class Fixture(object):
 
     def anon_access(self, status):
         """
         Context process for disabling anonymous access. use like:
         fixture = Fixture()
         with fixture.anon_access(False):
             #tests
 
         after this block anon access will be set to `not status`
         """
 
         class context(object):
             def __enter__(self):
                 anon = User.get_default_user()
                 anon.active = status
                 Session().add(anon)
                 Session().commit()
                 time.sleep(1.5)  # must sleep for cache (1s to expire)
 
             def __exit__(self, exc_type, exc_val, exc_tb):
                 anon = User.get_default_user()
                 anon.active = not status
                 Session().add(anon)
                 Session().commit()
 
         return context()
 
-    def login_restriction(self, login_restriction):
+    def auth_restriction(self, auth_restriction):
         """
-        Context process for changing the builtin rhodecode plugin login restrictions.
+        Context process for changing the builtin rhodecode plugin auth restrictions.
         Use like:
         fixture = Fixture()
-        with fixture.login_restriction('super_admin'):
+        with fixture.auth_restriction('super_admin'):
             #tests
 
-        after this block login restriction will be taken off
+        after this block auth restriction will be taken off
         """
 
         class context(object):
             def _get_pluing(self):
                 plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(
                     RhodeCodeAuthPlugin.uid)
                 plugin = RhodeCodeAuthPlugin(plugin_id)
                 return plugin
 
             def __enter__(self):
                 plugin = self._get_pluing()
                 plugin.create_or_update_setting(
-                    'login_restriction', login_restriction)
+                    'auth_restriction', auth_restriction)
                 Session().commit()
 
             def __exit__(self, exc_type, exc_val, exc_tb):
                 plugin = self._get_pluing()
                 plugin.create_or_update_setting(
-                    'login_restriction', RhodeCodeAuthPlugin.LOGIN_RESTRICTION_NONE)
+                    'auth_restriction', RhodeCodeAuthPlugin.AUTH_RESTRICTION_NONE)
+                Session().commit()
+
+        return context()
+
+    def scope_restriction(self, scope_restriction):
+        """
+        Context process for changing the builtin rhodecode plugin scope restrictions.
+        Use like:
+        fixture = Fixture()
+        with fixture.scope_restriction('scope_http'):
+            #tests
+
+        after this block scope restriction will be taken off
+        """
+
+        class context(object):
+            def _get_pluing(self):
+                plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(
+                    RhodeCodeAuthPlugin.uid)
+                plugin = RhodeCodeAuthPlugin(plugin_id)
+                return plugin
+
+            def __enter__(self):
+                plugin = self._get_pluing()
+                plugin.create_or_update_setting(
+                    'scope_restriction', scope_restriction)
+                Session().commit()
+
+            def __exit__(self, exc_type, exc_val, exc_tb):
+                plugin = self._get_pluing()
+                plugin.create_or_update_setting(
+                    'scope_restriction', RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_ALL)
                 Session().commit()
 
         return context()
 
     def _get_repo_create_params(self, **custom):
         defs = {
             'repo_name': None,
             'repo_type': 'hg',
             'clone_uri': '',
             'push_uri': '',
             'repo_group': '-1',
             'repo_description': 'DESC',
             'repo_private': False,
             'repo_landing_rev': 'rev:tip',
             'repo_copy_permissions': False,
             'repo_state': Repository.STATE_CREATED,
         }
         defs.update(custom)
         if 'repo_name_full' not in custom:
             defs.update({'repo_name_full': defs['repo_name']})
 
         # fix the repo name if passed as repo_name_full
         if defs['repo_name']:
             defs['repo_name'] = defs['repo_name'].split('/')[-1]
 
         return defs
 
     def _get_group_create_params(self, **custom):
         defs = {
             'group_name': None,
             'group_description': 'DESC',
             'perm_updates': [],
             'perm_additions': [],
             'perm_deletions': [],
             'group_parent_id': -1,
             'enable_locking': False,
             'recursive': False,
         }
         defs.update(custom)
 
         return defs
 
     def _get_user_create_params(self, name, **custom):
         defs = {
             'username': name,
             'password': 'qweqwe',
             'email': '[e-mail]' % name,
             'firstname': 'TestUser',
             'lastname': 'Test',
             'active': True,
             'admin': False,
             'extern_type': 'rhodecode',
             'extern_name': None,
         }
         defs.update(custom)
 
         return defs
 
     def _get_user_group_create_params(self, name, **custom):
         defs = {
             'users_group_name': name,
             'user_group_description': 'DESC',
             'users_group_active': True,
             'user_group_data': {},
         }
         defs.update(custom)
 
         return defs
 
     def create_repo(self, name, **kwargs):
         repo_group = kwargs.get('repo_group')
         if isinstance(repo_group, RepoGroup):
             kwargs['repo_group'] = repo_group.group_id
             name = name.split(Repository.NAME_SEP)[-1]
             name = Repository.NAME_SEP.join((repo_group.group_name, name))
 
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             r = Repository.get_by_repo_name(name)
             if r:
                 return r
 
         form_data = self._get_repo_create_params(repo_name=name, **kwargs)
         cur_user = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         RepoModel().create(form_data, cur_user)
         Session().commit()
         repo = Repository.get_by_repo_name(name)
         assert repo
         return repo
 
     def create_fork(self, repo_to_fork, fork_name, **kwargs):
         repo_to_fork = Repository.get_by_repo_name(repo_to_fork)
 
         form_data = self._get_repo_create_params(repo_name=fork_name,
                                             fork_parent_id=repo_to_fork.repo_id,
                                             repo_type=repo_to_fork.repo_type,
                                             **kwargs)
         #TODO: fix it !!
         form_data['description'] = form_data['repo_description']
         form_data['private'] = form_data['repo_private']
         form_data['landing_rev'] = form_data['repo_landing_rev']
 
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         RepoModel().create_fork(form_data, cur_user=owner)
         Session().commit()
         r = Repository.get_by_repo_name(fork_name)
         assert r
         return r
 
     def destroy_repo(self, repo_name, **kwargs):
         RepoModel().delete(repo_name, pull_requests='delete', **kwargs)
         Session().commit()
 
     def destroy_repo_on_filesystem(self, repo_name):
         rm_path = os.path.join(RepoModel().repos_path, repo_name)
         if os.path.isdir(rm_path):
             shutil.rmtree(rm_path)
 
     def create_repo_group(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             gr = RepoGroup.get_by_group_name(group_name=name)
             if gr:
                 return gr
         form_data = self._get_group_create_params(group_name=name, **kwargs)
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         gr = RepoGroupModel().create(
             group_name=form_data['group_name'],
             group_description=form_data['group_name'],
             owner=owner)
         Session().commit()
         gr = RepoGroup.get_by_group_name(gr.group_name)
         return gr
 
     def destroy_repo_group(self, repogroupid):
         RepoGroupModel().delete(repogroupid)
         Session().commit()
 
     def create_user(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             user = User.get_by_username(name)
             if user:
                 return user
         form_data = self._get_user_create_params(name, **kwargs)
         user = UserModel().create(form_data)
 
         # create token for user
         AuthTokenModel().create(
                 user=user, description=u'TEST_USER_TOKEN')
 
         Session().commit()
         user = User.get_by_username(user.username)
         return user
 
     def destroy_user(self, userid):
         UserModel().delete(userid)
         Session().commit()
 
     def create_additional_user_email(self, user, email):
         uem = UserEmailMap()
         uem.user = user
         uem.email = email
         Session().add(uem)
         return uem
 
     def destroy_users(self, userid_iter):
         for user_id in userid_iter:
             if User.get_by_username(user_id):
                 UserModel().delete(user_id)
                 Session().commit()
 
     def create_user_group(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             gr = UserGroup.get_by_group_name(group_name=name)
             if gr:
                 return gr
         # map active flag to the real attribute. For API consistency of fixtures
         if 'active' in kwargs:
             kwargs['users_group_active'] = kwargs['active']
             del kwargs['active']
         form_data = self._get_user_group_create_params(name, **kwargs)
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         user_group = UserGroupModel().create(
             name=form_data['users_group_name'],
             description=form_data['user_group_description'],
             owner=owner, active=form_data['users_group_active'],
             group_data=form_data['user_group_data'])
         Session().commit()
         user_group = UserGroup.get_by_group_name(user_group.users_group_name)
         return user_group
 
     def destroy_user_group(self, usergroupid):
         UserGroupModel().delete(user_group=usergroupid, force=True)
         Session().commit()
 
     def create_gist(self, **kwargs):
         form_data = {
             'description': 'new-gist',
             'owner': TEST_USER_ADMIN_LOGIN,
             'gist_type': GistModel.cls.GIST_PUBLIC,
             'lifetime': -1,
             'acl_level': Gist.ACL_LEVEL_PUBLIC,
             'gist_mapping': {'filename1.txt': {'content': 'hello world'},}
         }
         form_data.update(kwargs)
         gist = GistModel().create(
             description=form_data['description'], owner=form_data['owner'],
             gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],
             lifetime=form_data['lifetime'], gist_acl_level=form_data['acl_level']
         )
         Session().commit()
         return gist
 
     def destroy_gists(self, gistid=None):
         for g in GistModel.cls.get_all():
             if gistid:
                 if gistid == g.gist_access_id:
                     GistModel().delete(g)
             else:
                 GistModel().delete(g)
         Session().commit()
 
     def load_resource(self, resource_name, strip=False):
         with open(os.path.join(FIXTURES, resource_name)) as f:
             source = f.read()
             if strip:
                 source = source.strip()
 
         return source