Merge branch 'upstream'

VERSION.m4

@@ -21,7 +21,7 @@
 ########################################################
 define(IPA_VERSION_MAJOR, 4)
 define(IPA_VERSION_MINOR, 11)
-define(IPA_VERSION_RELEASE, 0)
+define(IPA_VERSION_RELEASE, 1)
 
 ########################################################
 # For 'pre' releases the version will be               #

ipaserver/rpcserver.py

@@ -156,6 +156,19 @@ _success_template = """<html>
 </html>"""
 
 class HTTP_Status(plugable.Plugin):
+    def check_referer(self, environ):
+        if "HTTP_REFERER" not in environ:
+            logger.error("Rejecting request with missing Referer")
+            return False
+        if (not environ["HTTP_REFERER"].startswith(
+                "https://%s/ipa" % self.api.env.host)
+                and not self.env.in_tree):
+            logger.error("Rejecting request with bad Referer %s",
+                         environ["HTTP_REFERER"])
+            return False
+        logger.debug("Valid Referer %s", environ["HTTP_REFERER"])
+        return True
+
     def not_found(self, environ, start_response, url, message):
         """
         Return a 404 Not Found error.
@@ -331,9 +344,6 @@ class wsgi_dispatch(Executioner, HTTP_Status):
         self.__apps[key] = app
 
 
-
-
-
 class WSGIExecutioner(Executioner):
     """
     Base class for execution backends with a WSGI application interface.
@@ -898,6 +908,9 @@ class jsonserver_session(jsonserver, KerberosSession):
 
         logger.debug('WSGI jsonserver_session.__call__:')
 
+        if not self.check_referer(environ):
+            return self.bad_request(environ, start_response, 'denied')
+
         # Redirect to login if no Kerberos credentials
         ccache_name = self.get_environ_creds(environ)
         if ccache_name is None:
@@ -950,6 +963,9 @@ class KerberosLogin(Backend, KerberosSession):
     def __call__(self, environ, start_response):
         logger.debug('WSGI KerberosLogin.__call__:')
 
+        if not self.check_referer(environ):
+            return self.bad_request(environ, start_response, 'denied')
+
         # Redirect to login if no Kerberos credentials
         user_ccache_name = self.get_environ_creds(environ)
         if user_ccache_name is None:
@@ -968,6 +984,9 @@ class login_x509(KerberosLogin):
     def __call__(self, environ, start_response):
         logger.debug('WSGI login_x509.__call__:')
 
+        if not self.check_referer(environ):
+            return self.bad_request(environ, start_response, 'denied')
+
         if 'KRB5CCNAME' not in environ:
             return self.unauthorized(
                 environ, start_response, 'KRB5CCNAME not set',
@@ -1016,6 +1035,9 @@ class login_password(Backend, KerberosSession):
 
         logger.debug('WSGI login_password.__call__:')
 
+        if not self.check_referer(environ):
+            return self.bad_request(environ, start_response, 'denied')
+
         # Get the user and password parameters from the request
         content_type = environ.get('CONTENT_TYPE', '').lower()
         if not content_type.startswith('application/x-www-form-urlencoded'):
@@ -1148,6 +1170,9 @@ class change_password(Backend, HTTP_Status):
     def __call__(self, environ, start_response):
         logger.info('WSGI change_password.__call__:')
 
+        if not self.check_referer(environ):
+            return self.bad_request(environ, start_response, 'denied')
+
         # Get the user and password parameters from the request
         content_type = environ.get('CONTENT_TYPE', '').lower()
         if not content_type.startswith('application/x-www-form-urlencoded'):
@@ -1365,6 +1390,9 @@ class xmlserver_session(xmlserver, KerberosSession):
 
         logger.debug('WSGI xmlserver_session.__call__:')
 
+        if not self.check_referer(environ):
+            return self.bad_request(environ, start_response, 'denied')
+
         ccache_name = environ.get('KRB5CCNAME')
 
         # Redirect to /ipa/xml if no Kerberos credentials

ipatests/test_ipaserver/httptest.py

@@ -36,7 +36,7 @@ class Unauthorized_HTTP_test:
     content_type = 'application/x-www-form-urlencoded'
     accept_language = 'en-us'
 
-    def send_request(self, method='POST', params=None):
+    def send_request(self, method='POST', params=None, host=None):
         """
         Send a request to HTTP server
 
@@ -45,7 +45,10 @@ class Unauthorized_HTTP_test:
         if params is not None:
             if self.content_type == 'application/x-www-form-urlencoded':
                 params = urllib.parse.urlencode(params, True)
-        url = 'https://' + self.host + self.app_uri
+        if host:
+            url = 'https://' + host + self.app_uri
+        else:
+            url = 'https://' + self.host + self.app_uri
 
         headers = {'Content-Type': self.content_type,
                    'Accept-Language': self.accept_language,

ipatests/test_ipaserver/test_changepw.py

@@ -53,10 +53,11 @@ class test_changepw(XMLRPC_test, Unauthorized_HTTP_test):
 
         request.addfinalizer(fin)
 
-    def _changepw(self, user, old_password, new_password):
+    def _changepw(self, user, old_password, new_password, host=None):
         return self.send_request(params={'user': str(user),
                                   'old_password' : str(old_password),
                                   'new_password' : str(new_password)},
+                                 host=host
                                  )
 
     def _checkpw(self, user, password):
@@ -89,6 +90,15 @@ class test_changepw(XMLRPC_test, Unauthorized_HTTP_test):
         # make sure that password is NOT changed
         self._checkpw(testuser, old_password)
 
+    def test_invalid_referer(self):
+        response = self._changepw(testuser, old_password, new_password,
+                                  'attacker.test')
+
+        assert_equal(response.status, 400)
+
+        # make sure that password is NOT changed
+        self._checkpw(testuser, old_password)
+
     def test_pwpolicy_error(self):
         response = self._changepw(testuser, old_password, '1')
 

ipatests/test_ipaserver/test_login_password.py

@@ -0,0 +1,88 @@
+# Copyright (C) 2023  Red Hat
+# see file 'COPYING' for use and warranty information
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# 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 General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import pytest
+import uuid
+
+from ipatests.test_ipaserver.httptest import Unauthorized_HTTP_test
+from ipatests.test_xmlrpc.xmlrpc_test import XMLRPC_test
+from ipatests.util import assert_equal
+from ipalib import api, errors
+from ipapython.ipautil import run
+
+testuser = u'tuser'
+password = u'password'
+
+
+@pytest.mark.tier1
+class test_login_password(XMLRPC_test, Unauthorized_HTTP_test):
+    app_uri = '/ipa/session/login_password'
+
+    @pytest.fixture(autouse=True)
+    def login_setup(self, request):
+        ccache = os.path.join('/tmp', str(uuid.uuid4()))
+        try:
+            api.Command['user_add'](uid=testuser, givenname=u'Test', sn=u'User')
+            api.Command['passwd'](testuser, password=password)
+            run(['kinit', testuser], stdin='{0}\n{0}\n{0}\n'.format(password),
+                env={"KRB5CCNAME": ccache})
+        except errors.ExecutionError as e:
+            pytest.skip(
+                'Cannot set up test user: %s' % e
+            )
+
+        def fin():
+            try:
+                api.Command['user_del']([testuser])
+            except errors.NotFound:
+                pass
+            os.unlink(ccache)
+
+        request.addfinalizer(fin)
+
+    def _login(self, user, password, host=None):
+        return self.send_request(params={'user': str(user),
+                                 'password' : str(password)},
+                                 host=host)
+
+    def test_bad_options(self):
+        for params in (
+            None,                             # no params
+            {"user": "foo"},                  # missing options
+            {"user": "foo", "password": ""},  # empty option
+        ):
+            response = self.send_request(params=params)
+            assert_equal(response.status, 400)
+            assert_equal(response.reason, 'Bad Request')
+
+    def test_invalid_auth(self):
+        response = self._login(testuser, 'wrongpassword')
+
+        assert_equal(response.status, 401)
+        assert_equal(response.getheader('X-IPA-Rejection-Reason'),
+                     'invalid-password')
+
+    def test_invalid_referer(self):
+        response = self._login(testuser, password, 'attacker.test')
+
+        assert_equal(response.status, 400)
+
+    def test_success(self):
+        response = self._login(testuser, password)
+
+        assert_equal(response.status, 200)
+        assert response.getheader('X-IPA-Rejection-Reason') is None

ipatests/test_ipaserver/test_referer.py

@@ -0,0 +1,136 @@
+# Copyright (C) 2023  Red Hat
+# see file 'COPYING' for use and warranty information
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# 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 General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import pytest
+import uuid
+
+from ipatests.test_ipaserver.httptest import Unauthorized_HTTP_test
+from ipatests.test_xmlrpc.xmlrpc_test import XMLRPC_test
+from ipatests.util import assert_equal
+from ipalib import api, errors
+from ipapython.ipautil import run
+
+testuser = u'tuser'
+password = u'password'
+
+
+@pytest.mark.tier1
+class test_referer(XMLRPC_test, Unauthorized_HTTP_test):
+
+    @pytest.fixture(autouse=True)
+    def login_setup(self, request):
+        ccache = os.path.join('/tmp', str(uuid.uuid4()))
+        tokenid = None
+        try:
+            api.Command['user_add'](uid=testuser, givenname=u'Test', sn=u'User')
+            api.Command['passwd'](testuser, password=password)
+            run(['kinit', testuser], stdin='{0}\n{0}\n{0}\n'.format(password),
+                env={"KRB5CCNAME": ccache})
+            result = api.Command["otptoken_add"](
+                type='HOTP', description='testotp',
+                ipatokenotpalgorithm='sha512', ipatokenowner=testuser,
+                ipatokenotpdigits='6')
+            tokenid = result['result']['ipatokenuniqueid'][0]
+        except errors.ExecutionError as e:
+            pytest.skip(
+                'Cannot set up test user: %s' % e
+            )
+
+        def fin():
+            try:
+                api.Command['user_del']([testuser])
+                api.Command['otptoken_del']([tokenid])
+            except errors.NotFound:
+                pass
+            os.unlink(ccache)
+
+        request.addfinalizer(fin)
+
+    def _request(self, params={}, host=None):
+        # implicit is that self.app_uri is set to the appropriate value
+        return self.send_request(params=params, host=host)
+
+    def test_login_password_valid(self):
+        """Valid authentication of a user"""
+        self.app_uri = "/ipa/session/login_password"
+        response = self._request(
+            params={'user': 'tuser', 'password': password})
+        assert_equal(response.status, 200, self.app_uri)
+
+    def test_change_password_valid(self):
+        """This actually changes the user password"""
+        self.app_uri = "/ipa/session/change_password"
+        response = self._request(
+            params={'user': 'tuser',
+                    'old_password': password,
+                    'new_password': 'new_password'}
+        )
+        assert_equal(response.status, 200, self.app_uri)
+
+    def test_sync_token_valid(self):
+        """We aren't testing that sync works, just that we can get there"""
+        self.app_uri = "/ipa/session/sync_token"
+        response = self._request(
+            params={'user': 'tuser',
+                    'first_code': '1234',
+                    'second_code': '5678',
+                    'password': 'password'})
+        assert_equal(response.status, 200, self.app_uri)
+
+    def test_i18n_messages_valid(self):
+        # i18n_messages requires a valid JSON request and we send
+        # nothing. If we get a 500 error then it got past the
+        # referer check.
+        self.app_uri = "/ipa/i18n_messages"
+        response = self._request()
+        assert_equal(response.status, 500, self.app_uri)
+
+    # /ipa/session/login_x509 is not tested yet as it requires
+    # significant additional setup.
+    # This can be manually verified by adding
+    # Satisfy Any and Require all granted to the configuration
+    # section and comment out all Auth directives. The request
+    # will fail and log that there is no KRB5CCNAME which comes
+    # after the referer check.
+
+    def test_endpoints_auth_required(self):
+        """Test endpoints that require pre-authorization which will
+           fail before we even get to the Referer check
+        """
+        self.endpoints = {
+            "/ipa/xml",
+            "/ipa/session/login_kerberos",
+            "/ipa/session/json",
+            "/ipa/session/xml"
+        }
+        for self.app_uri in self.endpoints:
+            response = self._request(host="attacker.test")
+
+            # referer is checked after auth
+            assert_equal(response.status, 401, self.app_uri)
+
+    def notest_endpoints_invalid(self):
+        """Pass in a bad Referer, expect a 400 Bad Request"""
+        self.endpoints = {
+            "/ipa/session/login_password",
+            "/ipa/session/change_password",
+            "/ipa/session/sync_token",
+        }
+        for self.app_uri in self.endpoints:
+            response = self._request(host="attacker.test")
+
+            assert_equal(response.status, 400, self.app_uri)

ipatests/util.py

@@ -163,12 +163,12 @@ class ExceptionNotRaised(Exception):
         return self.msg % self.expected.__name__
 
 
-def assert_equal(val1, val2):
+def assert_equal(val1, val2, msg=''):
     """
     Assert ``val1`` and ``val2`` are the same type and of equal value.
     """
     assert type(val1) is type(val2), '%r != %r' % (val1, val2)
-    assert val1 == val2, '%r != %r' % (val1, val2)
+    assert val1 == val2, '%r != %r %r' % (val1, val2, msg)
 
 
 def assert_not_equal(val1, val2):