Added hardening script

Author: andrecsilva

pyproject.toml

@@ -45,6 +45,7 @@ Repository = "https://github.com/pixee/codemodder-python"
 
 [project.scripts]
 codemodder = "codemodder.codemodder:main"
+codemodder_hardening = "codemodder.codemodder:harden"
 generate-docs = 'codemodder.scripts.generate_docs:main'
 get-hashes = 'codemodder.scripts.get_hashes:main'
 

src/codemodder/codemodder.py

@@ -73,6 +73,7 @@ def log_report(context, output, elapsed_ms, files_to_analyze, token_usage):
 def apply_codemods(
     context: CodemodExecutionContext,
     codemods_to_run: Sequence[BaseCodemod],
+    hardening: bool,
 ) -> TokenUsage:
     log_section("scanning")
     token_usage = TokenUsage()
@@ -89,7 +90,7 @@ def apply_codemods(
     for codemod in codemods_to_run:
         # NOTE: this may be used as a progress indicator by upstream tools
         logger.info("running codemod %s", codemod.id)
-        if codemod_token_usage := codemod.apply(context):
+        if codemod_token_usage := codemod.apply(context, hardening):
             log_token_usage(f"Codemod {codemod.id}", codemod_token_usage)
             token_usage += codemod_token_usage
 
@@ -135,6 +136,7 @@ def run(
     sast_only: bool = False,
     ai_client: bool = True,
     log_matched_files: bool = False,
+    hardening: bool = False,
 ) -> tuple[CodeTF | None, int, TokenUsage]:
     start = datetime.datetime.now()
 
@@ -206,7 +208,7 @@ def run(
         context.find_and_fix_paths,
     )
 
-    token_usage = apply_codemods(context, codemods_to_run)
+    token_usage = apply_codemods(context, codemods_to_run, hardening)
 
     elapsed = datetime.datetime.now() - start
     elapsed_ms = int(elapsed.total_seconds() * 1000)
@@ -231,7 +233,7 @@ def run(
     return codetf, 0, token_usage
 
 
-def _run_cli(original_args) -> int:
+def _run_cli(original_args, hardening=False) -> int:
     codemod_registry = registry.load_registered_codemods()
     argv = parse_args(original_args, codemod_registry)
     if not os.path.exists(argv.directory):
@@ -270,7 +272,8 @@ def _run_cli(original_args) -> int:
 
     _, status, _ = run(
         argv.directory,
-        argv.dry_run,
+        # Force dry-run if not hardening
+        argv.dry_run if hardening else True,
         argv.output,
         argv.output_format,
         argv.verbose,
@@ -284,10 +287,26 @@ def _run_cli(original_args) -> int:
         codemod_registry=codemod_registry,
         sast_only=argv.sonar_issues_json or argv.sarif,
         log_matched_files=True,
+        hardening=hardening,
     )
     return status
 
 
+"""
+Hardens a project. The application will write all the fixes into the files.
+"""
+
+
+def harden():
+    sys_argv = sys.argv[1:]
+    sys.exit(_run_cli(sys_argv, True))
+
+
+"""
+Remediates a project. The application will suggest fix for each separate issue found. No files will be written.
+"""
+
+
 def main():
     sys_argv = sys.argv[1:]
     sys.exit(_run_cli(sys_argv))

src/codemodder/codemods/base_codemod.py

@@ -189,6 +189,7 @@ def _apply(
         self,
         context: CodemodExecutionContext,
         rules: list[str],
+        hardening: bool,
     ) -> None | TokenUsage:
         if self.provider and (
             not (provider := context.providers.get_provider(self.provider))
@@ -226,21 +227,26 @@ def _apply(
             logger.debug("No files matched for %s", self.id)
             return None
 
-        # Do each result independently
-        if results:
+        # Do each result independently and outputs the diffs
+        if not hardening:
             # gather positional arguments for the map
-            resultset_arguments = []
+            resultset_arguments: list[ResultSet | None] = []
             path_arguments = []
-            for result in results.results_for_rules(rules):
-                # this need to be the same type of ResultSet as results
-                singleton = results.__class__()
-                singleton.add_result(result)
-                result_locations = self.get_files_to_analyze(context, singleton)
-                # We do an execution for each location in the result
-                # So we duplicate the resultset argument for each location
-                for loc in result_locations:
-                    resultset_arguments.append(singleton)
-                    path_arguments.append(loc)
+            if results:
+                for result in results.results_for_rules(rules):
+                    # this need to be the same type of ResultSet as results
+                    singleton = results.__class__()
+                    singleton.add_result(result)
+                    result_locations = self.get_files_to_analyze(context, singleton)
+                    # We do an execution for each location in the result
+                    # So we duplicate the resultset argument for each location
+                    for loc in result_locations:
+                        resultset_arguments.append(singleton)
+                        path_arguments.append(loc)
+            # An exception for find-and-fix codemods
+            else:
+                resultset_arguments = [None]
+                path_arguments = files_to_analyze
 
             contexts: list = []
             with ThreadPoolExecutor() as executor:
@@ -251,13 +257,13 @@ def _apply(
                             path, context, resultset, rules
                         ),
                         path_arguments,
-                        resultset_arguments,
+                        resultset_arguments or [None],
                     )
                 )
                 executor.shutdown(wait=True)
 
             context.process_results(self.id, contexts)
-        # for find and fix codemods
+        # Hardens all findings per file at once and writes the fixed code into the file
         else:
             process_file = functools.partial(
                 self._process_file, context=context, results=results, rules=rules
@@ -276,7 +282,9 @@ def _apply(
             context.process_results(self.id, contexts)
         return None
 
-    def apply(self, context: CodemodExecutionContext) -> None | TokenUsage:
+    def apply(
+        self, context: CodemodExecutionContext, hardening: bool = False
+    ) -> None | TokenUsage:
         """
         Apply the codemod with the given codemod execution context
 
@@ -292,7 +300,7 @@ def apply(self, context: CodemodExecutionContext) -> None | TokenUsage:
 
         :param context: The codemod execution context
         """
-        return self._apply(context, [self._internal_name])
+        return self._apply(context, [self._internal_name], hardening)
 
     def _process_file(
         self,
@@ -390,8 +398,10 @@ def __init__(
         if requested_rules:
             self.requested_rules.extend(requested_rules)
 
-    def apply(self, context: CodemodExecutionContext) -> None | TokenUsage:
-        return self._apply(context, self.requested_rules)
+    def apply(
+        self, context: CodemodExecutionContext, hardening: bool = False
+    ) -> None | TokenUsage:
+        return self._apply(context, self.requested_rules, hardening)
 
     def get_files_to_analyze(
         self,

src/core_codemods/defectdojo/api.py

@@ -76,9 +76,11 @@ def from_core_codemod(
     def apply(
         self,
         context: CodemodExecutionContext,
+        hardening: bool = False,
     ) -> None:
         self._apply(
             context,
             # We know this has a tool because we created it with `from_core_codemod`
             cast(ToolMetadata, self._metadata.tool).rule_ids,
+            hardening,
         )