Fix Kerberos login failures returning a 500 JSON response, and first-time auto-created Kerberos logins being rejected. #10419 (#10455)

When SPNEGO negotiation failed, the exception object was passed to flash(), so the session could not be serialised and the login page was replaced by a 500 JSON response. A generic message is now flashed; the detail is already logged.

Also fix a redirect to a non-existent 'authenticate.kerberos_login' endpoint (it is 'kerberos.login'), which had caused every Kerberos mock test to skip unconditionally, and return the username from Kerberos auto-creation so a user's first login is not rejected with "Incorrect username or password".

Co-authored-by: Dave Page <dpage@pgadmin.org>

web/pgadmin/authenticate/__init__.py

@@ -209,7 +209,7 @@ def _login():
             if current_auth_obj['current_source'] == \
                     KERBEROS:
                 return redirect('{0}?next={1}'.format(url_for(
-                    'authenticate.kerberos_login'), url_for('browser.index')))
+                    'kerberos.login'), url_for('browser.index')))
 
             flash(msg, MessageType.ERROR)
             return redirect(pga_utils.get_safe_post_logout_redirect())

web/pgadmin/authenticate/kerberos.py

@@ -197,7 +197,8 @@ class KerberosAuthentication(BaseAuthentication):
                         retval = self.__auto_create_user(
                             str(negotiate.initiator_name))
                     elif isinstance(negotiate, Exception):
-                        flash(gettext(negotiate), MessageType.ERROR)
+                        flash(gettext("Kerberos authentication failed."),
+                              MessageType.ERROR)
                         retval = [status,
                                   Response(render_template(
                                       "security/login_user.html",
@@ -275,12 +276,16 @@ class KerberosAuthentication(BaseAuthentication):
                               "from auth source KERBEROS.")
                 current_app.logger.info(create_msg.format(username,
                                                           username))
-                return create_user({
+                status, msg = create_user({
                     'username': username,
                     'email': username,
                     'role': 2,
                     'active': True,
                     'auth_source': KERBEROS
                 })
+                if not status:
+                    return status, msg
 
+        # The caller needs the username to log the user in, including
+        # when the user has just been created.
         return True, {'username': username}

web/pgadmin/browser/tests/test_kerberos_with_mocking.py

@@ -38,6 +38,11 @@ class KerberosLoginMockTestCase(BaseTestGenerator):
             auth_source=[KERBEROS],
             auto_create_user=True,
             flag=3
+        )),
+        ('Spnego/Kerberos Authentication: Test Negotiate Failure', dict(
+            auth_source=[KERBEROS],
+            auto_create_user=True,
+            flag=4
         ))
     ]
 
@@ -68,7 +73,7 @@ class KerberosLoginMockTestCase(BaseTestGenerator):
         # path that redirects to /kerberos/login would 500 with
         # BuildError. Skip when that endpoint isn't reachable.
         endpoint_map = self.app.url_map._rules_by_endpoint
-        if 'authenticate.kerberos_login' not in endpoint_map:
+        if 'kerberos.login' not in endpoint_map:
             self.skipTest(
                 "Kerberos blueprint not loaded — set "
                 "AUTHENTICATION_SOURCES=['kerberos'] in config_local.py at "
@@ -97,6 +102,12 @@ class KerberosLoginMockTestCase(BaseTestGenerator):
                     "Can not run Kerberos Authentication in the Desktop mode."
                 )
             self.test_update_ticket()
+        elif self.flag == 4:
+            if app_config.SERVER_MODE is False:
+                self.skipTest(
+                    "Can not run Kerberos Authentication in the Desktop mode."
+                )
+            self.test_negotiate_failure_exception()
 
     def test_unauthorized(self):
         """
@@ -155,6 +166,29 @@ class KerberosLoginMockTestCase(BaseTestGenerator):
         self.assertEqual(response.status_code, 200)
         self.tester.logout()
 
+    def test_negotiate_failure_exception(self):
+        """
+        Ensure that when negotiate_start returns an Exception (e.g.
+        MissingCredentialsError), the server returns the HTML login page with
+        the error instead of crashing with a 500 JSON serialisation error.
+        """
+        class MissingCredentialsError(Exception):
+            pass
+
+        with patch.object(
+            AuthSourceRegistry._registry[KERBEROS], 'negotiate_start',
+            return_value=[False, MissingCredentialsError(
+                "No Kerberos credentials available")]
+        ):
+            res = self.tester.login(
+                None, None, True,
+                headers={'Authorization': 'Negotiate CTOKEN'})
+            self.assertEqual(res.status_code, 200)
+            self.assertIn('text/html', res.content_type)
+            self.assertIn('Kerberos authentication failed.',
+                          res.data.decode('utf-8'))
+            self.tester.logout()
+
     def tearDown(self):
         pass