fix(security): replace yaml.full_load with yaml.safe_load to prevent RCE (#1242)

Signed-off-by: Nitesh <nitesh@example.com>
Co-authored-by: Nitesh <nitesh@example.com>

krkn/scenario_plugins/application_outage/application_outage_scenario_plugin.py

@@ -34,7 +34,7 @@ class ApplicationOutageScenarioPlugin(AbstractScenarioPlugin):
     ) -> int:
         try:
             with open(scenario, "r") as f:
-                app_outage_config_yaml = yaml.full_load(f)
+                app_outage_config_yaml = yaml.safe_load(f)
                 scenario_config = app_outage_config_yaml["application_outage"]
                 pod_selector = get_yaml_item_value(
                     scenario_config, "pod_selector", "{}"

krkn/scenario_plugins/container/container_scenario_plugin.py

@@ -36,7 +36,7 @@ class ContainerScenarioPlugin(AbstractScenarioPlugin):
     ) -> int:
         try:
             with open(scenario, "r") as f:
-                cont_scenario_config = yaml.full_load(f)
+                cont_scenario_config = yaml.safe_load(f)
                 
                 for kill_scenario in cont_scenario_config["scenarios"]:
                     future_snapshot = self.start_monitoring(

krkn/scenario_plugins/hogs/hogs_scenario_plugin.py

@@ -40,7 +40,7 @@ class HogsScenarioPlugin(AbstractScenarioPlugin):
             scenario_telemetry: ScenarioTelemetry) -> int:
         try:
             with open(scenario, "r") as f:
-                scenario = yaml.full_load(f)
+                scenario = yaml.safe_load(f)
             scenario_config = HogConfig.from_yaml_dict(scenario)
             
             # Get node-name if provided

krkn/scenario_plugins/kubevirt_vm_outage/kubevirt_vm_outage_scenario_plugin.py

@@ -56,7 +56,7 @@ class KubevirtVmOutageScenarioPlugin(AbstractScenarioPlugin):
         """
         try:
             with open(scenario, "r") as f:
-                scenario_config = yaml.full_load(f)
+                scenario_config = yaml.safe_load(f)
             
             self.init_clients(lib_telemetry.get_lib_kubernetes())
             vmis_status = VmisStatus()

krkn/scenario_plugins/managed_cluster/managed_cluster_scenario_plugin.py

@@ -33,7 +33,7 @@ class ManagedClusterScenarioPlugin(AbstractScenarioPlugin):
         scenario_telemetry: ScenarioTelemetry,
     ) -> int:
         with open(scenario, "r") as f:
-            scenario = yaml.full_load(f)
+            scenario = yaml.safe_load(f)
             for managedcluster_scenario in scenario["managedcluster_scenarios"]:
                 managedcluster_scenario_object = Scenarios(
                     lib_telemetry.get_lib_kubernetes()

krkn/scenario_plugins/node_actions/node_actions_scenario_plugin.py

@@ -57,7 +57,7 @@ class NodeActionsScenarioPlugin(AbstractScenarioPlugin):
         scenario_telemetry: ScenarioTelemetry,
     ) -> int:
         with open(scenario, "r") as f:
-            node_scenario_config = yaml.full_load(f)
+            node_scenario_config = yaml.safe_load(f)
             for node_scenario in node_scenario_config["node_scenarios"]:
                 try:
                     node_scenario_object = self.get_node_scenario_object(

krkn/scenario_plugins/pod_disruption/pod_disruption_scenario_plugin.py

@@ -46,7 +46,7 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin):
     ) -> int:
         try:
             with open(scenario, "r") as f:
-                cont_scenario_config = yaml.full_load(f)
+                cont_scenario_config = yaml.safe_load(f)
                 for kill_scenario in cont_scenario_config:
                     kill_scenario_config = InputParams(kill_scenario["config"])
                     future_snapshot=self.start_monitoring(

krkn/scenario_plugins/pvc/pvc_scenario_plugin.py

@@ -40,7 +40,7 @@ class PvcScenarioPlugin(AbstractScenarioPlugin):
     ) -> int:
         try:
             with open(scenario, "r") as f:
-                config_yaml = yaml.full_load(f)
+                config_yaml = yaml.safe_load(f)
                 scenario_config = config_yaml["pvc_scenario"]
                 pvc_name = get_yaml_item_value(scenario_config, "pvc_name", "")
                 pod_name = get_yaml_item_value(scenario_config, "pod_name", "")

krkn/scenario_plugins/service_disruption/service_disruption_scenario_plugin.py

@@ -34,7 +34,7 @@ class ServiceDisruptionScenarioPlugin(AbstractScenarioPlugin):
     ) -> int:
         try:
             with open(scenario, "r") as f:
-                scenario_config_yaml = yaml.full_load(f)
+                scenario_config_yaml = yaml.safe_load(f)
                 for scenario in scenario_config_yaml["scenarios"]:
                     scenario_namespace = get_yaml_item_value(scenario, "namespace", "")
                     scenario_label = get_yaml_item_value(scenario, "label_selector", "")

krkn/scenario_plugins/shut_down/shut_down_scenario_plugin.py

@@ -41,7 +41,7 @@ class ShutDownScenarioPlugin(AbstractScenarioPlugin):
     ) -> int:
         try:
             with open(scenario, "r") as f:
-                shut_down_config_yaml = yaml.full_load(f)
+                shut_down_config_yaml = yaml.safe_load(f)
                 shut_down_config_scenario = shut_down_config_yaml[
                     "cluster_shut_down_scenario"
                 ]

krkn/scenario_plugins/time_actions/time_actions_scenario_plugin.py

@@ -37,7 +37,7 @@ class TimeActionsScenarioPlugin(AbstractScenarioPlugin):
     ) -> int:
         try:
             with open(scenario, "r") as f:
-                scenario_config = yaml.full_load(f)
+                scenario_config = yaml.safe_load(f)
                 for time_scenario in scenario_config["time_scenarios"]:
                     object_type, object_names = self.skew_time(
                         time_scenario, lib_telemetry.get_lib_kubernetes()

krkn/scenario_plugins/zone_outage/zone_outage_scenario_plugin.py

@@ -46,7 +46,7 @@ class ZoneOutageScenarioPlugin(AbstractScenarioPlugin):
     ) -> int:
         try:
             with open(scenario, "r") as f:
-                zone_outage_config_yaml = yaml.full_load(f)
+                zone_outage_config_yaml = yaml.safe_load(f)
                 scenario_config = zone_outage_config_yaml["zone_outage"]
                 cloud_type = scenario_config["cloud_type"]
                 kube_check = get_yaml_item_value(scenario_config, "kube_check", True)

run_kraken.py

@@ -80,7 +80,7 @@ def main(options, command: Optional[str]) -> int:
     # Parse and read the config
     if os.path.isfile(cfg):
         with open(cfg, "r") as f:
-            config = yaml.full_load(f)
+            config = yaml.safe_load(f)
         kubeconfig_path = os.path.expanduser(
             get_yaml_item_value(config["kraken"], "kubeconfig_path", "")
         )

tests/test_managed_cluster_scenario_plugin.py

@@ -38,7 +38,7 @@ class TestManagedClusterScenarioPlugin(unittest.TestCase):
 
     @patch('time.time')
     @patch('builtins.open', create=True)
-    @patch('yaml.full_load')
+    @patch('yaml.safe_load')
     @patch('krkn.cerberus.get_status')
     def test_run_multiple_actions_executes_all(self, mock_cerberus, mock_yaml, mock_open, mock_time):
         """
@@ -94,7 +94,7 @@ class TestManagedClusterScenarioPlugin(unittest.TestCase):
 
     @patch('time.time')
     @patch('builtins.open', create=True)
-    @patch('yaml.full_load')
+    @patch('yaml.safe_load')
     def test_run_stops_on_first_error(self, mock_yaml, mock_open, mock_time):
         """
         Test that run() returns 1 and stops executing on first error
@@ -146,7 +146,7 @@ class TestManagedClusterScenarioPlugin(unittest.TestCase):
 
 
     @patch('builtins.open', create=True)
-    @patch('yaml.full_load')
+    @patch('yaml.safe_load')
     def test_run_returns_error_when_actions_empty(self, mock_yaml, _mock_open):
         """
         Test that run() returns 1 and logs an error when actions is an empty list
@@ -182,7 +182,7 @@ class TestManagedClusterScenarioPlugin(unittest.TestCase):
         )
 
     @patch('builtins.open', create=True)
-    @patch('yaml.full_load')
+    @patch('yaml.safe_load')
     def test_run_returns_error_when_actions_none(self, mock_yaml, _mock_open):
         """
         Test that run() returns 1 and logs an error when actions is None

tests/test_node_actions_scenario_plugin.py

@@ -667,7 +667,7 @@ class TestNodeActionsScenarioPlugin(unittest.TestCase):
         mock_common_funcs.get_node_by_name.return_value = ["test-node"]
         mock_cerberus.get_status.return_value = None
 
-        with patch('yaml.full_load', return_value=scenario_yaml):
+        with patch('yaml.safe_load', return_value=scenario_yaml):
             result = self.plugin.run(
                 "test-uuid",
                 "/path/to/scenario.yaml",
@@ -692,7 +692,7 @@ class TestNodeActionsScenarioPlugin(unittest.TestCase):
             ]
         }
 
-        with patch('yaml.full_load', return_value=scenario_yaml):
+        with patch('yaml.safe_load', return_value=scenario_yaml):
             result = self.plugin.run(
                 "test-uuid",
                 "/path/to/scenario.yaml",

tests/test_shut_down_scenario_plugin.py

@@ -61,7 +61,7 @@ class TestShutDownScenarioPlugin(unittest.TestCase):
         mock_time.side_effect = [1000, 2000]
         self.mock_kubecli.list_nodes.return_value = ["node1", "node2"]
 
-        with patch('yaml.full_load', return_value=scenario_yaml):
+        with patch('yaml.safe_load', return_value=scenario_yaml):
             with patch.object(self.plugin, 'cluster_shut_down') as mock_cluster_shutdown:
                 result = self.plugin.run(
                     "test-uuid",

tests/test_time_actions_scenario_plugin.py

@@ -55,7 +55,7 @@ class TestTimeActionsScenarioPlugin(unittest.TestCase):
         self.assertIn("disk quota exceeded", logged_msg)
         self.assertNotIn("NameError", logged_msg)
     @unittest.mock.patch('builtins.open', create=True)
-    @unittest.mock.patch('yaml.full_load')
+    @unittest.mock.patch('yaml.safe_load')
     @unittest.mock.patch('logging.error')
     def test_run_exception_handling_with_variable(self, mock_logging_error, mock_yaml, mock_open):
         """
@@ -86,7 +86,7 @@ class TestTimeActionsScenarioPlugin(unittest.TestCase):
         self.assertIn("TimeActionsScenarioPlugin", error_call_args)
 
     @unittest.mock.patch('builtins.open', create=True)
-    @unittest.mock.patch('yaml.full_load')
+    @unittest.mock.patch('yaml.safe_load')
     def test_run_with_skew_time_exception(self, mock_yaml, mock_open):
         """
         Test that run() handles exceptions from skew_time method

tests/test_zone_outage_scenario_plugin.py

@@ -55,7 +55,7 @@ class TestZoneOutageScenarioPlugin(unittest.TestCase):
         self.assertEqual(len(result), 1)
 
     @unittest.mock.patch('builtins.open', create=True)
-    @unittest.mock.patch('yaml.full_load')
+    @unittest.mock.patch('yaml.safe_load')
     @unittest.mock.patch('krkn.scenario_plugins.zone_outage.zone_outage_scenario_plugin.gcp_node_scenarios')
     @unittest.mock.patch('krkn.cerberus.publish_kraken_status')
     def test_run_propagates_node_based_zone_failure(self, mock_cerberus, mock_gcp_scenarios, mock_yaml, mock_open):
@@ -102,7 +102,7 @@ class TestZoneOutageScenarioPlugin(unittest.TestCase):
             self.assertEqual(result, 1)
 
     @unittest.mock.patch('builtins.open', create=True)
-    @unittest.mock.patch('yaml.full_load')
+    @unittest.mock.patch('yaml.safe_load')
     @unittest.mock.patch('krkn.scenario_plugins.zone_outage.zone_outage_scenario_plugin.gcp_node_scenarios')
     @unittest.mock.patch('krkn.cerberus.publish_kraken_status')
     def test_run_succeeds_when_node_based_zone_succeeds(self, mock_cerberus, mock_gcp_scenarios, mock_yaml, mock_open):
@@ -148,7 +148,7 @@ class TestZoneOutageScenarioPlugin(unittest.TestCase):
             self.assertEqual(result, 0)
 
     @unittest.mock.patch('builtins.open', create=True)
-    @unittest.mock.patch('yaml.full_load')
+    @unittest.mock.patch('yaml.safe_load')
     @unittest.mock.patch('krkn.scenario_plugins.zone_outage.zone_outage_scenario_plugin.AWS')
     @unittest.mock.patch('krkn.cerberus.publish_kraken_status')
     def test_run_aws_network_based_zone(self, mock_cerberus, mock_aws_class, mock_yaml, mock_open):
@@ -186,7 +186,7 @@ class TestZoneOutageScenarioPlugin(unittest.TestCase):
             self.assertEqual(result, 0)
 
     @unittest.mock.patch('builtins.open', create=True)
-    @unittest.mock.patch('yaml.full_load')
+    @unittest.mock.patch('yaml.safe_load')
     @unittest.mock.patch('krkn.scenario_plugins.zone_outage.zone_outage_scenario_plugin.AWS')
     @unittest.mock.patch('krkn.cerberus.publish_kraken_status')
     def test_run_aws_network_based_zone_failure(self, mock_cerberus, mock_aws_class, mock_yaml, mock_open):