CVE-2024-10081

@@ -74,6 +74,7 @@ service codeCheckerProductService {
 
   // Get the list of product that matches the display name and endpoint
   // filters specified.
+  // PERMISSION: PRODUCT_VIEW
   Products getProducts(1: string productEndpointFilter,
                        2: string productNameFilter)
                        throws (1: codechecker_api_shared.RequestFailed requestError),

@@ -688,7 +688,7 @@ service codeCheckerDBAccess {
                                                           throws (1: codechecker_api_shared.RequestFailed requestError),
 
   // Return true if review status change is disabled.
-  // PERMISSION: PRODUCT_ACCESS or PRODUCT_STORE
+  // PERMISSION: PRODUCT_VIEW
   bool isReviewStatusChangeDisabled()
                                     throws (1: codechecker_api_shared.RequestFailed requestError),
 
@@ -769,6 +769,7 @@ service codeCheckerDBAccess {
   // get the md documentation for a checker
   // DEPRECATED. Use getCheckerLabels() instead which contains checker
   // documentation URL.
+  // PERMISSION: PRODUCT_VIEW
   string getCheckerDoc(1: string checkerId)
                        throws (1: codechecker_api_shared.RequestFailed requestError),
 

@@ -31,14 +31,23 @@ class ThriftConfigHandler:
     Manages Thrift requests regarding configuration.
     """
 
-    def __init__(self, auth_session, config_session):
+    def __init__(self, auth_session, config_session, session_manager):
         self.__auth_session = auth_session
         self.__session = config_session
+        self.__session_manager = session_manager
 
     def __require_supermission(self):
         """
         Checks if the current user isn't a SUPERUSER.
         """
+
+        # Anonymous access is only allowed if authentication is
+        # turned off
+        if self.__session_manager.is_enabled and not self.__auth_session:
+            raise codechecker_api_shared.ttypes.RequestFailed(
+                codechecker_api_shared.ttypes.ErrorCode.UNAUTHORIZED,
+                "You are not authorized to execute this action.")
+
         if (not (self.__auth_session is None) and
                 not self.__auth_session.is_root):
             raise codechecker_api_shared.ttypes.RequestFailed(
@@ -69,7 +78,7 @@ def getNotificationBannerText(self):
     def setNotificationBannerText(self, notification_b64):
         """
         Sets the notification banner remove_products_except.
-        Bevare: This method only works if the use is a SUPERUSER.
+        Beware: This method only works if the use is a SUPERUSER.
         """
 
         self.__require_supermission()

@@ -69,6 +69,13 @@ def __require_permission(self, required, args=None):
                 args = dict(self.__permission_args)
                 args['config_db_session'] = session
 
+            # Anonymous access is only allowed if authentication is
+            # turned off
+            if self.__server.manager.is_enabled and not self.__auth_session:
+                raise codechecker_api_shared.ttypes.RequestFailed(
+                    codechecker_api_shared.ttypes.ErrorCode.UNAUTHORIZED,
+                    "You are not authorized to execute this action.")
+
             if not any(permissions.require_permission(
                            perm, args, self.__auth_session)
                        for perm in required):
@@ -247,6 +254,7 @@ def getProductConfiguration(self, product_id):
         Get the product configuration --- WITHOUT THE DB PASSWORD --- of the
         given product.
         """
+        self.__require_permission([permissions.PRODUCT_VIEW])
 
         with DBSession(self.__session) as session:
             product = session.query(Product).get(product_id)

@@ -1447,6 +1447,13 @@ def __require_permission(self, required):
             args = dict(self.__permission_args)
             args['config_db_session'] = session
 
+            # Anonymous access is only allowed if authentication is
+            # turned off
+            if self._manager.is_enabled and not self._auth_session:
+                raise codechecker_api_shared.ttypes.RequestFailed(
+                    codechecker_api_shared.ttypes.ErrorCode.UNAUTHORIZED,
+                    "You are not authorized to execute this action.")
+
             if not any(permissions.require_permission(
                     perm, args, self._auth_session)
                     for perm in required):
@@ -2320,6 +2327,7 @@ def _setReviewStatus(self, session, report_hash, status,
         database transaction. This is needed because during storage a specific
         session object has to be used.
         """
+
         review_status = session.query(ReviewStatus).get(report_hash)
         if review_status is None:
             review_status = ReviewStatus()
@@ -2421,6 +2429,8 @@ def isReviewStatusChangeDisabled(self):
         """
         Return True if review status change is disabled.
         """
+        self.__require_view()
+
         with DBSession(self._config_database) as session:
             product = session.query(Product).get(self._product.id)
             return product.is_review_status_change_disabled
@@ -2746,7 +2756,7 @@ def getCheckerDoc(self, _):
         Parameters:
          - checkerId
         """
-
+        self.__require_view()
         return ""
 
     @exc_to_thrift_reqfail
@@ -2756,6 +2766,8 @@ def getCheckerLabels(
         checkers: List[ttypes.Checker]
     ) -> List[List[str]]:
         """ Return the list of labels to each checker. """
+        self.__require_view()
+
         labels = []
         for checker in checkers:
             analyzer_name = None if not checker.analyzerName \
@@ -3569,6 +3581,8 @@ def getFailedFilesCount(self, run_ids):
         given run. If the run id list is empty the number of failed files will
         be counted for all of the runs.
         """
+        self.__require_view()
+
         # Unfortunately we can't distinct the failed file paths by using SQL
         # queries because the list of failed files for a run / analyzer are
         # stored in one column in a compressed way. For this reason we need to
@@ -3611,6 +3625,8 @@ def getFailedFiles(self, run_ids):
     # -----------------------------------------------------------------------
     @timeit
     def getPackageVersion(self):
+        self.__require_view()
+
         return self.__package_version
 
     # -----------------------------------------------------------------------

@@ -71,6 +71,8 @@ def is_supported_version(version):
 
     version = version.lstrip('v')
     version_parts = version.split('.')
+    if len(version_parts) < 2:
+        return False
 
     # We don't care if accidentally the version tag contains a revision number.
     major, minor = int(version_parts[0]), int(version_parts[1])
@@ -115,24 +117,27 @@ def split_client_POST_request(path):
     """
 
     # A standard POST request from an API client looks like:
-    # http://localhost:8001/[product-name]/<API version>/<API service>
+    # http://localhost:8001/[product-name/]<API version>/<API service>
     # where specifying the product name is optional.
 
     split_path = urlparse(path).path.split('/', 3)
 
     endpoint_part = split_path[1]
-    if is_valid_product_endpoint(split_path[1]):
+    if is_valid_product_endpoint(split_path[1]) and len(split_path) == 4:
         version_tag = split_path[2].lstrip('v')
-        remainder = split_path[3]
+        if not is_supported_version(version_tag):
+            return None, None, None
+        endpoint = split_path[3]
+        return endpoint_part, version_tag, endpoint
 
-        return endpoint_part, version_tag, remainder
-    elif split_path[1].startswith('v'):
+    elif split_path[1].startswith('v') and len(split_path) == 3:
         # Request came through without a valid product URL endpoint to
         # possibly the main server.
         version_tag = split_path[1].lstrip('v')
-        remainder = split_path[2]
-
-        return None, version_tag, remainder
+        if not is_supported_version(version_tag):
+            return None, None, None
+        endpoint = split_path[2]
+        return None, version_tag, endpoint
 
     return None, None, None
 

@@ -328,10 +328,21 @@ def do_POST(self):
         otrans = TTransport.TMemoryBuffer()
         oprot = output_protocol_factory.getProtocol(otrans)
 
+        product_endpoint, api_ver, request_endpoint = \
+            routing.split_client_POST_request(self.path)
+
+        if product_endpoint is None and api_ver is None and \
+                request_endpoint is None:
+            self.send_thrift_exception("Invalid request endpoint path.", iprot,
+                                       oprot, otrans)
+            return
+
+        # Only Authentication, Configuration, ServerInof
+        # endpoints are allowed for Anonymous users
+        # if authentication is required.
         if self.server.manager.is_enabled and \
-                not self.path.endswith(('/Authentication',
-                                        '/Configuration',
-                                        '/ServerInfo')) and \
+                request_endpoint not in \
+                ['Authentication', 'Configuration', 'ServerInfo'] and \
                 not self.auth_session:
             # Bail out if the user is not authenticated...
             # This response has the possibility of melting down Thrift clients,
@@ -347,12 +358,6 @@ def do_POST(self):
 
         # Authentication is handled, we may now respond to the user.
         try:
-            product_endpoint, api_ver, request_endpoint = \
-                routing.split_client_POST_request(self.path)
-            if product_endpoint is None and api_ver is None and \
-                    request_endpoint is None:
-                raise ValueError("Invalid request endpoint path.")
-
             product = None
             if product_endpoint:
                 # The current request came through a product route, and not
@@ -373,7 +378,8 @@ def do_POST(self):
                     elif request_endpoint == 'Configuration':
                         conf_handler = ConfigHandler_v6(
                             self.auth_session,
-                            self.server.config_session)
+                            self.server.config_session,
+                            self.server.manager)
                         processor = ConfigAPI_v6.Processor(conf_handler)
                     elif request_endpoint == 'ServerInfo':
                         server_info_handler = ServerInfoHandler_v6(version)

@@ -52,16 +52,12 @@ def test_post(self):
         # It is the server code's responsibility to give a 404 Not Found.
         self.assertEqual(post(''), (None, None, None))
         self.assertEqual(post('CodeCheckerService'), (None, None, None))
-
-        # Raise an exception if URL is malformed, such as contains a
-        # product-endpoint-like component which is badly encoded version
-        # string.
-        with self.assertRaises(Exception):
-            post('v6.0')
-            post('/v6/CodeCheckerService')
+        self.assertEqual(post('v6.0'), (None, None, None))
+        self.assertEqual(post('/v6.0/product/Authentication/Service'),
+                         (None, None, None))
 
         self.assertEqual(post('/v6.0/Authentication'),
                          (None, '6.0', 'Authentication'))
 
-        self.assertEqual(post('/DummyProduct/v0.0/FoobarService'),
-                         ('DummyProduct', '0.0', 'FoobarService'))
+        self.assertEqual(post('/DummyProduct/v6.0/FoobarService'),
+                         ('DummyProduct', '6.0', 'FoobarService'))

@@ -150,7 +150,7 @@ def test_get_product_data(self):
 
         # Now get the SERVERSPACE (configuration) for the product.
         # TODO: These things usually should only work for superusers!
-        pr_conf = self._pr_client.getProductConfiguration(pr_data.id)
+        pr_conf = self._root_client.getProductConfiguration(pr_data.id)
 
         self.assertIsNotNone(pr_conf, "Product configuration must come.")
         self.assertEqual(pr_conf.endpoint, self.product_name,
@@ -189,7 +189,7 @@ def test_editing(self):
         pr_client = env.setup_product_client(
             self.test_workspace, product=self.product_name)
         product_id = pr_client.getCurrentProduct().id
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
 
         old_name = config.displayedName_b64
 
@@ -202,7 +202,7 @@ def test_editing(self):
         self.assertTrue(self._root_client.editProduct(product_id, config),
                         "Product edit didn't conclude.")
 
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
         self.assertEqual(config.endpoint, self.product_name,
                          "The product edit changed the endpoint, when it "
                          "shouldn't have!")
@@ -214,7 +214,7 @@ def test_editing(self):
         self.assertTrue(self._root_client.editProduct(product_id, config),
                         "Product config restore didn't conclude.")
 
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
         self.assertEqual(config.displayedName_b64, old_name,
                          "The product edit didn't change the name back.")
 
@@ -225,7 +225,7 @@ def test_editing(self):
         self.assertTrue(self._root_client.editProduct(product_id, config),
                         "Product edit didn't conclude.")
 
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
         self.assertEqual(config.confidentiality,
                          new_confidentiality,
                          "Couldn't change the confidentiality to OPEN")
@@ -235,7 +235,7 @@ def test_editing(self):
         self.assertTrue(self._root_client.editProduct(product_id, config),
                         "Product edit didn't conclude.")
 
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
         self.assertEqual(config.confidentiality,
                          new_confidentiality,
                          "Couldn't change the confidentiality to INTERNAL")
@@ -245,7 +245,7 @@ def test_editing(self):
         self.assertTrue(self._root_client.editProduct(product_id, config),
                         "Product edit didn't conclude.")
 
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
         self.assertEqual(config.confidentiality,
                          new_confidentiality,
                          "Couldn't change the confidentiality to CONFIDENTIAL")
@@ -255,7 +255,7 @@ def test_editing(self):
         self.assertTrue(self._root_client.editProduct(product_id, config),
                         "Product config restore didn't conclude.")
 
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
         self.assertEqual(config.confidentiality,
                          old_confidentiality,
                          "The edit didn't change back the confidentiality.")
@@ -271,7 +271,7 @@ def test_editing_reconnect(self):
         pr_client = env.setup_product_client(
             self.test_workspace, product=self.product_name)
         product_id = pr_client.getCurrentProduct().id
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
 
         old_db_name = config.connection.database
 
@@ -292,7 +292,7 @@ def test_editing_reconnect(self):
                         "Product edit didn't conclude.")
 
         # Check if the configuration now uses the new values.
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
 
         self.assertEqual(config.connection.database, new_db_name,
                          "Server didn't save new database name.")
@@ -311,7 +311,7 @@ def test_editing_reconnect(self):
         self.assertTrue(self._root_client.editProduct(product_id, config),
                         "Product configuration restore didn't conclude.")
 
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
         self.assertEqual(config.connection.database, old_db_name,
                          "Server didn't save back to old database name.")
 
@@ -336,7 +336,7 @@ def test_editing_endpoint(self):
         pr_client = env.setup_product_client(
             self.test_workspace, product=self.product_name)
         product_id = pr_client.getCurrentProduct().id
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
 
         old_endpoint = config.endpoint
         new_endpoint = "edited_endpoint"
@@ -347,7 +347,7 @@ def test_editing_endpoint(self):
                         "Product edit didn't conclude.")
 
         # Check if the configuration now uses the new values.
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
         self.assertEqual(config.endpoint, new_endpoint,
                          "Server didn't save new endpoint.")
 
@@ -372,7 +372,7 @@ def test_editing_endpoint(self):
         self.assertTrue(self._root_client.editProduct(product_id, config),
                         "Product configuration restore didn't conclude.")
 
-        config = self._pr_client.getProductConfiguration(product_id)
+        config = self._root_client.getProductConfiguration(product_id)
         self.assertEqual(config.endpoint, old_endpoint,
                          "Server didn't save back to old endpoint.")
 

@@ -104,23 +104,20 @@ def setup_method(self, _):
 
     def test_noauth_notification_edit(self):
         """
-        Test for editing the notification text on a non authenting server.
+        Test for editing the notification text on a non-authenticating user
+        on an authenticating server
         """
 
         # A non-authenticated session should return an empty user.
         user = self.auth_client.getLoggedInUser()
         self.assertEqual(user, "")
 
-        # Server without authentication should allow notification setting.
-        self.config_client.setNotificationBannerText(
-            convert.to_b64('noAuth notif'))
-        self.assertEqual(convert.from_b64(
-            self.config_client.getNotificationBannerText()), 'noAuth notif')
+        # Anonymous user should not be allowed to change the banner
+        with self.assertRaises(RequestFailed):
+            self.config_client.setNotificationBannerText(
+                    convert.to_b64('non su notification'))
 
-    def test_auth_su_notification_edit(self):
-        """
-        Test that SUPERADMINS can edit the notification text.
-        """
+    def get_su_config_client(self):
         # Create a SUPERUSER login.
         self.session_token = self.auth_client.performLogin(
             "Username:Password", "root:root")
@@ -142,8 +139,14 @@ def test_auth_su_notification_edit(self):
 
         user = su_auth_client.getLoggedInUser()
         self.assertEqual(user, "root")
-        # we are root
+        return su_config_client
 
+    def test_auth_su_notification_edit(self):
+        """
+        Test that SUPERADMINS can edit the notification text.
+        """
+
+        su_config_client = self.get_su_config_client()
         su_config_client.setNotificationBannerText(
                 convert.to_b64('su notification'))
         self.assertEqual(convert.from_b64(
@@ -172,21 +175,18 @@ def test_auth_non_su_notification_edit(self):
             authd_config_client.setNotificationBannerText(
                     convert.to_b64('non su notification'))
 
-            print("You are not authorized to modify notifications!")
+        print("You are not authorized to modify notifications!")
 
     def test_unicode_string(self):
         """
         Test for non ascii strings. Needed because the used Thrift
         version won't eat them.
         """
-
-        # A non-authenticated session should return an empty user.
-        user = self.auth_client.getLoggedInUser()
-        self.assertEqual(user, "")
+        su_config_client = self.get_su_config_client()
 
         # Check if utf-8 encoded strings are okay.
-        self.config_client.setNotificationBannerText(
+        su_config_client.setNotificationBannerText(
                 convert.to_b64('árvíztűrő tükörfúrógép'))
         self.assertEqual(convert.from_b64(
-            self.config_client.getNotificationBannerText()),
+             su_config_client.getNotificationBannerText()),
             'árvíztűrő tükörfúrógép')