.

checkov/common/checks_infra/solvers/connections_solvers/connection_exists_solver.py

@@ -54,15 +54,7 @@ def get_operation(
     def _get_operation(
         self, graph_connector: LibraryGraph
     ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]:
-        passed: List[Dict[str, Any]] = []
-        failed: List[Dict[str, Any]] = []
-        unknown: List[Dict[str, Any]] = []
-        if not self.vertices_under_resource_types or not self.vertices_under_connected_resources_types:
-            failed.extend(self.vertices_under_resource_types)
-            failed.extend(self.vertices_under_connected_resources_types)
-            return passed, failed, unknown
-
-        if isinstance(graph_connector, Graph):
+        def get_igraph_operation():
             for root_vertex in graph_connector.vs:
                 inverted = False
                 origin_attributes = None
@@ -92,7 +84,7 @@ def _get_operation(
                             unknown=unknown,
                         )
 
-        elif isinstance(graph_connector, DiGraph):
+        def get_networkx_operation():
             for u, v in edge_dfs(graph_connector):
                 origin_attributes = graph_connector.nodes(data=True)[u]
                 opposite_vertices = None
@@ -130,8 +122,7 @@ def _get_operation(
                     except StopIteration:
                         continue
 
-        # isinstance(graph_connector, PyDiGraph):
-        else:
+        def get_rustworkx_operation():
             for edge in iter(graph_connector.edge_list()):
                 u, v = edge
                 origin_attributes = graph_connector.nodes()[u][1]
@@ -168,6 +159,22 @@ def _get_operation(
                             passed.extend([origin_attributes, output_destination])
                     except StopIteration:
                         continue
+        passed: List[Dict[str, Any]] = []
+        failed: List[Dict[str, Any]] = []
+        unknown: List[Dict[str, Any]] = []
+        if not self.vertices_under_resource_types or not self.vertices_under_connected_resources_types:
+            failed.extend(self.vertices_under_resource_types)
+            failed.extend(self.vertices_under_connected_resources_types)
+            return passed, failed, unknown
+
+        if isinstance(graph_connector, Graph):
+            get_igraph_operation()
+
+        elif isinstance(graph_connector, DiGraph):
+            get_networkx_operation()
+
+        else:
+            get_rustworkx_operation()
 
         failed.extend(
             [

tests/terraform/runner/test_runner.py

@@ -12,13 +12,15 @@
 from igraph import Graph
 from networkx import DiGraph
 from parameterized import parameterized, parameterized_class
+from rustworkx import PyDiGraph
 
 from checkov.common.bridgecrew.check_type import CheckType
 from checkov.common.bridgecrew.severities import Severities, BcSeverities
 
 from checkov.common.checks_infra.registry import get_graph_checks_registry
 from checkov.common.graph.db_connectors.igraph.igraph_db_connector import IgraphConnector
 from checkov.common.graph.db_connectors.networkx.networkx_db_connector import NetworkxConnector
+from checkov.common.graph.db_connectors.rustworkx.rustworkx_db_connector import RustworkxConnector
 from checkov.common.graph.graph_builder import CustomAttributes
 from checkov.common.models.enums import CheckCategories, CheckResult
 from checkov.common.output.report import Report
@@ -46,6 +48,8 @@
     {"db_connector": NetworkxConnector, "tf_split_graph": "False", "graph": "NETWORKX"},
     {"db_connector": IgraphConnector, "tf_split_graph": "True", "graph": "IGRAPH"},
     {"db_connector": IgraphConnector, "tf_split_graph": "False", "graph": "IGRAPH"},
+    {"db_connector": RustworkxConnector, "tf_split_graph": "True", "graph": "RUSTWORKX"},
+    {"db_connector": RustworkxConnector, "tf_split_graph": "False", "graph": "RUSTWORKX"},
 ])
 class TestRunnerValid(unittest.TestCase):
     def setUp(self) -> None:
@@ -1593,7 +1597,8 @@ def tearDown(self):
 
     @parameterized.expand([
         (NetworkxConnector,),
-        (IgraphConnector,)
+        (IgraphConnector,),
+        (RustworkxConnector,)
     ])
     def test_get_graph_resource_entity_config(self, graph_connector):
         current_dir = os.path.dirname(os.path.realpath(__file__))
@@ -1610,6 +1615,10 @@ def test_get_graph_resource_entity_config(self, graph_connector):
             for data in graph_connector.vs.select()["attr"]:
                 config = Runner.get_graph_resource_entity_config(data)
                 self.assertIn(CustomAttributes.TF_RESOURCE_ADDRESS, config)
+        if isinstance(graph_connector, PyDiGraph):
+            for _, data in graph_connector.nodes():
+                config = Runner.get_graph_resource_entity_config(data)
+                self.assertIn(CustomAttributes.TF_RESOURCE_ADDRESS, config)
 
     @mock.patch.dict(os.environ, {"ENABLE_DEFINITION_KEY": "True"})
     def test_entity_context_fetching_with_TFDefinitionKey(self):