added fixes for comments batch 1

Author: feyruzb

web/server/codechecker_server/session_manager.py

@@ -323,9 +323,9 @@ def __oauth_apply_templates(self):
                     if param == 'callback_url':
                         if not self.check_callback_url_format(
                                 provider_name, provider[param]):
-                            LOG.warning("Disabling OAuth "
-                                        f"provider {provider_name} "
-                                        "due to invalid callback URL format.")
+                            LOG.error("Disabling OAuth "
+                                      f"provider {provider_name} "
+                                      "due to invalid callback URL format.")
                             provider['enabled'] = False
                 except KeyError as e:
                     LOG.warning(f"Parameter {param} in OAuth provider "
@@ -389,8 +389,8 @@ def get_oauth_config(self, provider):
             self.__auth_config["method_oauth"]["providers"][provider][
                 "enabled"] = False
 
-            LOG.warning("OAuth configuration was set to default values. " +
-                        "Disabling oauth provider: %s", provider)
+            LOG.error("OAuth configuration was set to default values. " +
+                      "Disabling oauth provider: %s", provider)
 
         return self.__auth_config.get(
             'method_oauth', {}).get("providers", {}).get(provider, {})

web/tests/functional/authentication/oauth_server.py

@@ -1,10 +1,22 @@
-# pylint: disable=invalid-name
+#
+# -------------------------------------------------------------------------
+#
+#  Part of the CodeChecker project, under the Apache License v2.0 with
+#  LLVM Exceptions. See LICENSE for license information.
+#  SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+#
+# -------------------------------------------------------------------------
+"""
+A mock OAuth server that simulates the behavior of an OAuth provider.
+"""
 
-from http.server import BaseHTTPRequestHandler, HTTPServer
 import json
 import os
+
 from authlib.oauth2.rfc7636 import create_s256_code_challenge as hash_s256
 
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
 # Server config
 HOSTNAME = "0.0.0.0"
 SERVERPORT = int(os.getenv("PORT")) if os.getenv("PORT") else 3000
@@ -116,24 +128,10 @@ def login_tester(self):
                 params[key] = value
 
             if "username" in query_params:
+                # print(f"Login request with username: {params['username']}")
                 query = f"{params['username']}:{params['password']}"
                 query_result = self.users_by_data.get(query, None)
-                # csrf attack case
-                if params['username'] == "user_csrf":
-                    print("CSRF attack detected")
-                    state = "fake_state"
-                    code = query_result['code']
-                    code_challenge = params['code_challenge']
-                    # store code_challenge in the server
-                    self.code_challenges[code] = {
-                        "code_challenge": code_challenge,
-                        "code_challenge_method": params[
-                            'code_challenge_method']}
-
-                    return self.show_json({"code": code,
-                                           "state": state})
-                # normal case
-                elif query_result:
+                if query_result:
                     state = params['state']
                     code = query_result['code']
                     code_challenge = params['code_challenge']
@@ -149,7 +147,7 @@ def login_tester(self):
         except IndexError:
             return self.show_rejection("Invalid query parameters")
         except Exception as ex:
-            print(f"Error: {ex}")
+            print(f"Error in login_tester of OAuth mock server: {ex}")
             return self.show_rejection("Internal server error")
 
     def get_user(self):
@@ -204,13 +202,15 @@ def handle_user_token_request(self):
                 return self.show_rejection("Invalid code")
         return self.path
 
+    # pylint: disable=invalid-name
     def do_GET(self):
         if self.path.startswith("/login"):
             return self.login_tester()
         elif self.path.startswith("/get_user"):
             return self.get_user()
         return self.path
 
+    # pylint: disable=invalid-name
     def do_POST(self):
         if self.path.endswith("/token"):
             return self.handle_user_token_request()
@@ -220,8 +220,8 @@ def do_POST(self):
 
 webServer = HTTPServer((HOSTNAME, SERVERPORT), OauthServer)
 webServer.allow_reuse_address = True
-print(f"Server started http://{HOSTNAME}:{SERVERPORT}")
+# print(f"OAuth mock server started on http://{HOSTNAME}:{SERVERPORT}")
 
 webServer.serve_forever()
 webServer.server_close()
-print("Server stopped.")
+print(f"OAuth mock server stopped on http://{HOSTNAME}:{SERVERPORT}")

web/tests/functional/authentication/test_authentication.py

@@ -225,6 +225,11 @@ def try_login(self, provider, username, password):
         link = link.split('?')[0]
 
         code, state = data['code'], data['state']
+
+        # CSRF attack case
+        if username == "user_csrf":
+            state = "FAKESTATE"
+
         auth_string = f"{link}?code={code}&state={state}"
 
         # PKCE attack case
@@ -295,15 +300,28 @@ def test_oauth_create_link(self):
         Tests functionality of create_link method
         that checks if it creates unique links correctly.
         """
+
+        from urllib.parse import urlparse, parse_qs
+
         auth_client = env.setup_auth_client(
             self._test_workspace, session_token='_PROHIBIT')
+        session_factory = env.create_sqlalchemy_session(self._test_workspace)
 
+        # check 1
         link_github = auth_client.createLink("github")
-        link_google = auth_client.createLink("google")
-
+        parsed_query = parse_qs(urlparse(link_github).query)
+        state = parsed_query.get("state")[0]
+        result = env.validate_oauth_session(session_factory, state)
         self.assertIsNotNone(link_github,
                              "Authorization link for Github created empty")
+        self.assertTrue(result, "create link wasn't seccesfully executed")
 
+        # check 2
+        link_google = auth_client.createLink("google")
+        parsed_query = parse_qs(urlparse(link_google).query)
+        state = parsed_query.get("state")[0]
+        result = env.validate_oauth_session(session_factory, state)
+        self.assertTrue(result, "create link wasn't seccesfully executed")
         self.assertIsNotNone(link_google,
                              "Authorization link for Google created empty")