fixed Username:Password vulnerability of accepting illegal OAuth format

web/server/codechecker_server/api/authentication.py

@@ -96,37 +96,6 @@ def getAuthParameters(self):
     def getLoggedInUser(self):
         return self.__auth_session.user if self.__auth_session else ""
 
-    @timeit
-    def createLink(self, provider):
-        """
-        For creating a autehntication link for OAuth for specified provider
-        """
-        oauth_config = self.__manager.get_oauth_config(provider)
-        if not oauth_config.get('enabled'):
-            raise codechecker_api_shared.ttypes.RequestFailed(
-                codechecker_api_shared.ttypes.ErrorCode.AUTH_DENIED,
-                "OAuth authentication is not enabled.")
-
-        stored_state = generate_token()
-        client_id = oauth_config["oauth_client_id"]
-        client_secret = oauth_config["oauth_client_secret"]
-        scope = oauth_config["oauth_scope"]
-        authorization_uri = oauth_config["oauth_authorization_uri"]
-        redirect_uri = oauth_config["oauth_redirect_uri"]
-
-        # Create an OAuth2Session instance
-        session = OAuth2Session(
-            client_id,
-            client_secret,
-            scope=scope,
-            redirect_uri=redirect_uri)
-
-        # Create authorization URL
-        nonce = generate_token()
-        url = session.create_authorization_url(
-            authorization_uri, nonce=nonce, state=stored_state)[0]
-        return url
-
     @timeit
     def getAcceptedAuthMethods(self):
         return ["Username:Password", "oauth"]
@@ -181,6 +150,37 @@ def getAccessControl(self):
     def getOauthProviders(self):
         return self.__manager.get_oauth_providers()
 
+    @timeit
+    def createLink(self, provider):
+        """
+        For creating a autehntication link for OAuth for specified provider
+        """
+        oauth_config = self.__manager.get_oauth_config(provider)
+        if not oauth_config.get('enabled'):
+            raise codechecker_api_shared.ttypes.RequestFailed(
+                codechecker_api_shared.ttypes.ErrorCode.AUTH_DENIED,
+                "OAuth authentication is not enabled.")
+
+        stored_state = generate_token()
+        client_id = oauth_config["oauth_client_id"]
+        client_secret = oauth_config["oauth_client_secret"]
+        scope = oauth_config["oauth_scope"]
+        authorization_uri = oauth_config["oauth_authorization_uri"]
+        redirect_uri = oauth_config["oauth_redirect_uri"]
+
+        # Create an OAuth2Session instance
+        session = OAuth2Session(
+            client_id,
+            client_secret,
+            scope=scope,
+            redirect_uri=redirect_uri)
+
+        # Create authorization URL
+        nonce = generate_token()
+        url = session.create_authorization_url(
+            authorization_uri, nonce=nonce, state=stored_state)[0]
+        return url
+
     @timeit
     def performLogin(self, auth_method, auth_string):
         if not auth_string:
@@ -269,8 +269,8 @@ def performLogin(self, auth_method, auth_string):
 
             if allowed_users == ["*"] or username in allowed_users:
                 LOG.info("User %s is authorized.", username)
-                session = self.__manager.create_session(
-                    f"{provider}@{username}:{token['access_token']}")
+                session = self.__manager.create_session_oauth(
+                    provider, username, token['access_token'])
                 return session.token
 
             if len(allowed_users) == 0:

web/server/codechecker_server/session_manager.py

@@ -398,8 +398,7 @@ def __handle_validation(self, auth_string):
         validation = self.__try_auth_root(auth_string) \
             or self.__try_auth_dictionary(auth_string) \
             or self.__try_auth_pam(auth_string) \
-            or self.__try_auth_ldap(auth_string) \
-            or self.__try_auth_oauth(auth_string)
+            or self.__try_auth_ldap(auth_string)
         if not validation:
             return False
 
@@ -520,28 +519,6 @@ def __try_auth_ldap(self, auth_string):
 
         return False
 
-    def __try_auth_oauth(self, auth_string):
-        """
-        Try to authenticate user based on the OAuth configuration.
-        """
-        if not self.__is_method_enabled('oauth') \
-                or "@" not in auth_string:
-            return False
-
-        providers = self.__auth_config.get(
-            'method_oauth', {}).get("providers", {})
-
-        provider, data = auth_string.split('@', 1)
-
-        if provider in providers:
-            if not providers.get(provider).get('enabled'):
-                return False
-
-            username, token = data.split(':')
-            return {'username': username, 'token': token, 'groups': []}
-
-        return False
-
     def __update_groups(self, user_name, groups):
         """
         Updates group field of the users tokens.
@@ -704,6 +681,54 @@ def create_session(self, auth_string):
 
         return local_session
 
+    def create_session_oauth(self, provider, username, token):
+        """ Creates a new session for the given auth-string. """
+
+        if not self.__is_method_enabled('oauth'):
+            return False
+
+        providers = self.__auth_config.get(
+            'method_oauth', {}).get("providers", {})
+
+        if provider not in providers or \
+                not providers.get(provider).get('enabled'):
+            return False
+
+        # To be parsed later
+        user_data = {'username': username,
+                     'token': token,
+                     'groups': [],
+                     'is_root': False}
+
+        # Generate a new token and create a local session.
+        token = generate_session_token()
+        local_session = self.__create_local_session(token,
+                                                    user_data.get('username'),
+                                                    user_data.get('groups'),
+                                                    user_data.get('is_root'))
+        self.__sessions.append(local_session)
+
+        # Store the session in the database.
+        transaction = None
+        if self.__database_connection:
+            try:
+                transaction = self.__database_connection()
+
+                record = SessionRecord(token,
+                                       user_data.get('username'),
+                                       ';'.join(user_data.get('groups')))
+                transaction.add(record)
+                transaction.commit()
+            except Exception as e:
+                LOG.error("Couldn't store or update login record in "
+                          "database:")
+                LOG.error(str(e))
+            finally:
+                if transaction:
+                    transaction.close()
+
+        return local_session
+
     def get_max_run_count(self):
         """
         Returns the maximum storable run count. If the value is None it means