Harden security, robustness, and quality across automation, SSH, diagram, and AI modules

.codacy.yml

@@ -0,0 +1,7 @@
+---
+# Test code legitimately exercises security-sensitive patterns (subprocess
+# execution, fake SSH credentials, cleartext URLs fed to the SSRF validator),
+# so security linters raise false positives there. Exclude tests from analysis;
+# production code under pybreeze/ is still fully scanned.
+exclude_paths:
+  - 'test/**'

.github/workflows/dev.yml

@@ -29,8 +29,8 @@ jobs:
       run: python -m pip install --upgrade pip setuptools wheel
     - name: Install dev dependencies
       run: python -m pip install -r dev_requirements.txt
-    - name: Install pytest
-      run: python -m pip install pytest
+    - name: Install test tooling
+      run: python -m pip install pytest hypothesis
     - name: Run unit tests (pytest)
       run: python -m pytest test/test_utils/ -v --tb=short
     - name: Run AutomationEditor With Debug Mode

.github/workflows/stable.yml

@@ -29,8 +29,8 @@ jobs:
       run: python -m pip install --upgrade pip setuptools wheel
     - name: Install dependencies
       run: python -m pip install -r requirements.txt
-    - name: Install pytest
-      run: python -m pip install pytest
+    - name: Install test tooling
+      run: python -m pip install pytest hypothesis
     - name: Run unit tests (pytest)
       run: python -m pytest test/test_utils/ -v --tb=short
     - name: Run AutomationEditor With Debug Mode

.sonarcloud.properties

@@ -0,0 +1,3 @@
+sonar.sources=pybreeze
+sonar.tests=test
+sonar.python.version=3.10, 3.11, 3.12, 3.13, 3.14

pybreeze/extend/process_executor/api_testka/api_testka_process.py

@@ -2,62 +2,41 @@
 
 from typing import TYPE_CHECKING
 
-from pybreeze.extend.process_executor.process_executor_utils import build_process
+from pybreeze.extend.process_executor.process_executor_utils import (
+    build_process, run_dir_files_with_package,
+)
 
 if TYPE_CHECKING:
     from pybreeze.pybreeze_ui.editor_main.main_ui import PyBreezeMainWindow
 
-from pybreeze.utils.file_process.get_dir_file_list import ask_and_get_dir_files_as_list
-from pybreeze.utils.logging.logger import pybreeze_logger
+_PACKAGE = "je_api_testka"
 
 
 def call_api_testka(
         main_window: PyBreezeMainWindow,
         exec_str: str | None = None,
         program_buffer: int = 1024000
 ):
-    build_process(main_window, "je_api_testka", exec_str, False, program_buffer)
+    build_process(main_window, _PACKAGE, exec_str, False, program_buffer)
 
 
 def call_api_testka_with_send(
         main_window: PyBreezeMainWindow,
         exec_str: str | None = None,
         program_buffer: int = 1024000
 ):
-    build_process(main_window, "je_api_testka", exec_str, True, program_buffer)
+    build_process(main_window, _PACKAGE, exec_str, True, program_buffer)
 
 
 def call_api_testka_multi_file(
         main_window: PyBreezeMainWindow,
         program_buffer: int = 1024000
 ):
-    try:
-        need_to_execute_list = ask_and_get_dir_files_as_list(main_window)
-        if need_to_execute_list is not None and len(need_to_execute_list) > 0:
-            for execute_file in need_to_execute_list:
-                with open(execute_file, encoding="utf-8") as test_script_json:
-                    call_api_testka(
-                        main_window,
-                        test_script_json.read(),
-                        program_buffer
-                    )
-    except Exception as error:
-        pybreeze_logger.error(f"api_testka multi file error: {error}")
+    run_dir_files_with_package(main_window, _PACKAGE, False, program_buffer)
 
 
 def call_api_testka_multi_file_and_send(
         main_window: PyBreezeMainWindow,
         program_buffer: int = 1024000
 ):
-    try:
-        need_to_execute_list = ask_and_get_dir_files_as_list(main_window)
-        if need_to_execute_list is not None and len(need_to_execute_list) > 0:
-            for execute_file in need_to_execute_list:
-                with open(execute_file, encoding="utf-8") as test_script_json:
-                    call_api_testka_with_send(
-                        main_window,
-                        test_script_json.read(),
-                        program_buffer
-                    )
-    except Exception as error:
-        pybreeze_logger.error(f"api_testka multi file and send error: {error}")
+    run_dir_files_with_package(main_window, _PACKAGE, True, program_buffer)

pybreeze/extend/process_executor/auto_control/auto_control_process.py

@@ -2,62 +2,41 @@
 
 from typing import TYPE_CHECKING
 
-from pybreeze.extend.process_executor.process_executor_utils import build_process
+from pybreeze.extend.process_executor.process_executor_utils import (
+    build_process, run_dir_files_with_package,
+)
 
 if TYPE_CHECKING:
     from pybreeze.pybreeze_ui.editor_main.main_ui import PyBreezeMainWindow
 
-from pybreeze.utils.file_process.get_dir_file_list import ask_and_get_dir_files_as_list
-from pybreeze.utils.logging.logger import pybreeze_logger
+_PACKAGE = "je_auto_control"
 
 
 def call_auto_control(
         main_window: PyBreezeMainWindow,
         exec_str: str | None = None,
         program_buffer: int = 1024000
 ):
-    build_process(main_window, "je_auto_control", exec_str, False, program_buffer)
+    build_process(main_window, _PACKAGE, exec_str, False, program_buffer)
 
 
 def call_auto_control_with_send(
         main_window: PyBreezeMainWindow,
         exec_str: str | None = None,
         program_buffer: int = 1024000
 ):
-    build_process(main_window, "je_auto_control", exec_str, True, program_buffer)
+    build_process(main_window, _PACKAGE, exec_str, True, program_buffer)
 
 
 def call_auto_control_multi_file(
         main_window: PyBreezeMainWindow,
         program_buffer: int = 1024000
 ):
-    try:
-        need_to_execute_list = ask_and_get_dir_files_as_list(main_window)
-        if need_to_execute_list is not None and len(need_to_execute_list) > 0:
-            for execute_file in need_to_execute_list:
-                with open(execute_file, encoding="utf-8") as test_script_json:
-                    call_auto_control(
-                        main_window,
-                        test_script_json.read(),
-                        program_buffer
-                    )
-    except Exception as error:
-        pybreeze_logger.error(f"auto control multi file error: {error}")
+    run_dir_files_with_package(main_window, _PACKAGE, False, program_buffer)
 
 
 def call_auto_control_multi_file_and_send(
         main_window: PyBreezeMainWindow,
         program_buffer: int = 1024000
 ):
-    try:
-        need_to_execute_list = ask_and_get_dir_files_as_list(main_window)
-        if need_to_execute_list is not None and len(need_to_execute_list) > 0:
-            for execute_file in need_to_execute_list:
-                with open(execute_file, encoding="utf-8") as test_script_json:
-                    call_auto_control_with_send(
-                        main_window,
-                        test_script_json.read(),
-                        program_buffer
-                    )
-    except Exception as error:
-        pybreeze_logger.error(f"auto control multi file and send error: {error}")
+    run_dir_files_with_package(main_window, _PACKAGE, True, program_buffer)

pybreeze/extend/process_executor/file_automation/file_automation_process.py

@@ -2,62 +2,41 @@
 
 from typing import TYPE_CHECKING
 
-from pybreeze.extend.process_executor.process_executor_utils import build_process
+from pybreeze.extend.process_executor.process_executor_utils import (
+    build_process, run_dir_files_with_package,
+)
 
 if TYPE_CHECKING:
     from pybreeze.pybreeze_ui.editor_main.main_ui import PyBreezeMainWindow
 
-from pybreeze.utils.file_process.get_dir_file_list import ask_and_get_dir_files_as_list
-from pybreeze.utils.logging.logger import pybreeze_logger
+_PACKAGE = "automation_file"
 
 
 def call_file_automation_test(
         main_window: PyBreezeMainWindow,
         exec_str: str | None = None,
         program_buffer: int = 1024000
 ):
-    build_process(main_window, "automation_file", exec_str, False, program_buffer)
+    build_process(main_window, _PACKAGE, exec_str, False, program_buffer)
 
 
 def call_file_automation_test_with_send(
         main_window: PyBreezeMainWindow,
         exec_str: str | None = None,
         program_buffer: int = 1024000
 ):
-    build_process(main_window, "automation_file", exec_str, True, program_buffer)
+    build_process(main_window, _PACKAGE, exec_str, True, program_buffer)
 
 
 def call_file_automation_test_multi_file(
         main_window: PyBreezeMainWindow,
         program_buffer: int = 1024000
 ):
-    try:
-        need_to_execute_list = ask_and_get_dir_files_as_list(main_window)
-        if need_to_execute_list is not None and len(need_to_execute_list) > 0:
-            for execute_file in need_to_execute_list:
-                with open(execute_file, encoding="utf-8") as test_script_json:
-                    call_file_automation_test(
-                        main_window,
-                        test_script_json.read(),
-                        program_buffer
-                    )
-    except Exception as error:
-        pybreeze_logger.error(f"file automation multi file error: {error}")
+    run_dir_files_with_package(main_window, _PACKAGE, False, program_buffer)
 
 
 def call_file_automation_test_multi_file_and_send(
         main_window: PyBreezeMainWindow,
         program_buffer: int = 1024000
 ):
-    try:
-        need_to_execute_list = ask_and_get_dir_files_as_list(main_window)
-        if need_to_execute_list is not None and len(need_to_execute_list) > 0:
-            for execute_file in need_to_execute_list:
-                with open(execute_file, encoding="utf-8") as test_script_json:
-                    call_file_automation_test_with_send(
-                        main_window,
-                        test_script_json.read(),
-                        program_buffer
-                    )
-    except Exception as error:
-        pybreeze_logger.error(f"file automation multi file and send error: {error}")
+    run_dir_files_with_package(main_window, _PACKAGE, True, program_buffer)

pybreeze/extend/process_executor/file_runner_process.py

@@ -21,6 +21,8 @@
 from je_editor.pyside_ui.main_ui.save_settings.user_color_setting_file import actually_color_dict
 
 from pybreeze.pybreeze_ui.show_code_window.code_window import CodeWindow
+from pybreeze.utils.logging.logger import pybreeze_logger
+from pybreeze.utils.subprocess_util import no_window_creationflags, utf8_subprocess_env
 
 
 class FileRunnerProcess:
@@ -81,6 +83,7 @@ def _compile_and_run(self, compiler: str, args: list, output_flag: str, file_pat
                 compile_cmd,
                 capture_output=True,
                 timeout=60,
+                creationflags=no_window_creationflags(),
             )
         except FileNotFoundError:
             self._append_text(f"[Error] Compiler not found: {compiler}\n", is_error=True)
@@ -117,6 +120,8 @@ def _start_process(self, command: list[str], cleanup_binary: str | None = None)
                 stderr=subprocess.PIPE,
                 stdin=subprocess.PIPE,
                 shell=False,
+                creationflags=no_window_creationflags(),
+                env=utf8_subprocess_env(self.program_encoding),
             )
         except FileNotFoundError:
             self._append_text(f"[Error] Command not found: {command[0]}\n", is_error=True)
@@ -189,8 +194,8 @@ def _finish(self) -> None:
         if self._cleanup_binary:
             try:
                 os.remove(self._cleanup_binary)
-            except OSError:
-                pass
+            except OSError as error:
+                pybreeze_logger.debug("Could not remove compiled binary %s: %s", self._cleanup_binary, error)
 
     def _drain_queues(self) -> None:
         """Drain all remaining messages from output/error queues to UI."""
@@ -211,37 +216,29 @@ def _drain_queues(self) -> None:
             except queue.Empty:
                 break
 
-    def _read_stdout(self) -> None:
+    def _read_stream(self, stream_name: str, target_queue: Queue) -> None:
+        # Empty read from readline means the pipe reached EOF (the child closed
+        # the stream), so stop immediately rather than spinning on a closed pipe
+        # until the process is reaped.
         try:
             while self.still_running:
                 proc = self.process
                 if proc is None:
                     break
-                data = proc.stdout.readline(self.program_buffer_size)
-                if not data and proc.poll() is not None:
+                data = getattr(proc, stream_name).readline(self.program_buffer_size)
+                if not data:
                     break
                 if isinstance(data, bytes):
                     data = data.decode(self.program_encoding, "replace")
-                if data:
-                    self.output_queue.put(data)
-        except (OSError, ValueError):
-            pass
+                target_queue.put(data)
+        except (OSError, ValueError) as error:
+            pybreeze_logger.debug("Reader for %s stopped: %s", stream_name, error)
+
+    def _read_stdout(self) -> None:
+        self._read_stream("stdout", self.output_queue)
 
     def _read_stderr(self) -> None:
-        try:
-            while self.still_running:
-                proc = self.process
-                if proc is None:
-                    break
-                data = proc.stderr.readline(self.program_buffer_size)
-                if not data and proc.poll() is not None:
-                    break
-                if isinstance(data, bytes):
-                    data = data.decode(self.program_encoding, "replace")
-                if data:
-                    self.error_queue.put(data)
-        except (OSError, ValueError):
-            pass
+        self._read_stream("stderr", self.error_queue)
 
     def _append_text(self, text: str, is_error: bool) -> None:
         """Append text to the code result widget."""