Headless decompilation of a complete binary fails (#116)

* Output error msg instead of crash, fix cannot decompile cause test if in symbols always returns true (dunno, maybe after Binja update?)

* Change to print error message in decompiled function body, updated error messages

* Remove unused import, add docstrings

Co-authored-by: Mariia Rybalka <[email protected]>

decompiler/backend/codegenerator.py

@@ -25,9 +25,9 @@ def generate(self, tasks: Iterable[DecompilerTask], run_cleanup: bool = True):
         """Generate C-Code for the given list of Tasks sharing global variables."""
         string_blocks: List[str] = []
         if self._declare_globals:
-            string_blocks.append(GlobalDeclarationGenerator.from_asts(task.syntax_tree for task in tasks))
+            string_blocks.append(GlobalDeclarationGenerator.from_asts(task.syntax_tree for task in tasks if not task.failed))
         for task in tasks:
-            if run_cleanup:
+            if run_cleanup and not task.failed:
                 task.syntax_tree.clean_up()
             string_blocks.append(self.generate_function(task))
         return "\n\n".join(string_blocks)
@@ -38,6 +38,6 @@ def generate_function(self, task: DecompilerTask) -> str:
             return_type=task.function_return_type,
             name=task.name,
             parameters=", ".join(map(lambda param: f"{param.type} {param.name}", task.function_parameters)),
-            local_declarations=LocalDeclarationGenerator.from_task(task),
-            function_body=CodeVisitor(task).visit(task.syntax_tree.root),
+            local_declarations=LocalDeclarationGenerator.from_task(task) if not task.failed else "",
+            function_body=CodeVisitor(task).visit(task.syntax_tree.root) if not task.failed else task.failure_message,
         )

decompiler/frontend/binaryninja/frontend.py

@@ -1,4 +1,5 @@
 """Class implementing the main binaryninja frontend interface."""
+import logging
 from typing import List, Tuple, Union
 
 from binaryninja import BinaryView, BinaryViewType, Function
@@ -54,8 +55,13 @@ def create_task(self, function: Union[str, Function], options: Options) -> Decom
         if not isinstance(function, Function):
             function = self._find_function(function)
         return_type, params = self._extract_return_type_and_params(function)
-        cfg = self._extract_cfg(function, options)
-        task = DecompilerTask(function.name, cfg, function_return_type=return_type, function_parameters=params, options=options)
+        try:
+            cfg = self._extract_cfg(function, options)
+            task = DecompilerTask(function.name, cfg, function_return_type=return_type, function_parameters=params, options=options)
+        except Exception as e:
+            task = DecompilerTask(function.name, None, function_return_type=return_type, function_parameters=params, options=options)
+            task.fail(origin="CFG creation")
+            logging.error(f"Failed to decompile {task.name}, error during CFG creation: {e}")
         task.function = function
         return task
 

decompiler/pipeline/pipeline.py

@@ -1,7 +1,7 @@
 """Module containing pipeline definitions for the decompiler."""
 from __future__ import annotations
 
-from logging import debug, warning
+from logging import debug, error, warning
 from typing import List
 
 from decompiler.pipeline.controlflowanalysis.restructuring import PatternIndependentRestructuring
@@ -91,12 +91,20 @@ def run(self, task: DecompilerTask):
         if show_starting_point:
             self._show_stage(task, "Starting point", print_ascii, show_in_tabs)
 
+        if task.failed:
+            return
+
         for stage in self.stages:
             debug(f"stage {stage.name}")
             instance = stage()
-            instance.run(task)
-            if show_all or stage.name in showed_stages:
-                self._show_stage(task, f"After {stage.name}", print_ascii, show_in_tabs)
+            try:
+                instance.run(task)
+                if show_all or stage.name in showed_stages:
+                    self._show_stage(task, f"After {stage.name}", print_ascii, show_in_tabs)
+            except Exception as e:
+                task.fail(origin=stage.name)
+                error(f"Failed to decompile {task.name}, error during stage {stage.name}: {e}")
+                break
 
     @staticmethod
     def _show_stage(task: DecompilerTask, stage_name: str, print_ascii: bool, show_in_tabs: bool):

decompiler/task.py

@@ -34,6 +34,8 @@ def __init__(
         self._function_return_type = function_return_type
         self._function_parameters = function_parameters if function_parameters else []
         self._options: Options = options if options else Options.load_default_options()
+        self._failed = False
+        self._failure_origin = None
 
     @property
     def name(self) -> str:
@@ -69,3 +71,23 @@ def options(self) -> Options:
     def options(self, value: Options):
         """Setter function for task options."""
         self._options = value
+
+    @property
+    def failed(self) -> bool:
+        """Returns True if an error occurred during a decompilation stage.
+
+        A failed tasks will not produce valid decompiled code but an error message will be shown."""
+        return self._failed
+
+    def fail(self, origin: str = None):
+        """Sets the task to failed and the origin to the name of the stage where failure occurred."""
+        self._failed = True
+        self._failure_origin = origin
+
+    @property
+    def failure_message(self) -> str:
+        """Returns the message to be shown for a failed task."""
+        msg = f"Failed to decompile"
+        if self._failure_origin:
+            msg += f" due to error during {self._failure_origin}."
+        return msg