Swith Backend to PyGhidra (Ghidra 12); Add Server/Client for DecompilerInterface

.github/workflows/dec-tests.yml

@@ -41,19 +41,18 @@ jobs:
         python ./.github/binja/download_headless.py --serial ${{ env.BN_SERIAL }} --output .github/binja/BinaryNinja-headless.zip
         unzip .github/binja/BinaryNinja-headless.zip -d .github/binja/
         python .github/binja/binaryninja/scripts/install_api.py --install-on-root --silent
-    - name: Set up Java 17
+    - name: Set up Java 21
       uses: actions/setup-java@v4
       with:
         distribution: "oracle"
-        java-version: "17"
+        java-version: "21"
     - name: Install Ghidra
       uses: antoniovazquezblanco/[email protected]
       with:
-        version: "11.1"
+        version: "12.0"
         auth_token: ${{ secrets.GITHUB_TOKEN }}
     - name: Pytest
       run: |
         # these two test must be run in separate python environments, due to the way ghidra bridge works
         # you also must run these tests in the exact order shown here
-        TEST_BINARIES_DIR=/tmp/bs-artifacts/binaries pytest ./tests/test_remote_ghidra.py -s
-        TEST_BINARIES_DIR=/tmp/bs-artifacts/binaries pytest ./tests/test_decompilers.py -s
+        TEST_BINARIES_DIR=/tmp/bs-artifacts/binaries pytest tests/test_decompilers.py tests/test_client_server.py -sv

README.md

@@ -13,14 +13,12 @@ pip install libbs
 ```
 
 The minimum Python version is **3.10**.
-**If you plan on using libbs alone (without installing some other plugin), 
-you must do `libbs --install` after pip install**. This will copy the appropriate files to your decompiler. 
 
 ## Supported Decompilers
 - IDA Pro: **>= 8.4** (if you have an older version, use `v1.26.0`)
 - Binary Ninja: **>= 2.4**
 - angr-management: **>= 9.0**
-- Ghidra: **>= 11.2**
+- Ghidra: **>= 12.0** (started in PyGhidra mode)
 
 ## Usage
 LibBS exposes all decompiler API through the abstract class `DecompilerInterface`. The `DecompilerInterface` 
@@ -45,6 +43,9 @@ for addr in deci.functions:
         deci.functions[function.addr] = function
 ```
 
+Note that for Ghidra in UI mode you must first start it in PyGhidra mode. You can do this by going to your install dir
+and running `./support/pyghidraRun`.
+
 ### Headless Mode 
 To use headless mode you must specify a decompiler to use. You can get the traditional interface using the following:
 
@@ -54,8 +55,8 @@ from libbs.api import DecompilerInterface
 deci = DecompilerInterface.discover(force_decompiler="ghidra", headless=True)
 ```
 
-In the case of Ghidra, you must have the environment variable `GHIDRA_HEADLESS_PATH` set to the path of the Ghidra
-headless binary. This is usually `ghidraRun` or `ghidraHeadlessAnalyzer`.
+In the case of Ghidra, you must have the environment variable `GHIDRA_INSTALL_DIR` set to the path of the Ghidra 
+installation (the place the `ghidraRun` script is located).
 
 ### Artifact Access Caveats
 In designing the dictionaries that contain all Artifacts in a decompiler, we had a clash between ease-of-use and speed. 
@@ -85,7 +86,7 @@ loaded_func = Function.loads(json_str, fmt="json")
 ```
 
 ## Sponsors
-BinSync and it's associated projects would not be possible without sponsorship.
+BinSync and its associated projects would not be possible without sponsorship.
 In no particular order, we'd like to thank all the organizations that have previously or are currently sponsoring
 one of the many BinSync projects.
 

examples/decompiler_client_example.py

@@ -0,0 +1,213 @@
+#!/usr/bin/env python3
+"""
+Example demonstrating the RPyC-based DecompilerClient.
+
+This script shows how to use the new RPyC DecompilerClient which provides
+identical API to DecompilerInterface but connects to a remote server.
+"""
+
+import logging
+import time
+import sys
+from typing import Optional
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+
+def example_with_local_decompiler():
+    """Example using local DecompilerInterface"""
+    try:
+        from libbs.api import DecompilerInterface
+
+        print("=== Using Local DecompilerInterface ===")
+        deci = DecompilerInterface.discover()
+        if deci is None:
+            print("No local decompiler found")
+            return
+
+        demo_decompiler_operations(deci)
+
+    except Exception as e:
+        print(f"Local decompiler error: {e}")
+
+
+def example_with_remote_decompiler(server_url: str = "rpyc://localhost:18861"):
+    """Example using remote DecompilerClient"""
+    try:
+        from libbs.api.decompiler_client import DecompilerClient
+
+        print(f"\n=== Using Remote DecompilerClient ({server_url}) ===")
+        with DecompilerClient.discover(server_url=server_url) as deci:
+            demo_decompiler_operations(deci)
+
+    except Exception as e:
+        print(f"Remote decompiler error: {e}")
+        print("Make sure to start the server first with: libbs --server")
+
+
+def demo_decompiler_operations(deci):
+    """
+    Demo function that works identically with both DecompilerInterface and DecompilerClient.
+
+    This shows the power of the unified API - the same code works regardless of whether
+    the decompiler is local or remote.
+    """
+    print(f"Decompiler: {deci.name}")
+    print(f"Binary path: {deci.binary_path}")
+    print(f"Binary hash: {deci.binary_hash}")
+    print(f"Base address: 0x{deci.binary_base_addr:x}" if deci.binary_base_addr else "None")
+    print(f"Decompiler available: {deci.decompiler_available}")
+
+    # Test fast collection operations (this is where RPyC shines)
+    print(f"\n=== Testing Fast Collection Operations ===")
+
+    # This should be fast - single bulk request for all light artifacts
+    start_time = time.time()
+    functions = list(deci.functions.items())
+    end_time = time.time()
+    print(f"Retrieved {len(functions)} functions in {end_time - start_time:.3f}s")
+
+    # Test other collections
+    collections = [
+        ("comments", deci.comments),
+        ("patches", deci.patches), 
+        ("global_vars", deci.global_vars),
+        ("structs", deci.structs),
+        ("enums", deci.enums),
+        ("typedefs", deci.typedefs)
+    ]
+
+    for name, collection in collections:
+        start_time = time.time()
+        items = list(collection.keys())
+        end_time = time.time()
+        print(f"  {name}: {len(items)} items in {end_time - start_time:.3f}s")
+
+    # Test function access (if any functions exist)
+    if len(deci.functions) > 0:
+        print(f"\n=== Testing Individual Access ===")
+        first_addr = functions[0][0]
+
+        # Test full artifact access via __getitem__ (standard behavior)
+        start_time = time.time()
+        full_func = deci.functions[first_addr]  # This gets the full artifact
+        end_time = time.time()
+        print(f"Full artifact access via []: {end_time - start_time:.3f}s")
+        print(f"Function: {full_func.name} at 0x{full_func.addr:x} (size: {full_func.size})")
+
+        # Test light artifact access (fast, cached)
+        if hasattr(deci.functions, 'get_light'):
+            start_time = time.time()
+            light_func = deci.functions.get_light(first_addr)
+            end_time = time.time()
+            print(f"Light artifact access via get_light(): {end_time - start_time:.6f}s")
+
+        # Show first few functions
+        print("\nFirst 5 functions:")
+        for addr, func in functions[:5]:
+            print(f"  0x{addr:x}: {func.name} (size: {func.size})")
+
+    # Test method calls
+    try:
+        print(f"\n=== Testing Method Calls ===")
+        if len(deci.functions) > 0:
+            first_addr = list(deci.functions.keys())[0]
+            light_func = deci.fast_get_function(first_addr)
+            if light_func:
+                print(f"  fast_get_function(0x{first_addr:x}): {light_func.name}")
+
+            func_size = deci.get_func_size(first_addr)
+            print(f"  get_func_size(0x{first_addr:x}): {func_size}")
+
+            # Test decompilation if available
+            if deci.decompiler_available:
+                start_time = time.time()
+                decomp = deci.decompile(first_addr)
+                end_time = time.time()
+                if decomp:
+                    lines = decomp.text.split('\n')
+                    print(f"  decompile(0x{first_addr:x}): {len(lines)} lines in {end_time - start_time:.3f}s")
+                    print(f"  First line: {lines[0][:80]}...")
+        else:
+            print("  No functions available for testing")
+
+    except Exception as e:
+        print(f"  Method call error: {e}")
+
+
+def discover_decompiler(prefer_remote: bool = False, server_url: str = "rpyc://localhost:18861"):
+    """
+    Smart discovery function that tries remote first if preferred, then falls back to local.
+
+    This demonstrates how you can write code that seamlessly works with either
+    local or remote decompilers based on availability.
+    """
+    if prefer_remote:
+        # Try remote first
+        try:
+            from libbs.api.decompiler_client import DecompilerClient
+            return DecompilerClient.discover(server_url=server_url)
+        except Exception:
+            pass
+
+        # Fall back to local
+        try:
+            from libbs.api import DecompilerInterface
+            return DecompilerInterface.discover()
+        except Exception:
+            return None
+    else:
+        # Try local first
+        try:
+            from libbs.api import DecompilerInterface
+            return DecompilerInterface.discover()
+        except Exception:
+            pass
+
+        # Fall back to remote
+        try:
+            from libbs.api.decompiler_client import DecompilerClient
+            return DecompilerClient.discover(server_url=server_url)
+        except Exception:
+            return None
+
+
+def main():
+    if len(sys.argv) > 1:
+        server_url = sys.argv[1]
+    else:
+        server_url = "rpyc://localhost:18861"
+
+    print("LibBS DecompilerClient Example")
+    print("==============================")
+
+    # Demo 1: Try local decompiler
+    example_with_local_decompiler()
+
+    # Demo 2: Try remote decompiler
+    example_with_remote_decompiler(server_url)
+
+    # Demo 3: Smart discovery
+    print(f"\n=== Smart Discovery (prefer remote) ===")
+    deci = discover_decompiler(prefer_remote=True, server_url=server_url)
+    if deci:
+        print(f"Discovered: {type(deci).__name__}")
+        demo_decompiler_operations(deci)
+        if hasattr(deci, 'shutdown'):
+            deci.shutdown()
+    else:
+        print("No decompiler available (local or remote)")
+
+    print(f"\n=== Smart Discovery (prefer local) ===")
+    deci = discover_decompiler(prefer_remote=False, server_url=server_url)
+    if deci:
+        print(f"Discovered: {type(deci).__name__}")
+        demo_decompiler_operations(deci)
+        if hasattr(deci, 'shutdown'):
+            deci.shutdown()
+    else:
+        print("No decompiler available (local or remote)")
+
+
+if __name__ == "__main__":
+    main()

libbs/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "2.16.5"
+__version__ = "3.0.0"
 
 
 import logging