Add Ghidra renaming interface

.github/workflows/release.yml

@@ -0,0 +1,40 @@
+name: Release
+
+on:
+  push:
+    tags:
+    - "v**"
+
+jobs:
+
+  release-github:
+    name: Create Github Release
+    permissions: write-all
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v2
+      - name: Create Release
+        uses: ncipollo/release-action@v1
+        with:
+          generateReleaseNotes: true
+
+  release-pypi:
+    name: Release pypi package
+    runs-on: ubuntu-latest
+    steps:
+    - name: Checkout source
+      uses: actions/checkout@v2
+    - name: Setup Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.8'
+    - name: Install build
+      run: pip install build
+    - name: Build dists
+      run: python -m build
+    - name: Release to PyPI
+      uses: pypa/gh-action-pypi-publish@27b31702a0e7fc50959f5ad993c78deac1bdfc29
+      with:
+        user: __token__
+        password: ${{ secrets.PYPI_API_TOKEN }}

yodalib/__init__.py

@@ -1 +1 @@
-__version__ = "0.0.0"
+__version__ = "0.1.0"

yodalib/api/decompiler_interface.py

@@ -21,6 +21,18 @@
 _l = logging.getLogger(name=__name__)
 
 
+def requires_decompilation(f):
+    @wraps(f)
+    def _requires_decompilation(self, *args, **kwargs):
+        if self._decompiler_available:
+            for arg in args:
+                if isinstance(arg, Function) and arg.dec_obj is None:
+                    arg.dec_obj = self._get_decompilation_object(arg)
+
+        return f(self, *args, **kwargs)
+    return _requires_decompilation
+
+
 def artifact_set_event(f):
     @wraps(f)
     def _artifact_set_event(self: "DecompilerInterface", *args, **kwargs):
@@ -42,7 +54,8 @@ def __init__(
         self,
         artifact_lifter: Optional[ArtifactLifter] = None,
         headless: bool = False,
-        error_on_artifact_duplicates: bool = False
+        error_on_artifact_duplicates: bool = False,
+        decompiler_available: bool = True,
     ):
         self.headless = headless
         self.artifact_lifer = artifact_lifter
@@ -59,6 +72,7 @@ def __init__(
         self.patches = ArtifactDict(Patch, self, error_on_duplicate=error_on_artifact_duplicates)
         #self.stack_vars = ArtifactDict(StackVariable, self, error_on_duplicate=error_on_artifact_duplicates)
 
+        self._decompiler_available = decompiler_available
         if not self.headless:
             self._init_ui_components()
 
@@ -170,6 +184,9 @@ def get_func_containing(self, addr: int) -> Optional[Function]:
     def _decompile(self, function: Function) -> Optional[str]:
         raise NotImplementedError
 
+    def _get_decompilation_object(self, function: Function) -> Optional[object]:
+        raise NotImplementedError
+
     #
     # Override Optional API:
     # There are API that provide extra introspection for plugins that may rely on YODA Interface

yodalib/data/artifacts/func.py

@@ -178,15 +178,20 @@ class Function(Artifact):
         "size",
         "header",
         "stack_vars",
+        "dec_obj",
     )
 
-    def __init__(self, addr, size, header=None, stack_vars=None, last_change=None):
+    def __init__(self, addr, size, header=None, stack_vars=None, dec_obj=None, last_change=None):
         super(Function, self).__init__(last_change=last_change)
         self.addr: int = addr
         self.size: int = size
         self.header: Optional[FunctionHeader] = header
         self.stack_vars: Dict[int, StackVariable] = stack_vars or {}
 
+        # a special property which can only be set while running inside the decompiler.
+        # contains a reference to the decompiler object associated with this function.
+        self.dec_obj = dec_obj
+
     def __str__(self):
         if self.header:
             return f"<Function: {self.header.type} {self.name}(args={len(self.args)}); @{hex(self.addr)} " \
@@ -281,7 +286,7 @@ def diff(self, other, **kwargs) -> Dict:
         return diff_dict
 
     def copy(self):
-        func = Function(self.addr, self.size, last_change=self.last_change)
+        func = Function(self.addr, self.size, last_change=self.last_change, dec_obj=self.dec_obj)
         func.header = self.header.copy() if self.header else None
         func.stack_vars = {k: v.copy() for k, v in self.stack_vars.items()}
 

yodalib/decompilers/ghidra/interface.py

@@ -3,6 +3,7 @@
 from functools import wraps
 
 from yodalib.api import DecompilerInterface
+from yodalib.api.decompiler_interface import requires_decompilation
 from yodalib.data import (
     Function, FunctionHeader, StackVariable, Comment, FunctionArgument, GlobalVariable, Struct, StructMember
 )
@@ -40,8 +41,10 @@ def __init__(self, **kwargs):
         self._last_func = None
         self.base_addr = None
 
-    def _init_headless_components(self):
-        self.connect_ghidra_bridge()
+        # connect to the remote bridge, assumes Ghidra is already running!
+        if not self.headless:
+            if not self.connect_ghidra_bridge():
+                raise Exception("Failed to connect to remote Ghidra Bridge. Did you start it first?")
 
     #
     # Controller API
@@ -87,7 +90,39 @@ def connect_ghidra_bridge(self):
         return self.ghidra.connected
 
     def _decompile(self, function: Function) -> Optional[str]:
-        return None
+        dec_obj = self._get_decompilation_object(function)
+        if dec_obj is None:
+            return None
+
+        dec_func = dec_obj.getDecompiledFunction()
+        if dec_func is None:
+            return None
+
+        return str(dec_func.getC())
+
+    def _get_decompilation_object(self, function: Function) -> Optional[object]:
+        return self._ghidra_decompile(self._get_nearest_function(function.addr))
+
+    #
+    # Override Optional API:
+    # There are API that provide extra introspection for plugins that may rely on YODA Interface
+    #
+
+    def local_variable_names(self, func: Function) -> List[str]:
+        symbols_by_name = self._get_local_variable_symbols(func)
+        return list(symbols_by_name.keys())
+
+    def rename_local_variables_by_names(self, func: Function, name_map: Dict[str, str]) -> bool:
+        symbols_by_name = self._get_local_variable_symbols(func)
+        symbols_to_update = {}
+        for name, new_name in name_map.items():
+            if name not in symbols_by_name or symbols_by_name[name].name == new_name or new_name in symbols_by_name:
+                continue
+
+            sym: "HighSymbol" = symbols_by_name[name]
+            symbols_to_update[sym] = (new_name, None)
+
+        return self._update_local_variable_symbols(symbols_to_update) if symbols_to_update else False
 
     #
     # Artifact API
@@ -117,7 +152,7 @@ def _get_function(self, addr, **kwargs) -> Optional[Function]:
         bs_func = Function(
             func.getEntryPoint().getOffset(), func.getBody().getNumAddresses(),
             header=FunctionHeader(func.getName(), func.getEntryPoint().getOffset()),
-            stack_vars=stack_variables
+            stack_vars=stack_variables, dec_obj=dec
         )
         return bs_func
 
@@ -169,19 +204,6 @@ def _set_function_header(self, fheader: FunctionHeader, decompilation=None, **kw
     # Filler/Setter API
     #
 
-    def fill_function(self, func_addr, user=None, artifact=None, **kwargs):
-        decompilation = self._ghidra_decompile(self._get_nearest_function(func_addr))
-        changes = super().fill_function(
-            func_addr, user=user, artifact=artifact, decompilation=decompilation, **kwargs
-        )
-
-        return changes
-
-    @ghidra_transaction
-    def fill_struct(self, struct_name, header=True, members=True, artifact=None, **kwargs):
-        struct: Struct = artifact;
-
-
     @ghidra_transaction
     def fill_stack_variable(self, func_addr, offset, user=None, artifact=None, decompilation=None, **kwargs):
         stack_var: StackVariable = artifact
@@ -265,7 +287,6 @@ def struct(self, name) -> Optional[Struct]:
         return bs_struct
 
     def structs(self) -> Dict[str, Struct]:
-        structures = self.ghidra.currentProgram.getDataTypeManager().getAllStructures()
         name_sizes: Optional[List[Tuple[str, int]]] = self.ghidra.bridge.remote_eval(
             "[(s.getPathName(), s.getLength())"
             "for s in currentProgram.getDataTypeManager().getAllStructures()]"
@@ -318,13 +339,37 @@ def stack_variable(self, func_addr, offset) -> Optional[StackVariable]:
         bs_stack_var = self._gstack_var_to_bsvar(gstack_var)
         return bs_stack_var
 
-
     #
     # Ghidra Specific API
     #
 
+    @ghidra_transaction
+    def _update_local_variable_symbols(
+        self, symbols: Dict["HighSymbol", Tuple[str, Optional["DataType"]]]
+    ) -> bool:
+        """
+        @param decompilation:
+        @param symbols: of form [Symbol] = (new_name, new_type)
+        """
+        HighFunctionDBUtil = self.ghidra.import_module_object("ghidra.program.model.pcode", "HighFunctionDBUtil")
+        SourceType = self.ghidra.import_module_object("ghidra.program.model.symbol", "SourceType")
+        update_list = self.ghidra.bridge.remote_eval(
+            "[HighFunctionDBUtil.updateDBVariable(sym, updates[0], updates[1], SourceType.ANALYSIS) "
+            "for sym, updates in symbols.items()]",
+            HighFunctionDBUtil=HighFunctionDBUtil, SourceType=SourceType, symbols=symbols
+        )
+        return any([u is not None for u in update_list])
+
+    @requires_decompilation
+    def _get_local_variable_symbols(self, func: Function) -> Dict[str, "HighSymbol"]:
+        high_func = func.dec_obj.getHighFunction()
+        return self.ghidra.bridge.remote_eval(
+            "{sym.name: sym for sym in high_func.getLocalSymbolMap().getSymbols() if sym.name}",
+            high_func=high_func
+        )
+
     def _get_struct_by_name(self, name: str) -> "GhidraStructure":
-        return self.ghidra.currentProgram.getDataTypeManager().getDataType(name);
+        return self.ghidra.currentProgram.getDataTypeManager().getDataType(name)
 
     def _get_nearest_function(self, addr: int) -> "GhidraFunction":
         func_manager = self.ghidra.currentProgram.getFunctionManager()