Upgrade project dependencies, fix typing errors

example.py

@@ -9,16 +9,19 @@
 import logging
 import re
 import tarfile
-from typing import Optional, TypedDict, Set
+from typing import Optional, TypedDict, Set, TYPE_CHECKING
 
 import docker
 
+if TYPE_CHECKING:
+    from docker.models.containers import ExecResult
+
 class ExecutionResult(TypedDict):
     success: bool
     result: Optional[str]
     locals: Optional[str]
     stdout: Optional[str]
-    docker_exec_result: docker.models.containers.ExecResult
+    docker_exec_result: 'ExecResult'
 
 class PythonExecutes:
     def __init__(self):
@@ -52,9 +55,10 @@ def analyze(self, python_code: str) -> guardx.analysis.types.AnalysisResults:
 
         result = guardx.Guardx().analyze(python_code, {AnalysisType.DETECT_SECRET, AnalysisType.UNSAFE_CODE})
         print(result)
+        return result
 
     def _format_result(
-        self, docker_result: docker.models.containers.ExecResult
+        self, docker_result: 'ExecResult'
     ) -> ExecutionResult:
         """Formats the result from the wrapper script.
 

example_gen_code.py

@@ -10,7 +10,7 @@
 
 def test_yaml_load():
     ystr = yaml.dump({'a': 1, 'b': 2, 'c': 3})
-    y = yaml.load(ystr) #NOSONAR
+    y = yaml.load(ystr, Loader=yaml.SafeLoader) #NOSONAR
     yaml.dump(y)
     try:
         y = yaml.load(ystr, Loader=yaml.CSafeLoader)
@@ -21,15 +21,15 @@ def test_yaml_load():
 
 def test_json_load():
     # no issue should be found
-    j = json.load("{}") #NOSONAR
+    j = json.loads("{}") #NOSONAR
 
 yaml.load("{}", Loader=yaml.Loader)
 
 # no issue should be found
-yaml.load("{}", SafeLoader)
-yaml.load("{}", yaml.SafeLoader)
-yaml.load("{}", CSafeLoader)
-yaml.load("{}", yaml.CSafeLoader)
+yaml.load("{}", Loader=SafeLoader)
+yaml.load("{}", Loader=yaml.SafeLoader)
+yaml.load("{}", Loader=CSafeLoader)
+yaml.load("{}", Loader=yaml.CSafeLoader)
 
 
 print("Hello World")

pyproject.toml

@@ -19,18 +19,18 @@ guardx = {reference = "guardx.tools.cli.guardx:main", type = "console"}
 [tool.poetry.dependencies]
 python = ">=3.10"
 docker = "^7.1.0"
-pydantic = "^2.7.1"
+pydantic = "^2.13.1"
 pyelftools = "^0.31"
 pyseccomp = "^0.1.2"
 pyyaml = "^6.0.1"
-requests = "2.32.4"
-urllib3 = "2.5.0"
+requests = "2.33.1"
+urllib3 = "2.6.3"
 
 [tool.poetry.group.dev]
 optional = true
 
 [tool.poetry.group.dev.dependencies]
-black = "^23.7.0"
+black = "^26.3.1"
 pytest = ">=8.0.0"
 pytest-cov = "^4.1.0"
 pytest-dotenv = "^0.5.2"

src/guardx/analysis/analysis.py

@@ -4,12 +4,17 @@
 import logging
 from typing import Set
 
+import importlib.metadata
 from guardx.analysis.specialization import SpecializationAnalysis, SpecializationAnalysisType
 from guardx.analysis.types import AnalysisResults, AnalysisType
 from guardx.containers import _container
 from guardx.schemas import Config
 
 logger = logging.getLogger(__name__)
+try:
+    __version__ = importlib.metadata.version("guardx")
+except Exception:
+    __version__ = "latest"
 
 
 class StaticAnalysis(object):
@@ -34,6 +39,7 @@ def __init__(self, code: str, analyses: Set[AnalysisType], config: Config) -> No
         self.container_work_dir = "/app"
         self.file_name = "file.py"
 
+
     def analyze(self) -> AnalysisResults:
         """Execute static analyses on the input code."""
         # TBD: add thread pool for running analyses concurrently
@@ -108,7 +114,7 @@ def __detect_unsafe(self):
         logger.debug(result)
         return self.__summarize_bandit(result)
 
-    def init_runner(self, image_name="lab-analyzer:latest"):
+    def init_runner(self, image_name=f"lab-analyzer:{__version__}"):
         """Initialize a container, move code into it, summarize results of static analysis.
 
         Args:

src/guardx/analysis/specialization/specialization.py

@@ -2,7 +2,7 @@
 import ast
 import logging
 import subprocess
-from typing import List, Set
+from typing import List, Optional, Set
 
 from elftools.elf.elffile import ELFFile
 from elftools.elf.sections import SymbolTableSection
@@ -44,7 +44,7 @@ def __check_code_syntax(self):
             logging.error(f"Invalid input code {repr(self.code)}: {err}")
             raise SyntaxError(f"Invalid input: {repr(self.code)}. This is not a valid Python code.") from err
 
-    def __analyze(self, pattern: List[str] = None) -> Set:
+    def __analyze(self, pattern: Optional[List[str]] = None) -> Set:
         logging.debug(f"Compiling input code: {repr(self.code)}")
         try:
             result = subprocess.run([SC_LIST, self.code], capture_output=True, shell=False, text=True, check=True)
@@ -57,7 +57,7 @@ def __analyze(self, pattern: List[str] = None) -> Set:
             elffile = ELFFile(f)
             return self.read_symbol_table_functions(elffile, pattern)
 
-    def read_symbol_table_functions(self, elffile, pattern: List[str] = None) -> Set:
+    def read_symbol_table_functions(self, elffile, pattern: Optional[List[str]] = None) -> Set:
         """Display the functions found in the symbol tables contained in the file."""
         symbol_tables = [(idx, s) for idx, s in enumerate(elffile.iter_sections()) if isinstance(s, SymbolTableSection)]
 
@@ -82,8 +82,7 @@ def get_sc_set(self) -> Set:
         sl = list(map(map_stdlib_sc, self.get_fn_set()))
         return set().union(*sl)
 
-    @staticmethod
     def get_capability_set(self) -> Set:
         """Return the capability set required by the program."""
         cl = {frozenset(map_sc_capabilities(sc)) for sc in self.get_sc_set()}
-        return set().union(*cl)
+        return set().union(*cl)

src/guardx/config.py

@@ -14,7 +14,7 @@ class ConfigLoader:
     """A configuration loader."""
 
     @staticmethod
-    def load_config(path: str = None) -> Config:
+    def load_config(path: str | None) -> Config:
         """Load the SDK configuration from a file path.
 
         Args:

src/guardx/containers/_container.py

@@ -3,11 +3,14 @@
 import logging
 import tarfile
 from importlib import resources as impresources
+from typing import TYPE_CHECKING
 
 import docker
-
 from guardx import sandbox
 
+if TYPE_CHECKING:
+    from docker.errors import DockerException
+
 
 class Container:
     """Internal class to handle container creation and context."""
@@ -25,12 +28,12 @@ def start_container(self):
         """Start the container, sleep for infinity and return handle."""
         try:
             client = docker.from_env()
-        except docker.errors.DockerException as de:
+        except docker.errors.DockerException as de:  # type: ignore[attr-defined]
             raise RuntimeError(
                 "DockerException when trying to get a docker client for the PythonExecutes validator. \
                     Perhaps you need to run a Docker daemon/podman machine?"
             ) from de
-        self.container = client.containers.create(self.docker_image, "sleep infinity")
+        self.container = client.containers.create(self.docker_image, "sleep infinity", detach=True)
         self.container.start()
         logging.info(f"Container using image {self.docker_image} has now started. Info: {self.container}")
 
@@ -133,3 +136,4 @@ def put_json(self, file_name, data):
         tarstream.seek(0)
         self.container.put_archive("/", tarstream)
         return self.container
+

src/guardx/guardx.py

@@ -6,12 +6,13 @@
 from guardx.analysis import AnalysisResults, AnalysisType, StaticAnalysis
 from guardx.config import ConfigLoader
 from guardx.sandbox.stypes import ExecutionResults
+from guardx.schemas import Config
 
 
 class Guardx(object):
     """GuardX class exposing the main consumer API."""
 
-    def __init__(self, config: Dict = None, config_path: str = None) -> None:
+    def __init__(self, config: Config | None = None, config_path: str | None = None) -> None:
         """Entry init block for Guardx.
 
         Args:
@@ -20,7 +21,7 @@ def __init__(self, config: Dict = None, config_path: str = None) -> None:
 
         """
         super().__init__()
-        self.config = ConfigLoader.load_config(config_path) if config is None else config
+        self.config: Config = ConfigLoader.load_config(config_path) if config is None else config
 
     def analyze(self, code: str, analyses: Set[AnalysisType]) -> AnalysisResults:
         """Analyze code using analysis type.
@@ -39,9 +40,9 @@ def analyze(self, code: str, analyses: Set[AnalysisType]) -> AnalysisResults:
     def execute(
         self,
         code: str,
-        analysis_results: AnalysisResults = None,
+        analysis_results: AnalysisResults | None = None,
         dryrun: bool = False, #NOSONAR
-        globals: dict = None,
+        globals: dict | None = None,
     ) -> ExecutionResults:
         """Execute code in a sandbox guarded by security policies.
 
@@ -57,6 +58,8 @@ def execute(
 
         """
         logging.getLogger().setLevel(logging.INFO)
+        if self.config.execution is None:
+            raise ValueError("Execution configuration is required but not provided")
         v = executor.PythonExecutesWithSeccomp(self.config.execution)
         result = v(code, globals)
         return result

src/guardx/sandbox/_executor.py

@@ -2,6 +2,7 @@
 import json
 import os
 import sys
+from typing import cast
 
 from pyseccomp import ALLOW, EQ, KILL, LOG, Arg, SyscallFilter
 
@@ -229,31 +230,37 @@ def setup_seccomp(category):
     f.load()
 
 
-def unsafe_exec_python(python_code, globals, locals):
+def unsafe_exec_python(python_code, globals_dict, locals_dict):
     parsed_statements = list(ast.iter_child_nodes(ast.parse(python_code)))
     if len(parsed_statements) > 1:
         exec(
             compile(
-                ast.Module(body=parsed_statements[:-1], type_ignores=[]),
+                ast.Module(body=cast(list[ast.stmt], parsed_statements[:-1]), type_ignores=[]),
                 filename="<ast>",
                 mode="exec",
             ),
-            globals,
-            locals,
+            globals_dict,
+            locals_dict,
         )
-    return (
-        eval(
-            compile(
-                ast.Expression(body=parsed_statements[-1].value),
-                filename="<ast>",
-                mode="eval",
+    last_statement = parsed_statements[-1]
+    # Type guard: ensure last statement is an Expr node with a value attribute
+    if isinstance(last_statement, ast.Expr):
+        return (
+            eval(
+                compile(
+                    ast.Expression(body=last_statement.value),
+                    filename="<ast>",
+                    mode="eval",
+                ),
+                globals_dict,
+                locals_dict,
             ),
-            globals,
-            locals,
-        ),
-        globals,
-        locals,
-    )
+            globals_dict,
+            locals_dict,
+        )
+    else:
+        # If last statement is not an expression, return None
+        return (None, globals_dict, locals_dict)
 
 
 python_code = open("file.py", "r").read()
@@ -270,13 +277,13 @@ def unsafe_exec_python(python_code, globals, locals):
     globals_dict["__builtins__"] = __builtins__
 
 setup_seccomp(sys.argv[1])
-result, globals, locals = unsafe_exec_python(python_code, globals_dict, {})
+result, result_globals, result_locals = unsafe_exec_python(python_code, globals_dict, {})
 
 # Try to serialize all of the locals. Exclude anything that isn't serializable.
 serialized_stringified_locals = {}
-for local in locals.keys():
+for local in result_locals.keys():
     try:
-        serialized_stringified_locals[local] = str(locals[local])
+        serialized_stringified_locals[local] = str(result_locals[local])
     except Exception:
         pass
 

src/guardx/sandbox/executor.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import re
+from typing import Any
 
 import guardx
 from guardx.containers import _container
@@ -62,7 +63,7 @@ def __init__(self, config: Execution):
             self.config.docker_image = f"lab-validator:{guardx.__version__}"
         self.c_wrapper = _container.Container(self.config.docker_image)
 
-    def __call__(self, code: str, globals: dict = None) -> ExecutionResults:
+    def __call__(self, code: str, globals: dict | None = None) -> ExecutionResults:
         """Execute code as a container under the specified policy.
 
         Args:
@@ -90,7 +91,7 @@ def __call__(self, code: str, globals: dict = None) -> ExecutionResults:
         serializable_globals = _serialize_globals(globals)
         if serializable_globals:
             self.c_wrapper.put_json('globals.json', serializable_globals)
-
+        result: Any = None
         # run the code in a containerized environment. Try 3 times.
         # TODO Kill the code if it takes too long.
         tries = 0

src/guardx/sandbox/stypes.py

@@ -1,9 +1,12 @@
 """Define types for the analysis package."""
 from enum import Enum
-from typing import Dict, List
+from typing import Dict, List, TYPE_CHECKING
 
 import docker
 
+if TYPE_CHECKING:
+    from docker.models.containers import ExecResult
+
 
 class PolicyType(str, Enum):
     """Enumeration of analysis types."""
@@ -32,7 +35,7 @@ def __str__(self) -> str:
 class ExecutionResults(Dict):
     """Execution summary dictionary."""
 
-    def __init__(self, docker_exec_result: docker.models.containers.ExecResult):
+    def __init__(self, docker_exec_result: 'ExecResult'):
         """Constructor.
 
         Initializes execution results.
@@ -43,14 +46,14 @@ def __init__(self, docker_exec_result: docker.models.containers.ExecResult):
         # self.stdout = stdout
         self.docker_exec_result = docker_exec_result
 
-    def get_exit_code(self) -> Dict:
+    def get_exit_code(self) -> int | None:
         """Return the exist code from the execution."""
         return self[ExecutionResultKey.EXIT_CODE] if ExecutionResultKey.EXIT_CODE in self else None
 
-    def get_violations(self) -> List[str]:
+    def get_violations(self) -> List[str] | None:
         """Return the set of policy violations detected in the program execution."""
         return self[ExecutionResultKey.VIOLATIONS] if ExecutionResultKey.VIOLATIONS in self else None
 
-    def get_docker_result(self) -> docker.models.containers.ExecResult:
+    def get_docker_result(self) -> 'ExecResult':
         """Return sandbox execution results."""
         return self.docker_exec_result