feat(PRO-3285)!: update plugin to use APIs introduced by V8 pipeline work

.gitignore

@@ -54,4 +54,8 @@ Thumbs.db
 
 reai_toolkit/vendor/
 reai_toolkit/vendor/*
-plugin_binary_ninja.egg-info/*
+plugin_binary_ninja.egg-info/*
+
+.dev-sdk/
+# Written by `npx @openapitools/openapi-generator-cli version-manager set`.
+openapitools.json

CLAUDE.md

@@ -0,0 +1,25 @@
+# CLAUDE.md
+
+RevEng.AI Binary Ninja plugin — a bridging layer between Binary Ninja and RevEng.AI's API.
+It calls the generated `revengai` PyPI SDK, then applies results into Binary Ninja.
+**Public repo** — no secrets, no internal hostnames/infra details in code, comments, references to code internal to RevEng.AI, or committed fixtures.
+Treat every change as reviewed by the outside world.
+
+## API surface
+
+All calls to RevEng.AI go through the generated `revengai` package — never hand-roll HTTP.
+`reai_toolkit/features/configuration/config.py::create_api_client` is the one shared client constructor;
+every other call site builds its own `revengai.<X>Api(api_client)` per call (no central wrapper class — see that file for the credential/host resolution).
+
+`tests/unit/sdk/test_sdk_schemas.py` pins the exact SDK surface (classes/methods/model fields) this plugin depends on —
+update it whenever an API call site changes, so a future SDK bump that silently drops something this plugin needs fails loudly here instead of at runtime for a user.
+
+## Tests
+
+Two tiers under `tests/`: `unit/` (mocked, run anywhere) and `headless/` (needs a real Binary Ninja install + license).
+
+## Features manifest
+
+`.revengai/features.json` is generated by `scripts/emit_features.py` and checked for drift in
+CI (`features-drift.yml`). Run `python scripts/emit_features.py --check` before pushing
+whenever a feature is added, renamed, or removed.

pyproject.toml

@@ -7,7 +7,7 @@ name = "plugin-binary-ninja"
 version = "0.0.1"
 requires-python = ">=3.10"
 dependencies = [
- "revengai>=3.123.0",
+ "revengai>=4.4.0",
  "urllib3>=2.0.0,<2.3.0",
  "libbs==2.15.1",
  "pydantic>=2.12.5",

reai_toolkit/features/match_current_function/match_current_function.py

@@ -60,9 +60,9 @@ def parse_confidence(item):
  return False, "Operation cancelled"
 
  with self.config.create_api_client() as api_client:
-analysis_core_instance = revengai.AnalysesResultsMetadataApi(api_client)
- analyzed_functions = analysis_core_instance.get_functions_list(analysis_id)
- analyzed_functions = analyzed_functions.to_dict()["data"]["functions"]
+functions_core_instance = revengai.FunctionsCoreApi(api_client)
+ analyzed_functions = functions_core_instance.list_analysis_functions(analysis_id=analysis_id)
+ analyzed_functions = analyzed_functions.to_dict()["functions"]
  if self.cancelled.is_set():
  return False, "Operation cancelled"
 

reai_toolkit/features/match_functions/match_functions.py

@@ -50,9 +50,9 @@ def parse_confidence(item):
  return False, "Operation cancelled"
 
  with self.config.create_api_client() as api_client:
-analysis_core_instance = revengai.AnalysesResultsMetadataApi(api_client)
- analyzed_functions = analysis_core_instance.get_functions_list(analysis_id)
- analyzed_functions = analyzed_functions.to_dict()["data"]["functions"]
+functions_core_instance = revengai.FunctionsCoreApi(api_client)
+ analyzed_functions = functions_core_instance.list_analysis_functions(analysis_id=analysis_id)
+ analyzed_functions = analyzed_functions.to_dict()["functions"]
  if self.cancelled.is_set():
  return False, "Operation cancelled"
 

reai_toolkit/features/view_function_in_portal/view_function_in_portal.py

@@ -27,9 +27,9 @@ def view_function_in_portal(self, bv: BinaryView, options: Dict) -> None:
  raise Exception("Analysis not found. Please choose one using the 'Attach to existing' feature.")
 
  with self.config.create_api_client() as api_client:
- api_instance = revengai.AnalysesResultsMetadataApi(api_client)
- api_response = api_instance.get_functions_list(analysis_id)
- analyzed_functions = api_response.data.functions
+ api_instance = revengai.FunctionsCoreApi(api_client)
+ api_response = api_instance.list_analysis_functions(analysis_id=analysis_id)
+ analyzed_functions = api_response.functions
  log_info(f"RevEng.AI | Analyzed functions: {analyzed_functions}")
 
  analyzed_function = next((f for f in analyzed_functions if f.function_vaddr == function.start), None)

reai_toolkit/utils/core/binary_ninja.py

@@ -10,9 +10,9 @@ def _rename_in_portal(config: revengai.Configuration, function_id:int, new_name:
  try:
  with config.create_api_client() as api_client:
  api_instance = revengai.FunctionsRenamingHistoryApi(api_client)
- api_instance.rename_function_id(
+ api_instance.rename_function(
  function_id=function_id,
-function_rename=revengai.FunctionRename(
+rename_input_body=revengai.RenameInputBody(
  new_name=new_name,
  new_mangled_name=new_mangled_name
  )
@@ -53,7 +53,7 @@ def parse_date(date_str: str) -> str:
  try:
  dt = datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S.%f")
  return dt.strftime("%Y-%m-%d %H:%M:%S")
- except Exception as e:
+ except Exception:
  return date_str
 
 def get_function_by_addr(bv: BinaryView, addr: int) -> Function:
@@ -72,10 +72,10 @@ def get_function_by_addr(bv: BinaryView, addr: int) -> Function:
 
 def get_function_id_by_addr(bv: BinaryView, addr: int, config):
  with config.create_api_client() as api_client:
- api_instance = revengai.AnalysesResultsMetadataApi(api_client)
+ api_instance = revengai.FunctionsCoreApi(api_client)
  analysis_id = config.get_analysis_id(bv)
- api_response = api_instance.get_functions_list(analysis_id)
- analyzed_functions = api_response.data.functions
+ api_response = api_instance.list_analysis_functions(analysis_id=analysis_id)
+ analyzed_functions = api_response.functions
  target_function = next((f for f in analyzed_functions if f.function_vaddr == addr), None)
  if not target_function:
  log_error(f"RevEng.AI | Function not found at 0x{addr:x}")

reai_toolkit/utils/features/datatypes.py

@@ -1,7 +1,7 @@
-from libbs.artifacts import _art_from_dict, Function, GlobalVariable, Enum, Struct, Typedef
+from libbs.artifacts import _art_from_dict, Function, FunctionArgument, FunctionHeader, GlobalVariable, Enum, Struct, Typedef
 from libbs.api import DecompilerInterface
 from binaryninja import log_error, log_info
-from typing import List
+from typing import List
 
 def apply_type(deci: DecompilerInterface, artifact, soft_skip=False) -> None | str:
  supported_types = [
@@ -115,4 +115,122 @@ def apply_data_types(function_addr: int = 0, signature=None, deci: DecompilerInt
  log_info("RevEng.AI | Successfully applied function signature and dependencies")
 
  except Exception as e:
- log_info(f"RevEng.AI | Error in _apply_data_types: {e}")
+ log_info(f"RevEng.AI | Error in _apply_data_types: {e}")
+
+
+def _resolve_struct_dep(entry: dict, data_types: dict, deps_by_name: dict) -> None:
+ name = entry.get("name")
+ if name is None or name in deps_by_name:
+ return
+
+ # Registered before its members are resolved so a self-referential struct (e.g. a member
+ # pointing back to its own type) terminates instead of recursing forever.
+ deps_by_name[name] = {
+ "artifact_type": "Struct",
+ "name": name,
+ "size": entry.get("size"),
+ "members": {},
+ }
+
+ members = {}
+ for member in (entry.get("definition") or {}).get("members") or []:
+ member_type, _ = _resolve_type(
+ member.get("data_type_id"), data_types, deps_by_name
+ )
+ members[hex(member["offset"])] = {
+ "name": member.get("name"),
+ "offset": member["offset"],
+ "type": member_type,
+ "size": member.get("size"),
+ }
+ deps_by_name[name]["members"] = members
+
+
+def _resolve_enum_dep(entry: dict, deps_by_name: dict) -> None:
+ name = entry.get("name")
+ if name is None or name in deps_by_name:
+ return
+
+ deps_by_name[name] = {
+ "artifact_type": "Enum",
+ "name": name,
+ "members": {
+ value["name"]: int(value["value"])
+ for value in (entry.get("definition") or {}).get("values") or []
+ },
+ }
+
+
+def _resolve_typedef_dep(entry: dict, data_types: dict, deps_by_name: dict) -> None:
+ name = entry.get("name")
+ if name is None or name in deps_by_name:
+ return
+
+ deps_by_name[name] = {"artifact_type": "Typedef", "name": name, "type": None}
+ base_type, _ = _resolve_type(
+ (entry.get("definition") or {}).get("target_data_type_id"),
+ data_types,
+ deps_by_name,
+ )
+ deps_by_name[name]["type"] = base_type
+
+
+def _resolve_type(data_type_id, data_types: dict, deps_by_name: dict) -> tuple:
+ """Resolve a v3 data_type_id to its type name and size, registering any Struct/Enum/Typedef
+ it (or a type it references) names as a dependency in deps_by_name."""
+ if data_type_id is None:
+ return None, None
+
+ entry = data_types.get(str(data_type_id))
+ if entry is None:
+ return None, None
+
+ kind = entry.get("kind")
+ definition = entry.get("definition") or {}
+ if kind in ("STRUCT", "UNION"):
+ _resolve_struct_dep(entry, data_types, deps_by_name)
+ elif kind == "ENUM":
+ _resolve_enum_dep(entry, deps_by_name)
+ elif kind == "TYPEDEF":
+ _resolve_typedef_dep(entry, data_types, deps_by_name)
+ elif kind == "POINTER":
+ _resolve_type(definition.get("pointee_data_type_id"), data_types, deps_by_name)
+ elif kind == "ARRAY":
+ _resolve_type(definition.get("element_data_type_id"), data_types, deps_by_name)
+
+ return entry.get("name"), entry.get("size")
+
+
+def build_signature_data(signature: dict, data_types: dict) -> dict | None:
+ """Convert a v3 BatchFunctionSignatureEntry into the libbs artifact graph apply_data_types
+ applies. Returns None when the analysis holds no signature for this function."""
+ if not signature.get("has_signature"):
+ return None
+
+ deps_by_name: dict = {}
+
+ args = {}
+ for parameter in signature.get("parameters") or []:
+ arg_type, arg_size = _resolve_type(
+ parameter.get("data_type_id"), data_types, deps_by_name
+ )
+ args[parameter["ordinal"]] = FunctionArgument(
+ offset=parameter["ordinal"],
+ name=parameter.get("name"),
+ type_=arg_type,
+ size=arg_size,
+ )
+
+ return_type, _ = _resolve_type(
+ signature.get("return_data_type_id"), data_types, deps_by_name
+ )
+
+ function = Function(
+ header=FunctionHeader(
+ name=signature.get("function_name"),
+ type_=return_type,
+ args=args,
+ )
+ )
+
+ return {"function": function, "deps": list(deps_by_name.values())}

reai_toolkit/utils/features/matching.py

@@ -2,8 +2,8 @@
 from typing import List, Dict, Tuple, Any
 import revengai
 import re
-import time
-from libbs.artifacts import _art_from_dict, Function, FunctionArgument
+from libbs.artifacts import Function, FunctionArgument
+from .datatypes import build_signature_data
 from threading import Event
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
@@ -160,67 +160,52 @@ def _search_binaries(self, query: Dict[str, Any] = {}):
  def _process_data_type_batch(self, chunk: List[Dict], chunk_index: int) -> List[Dict]:
  try:
  log_info(f"RevEng.AI | Processing chunk of {len(chunk)} functions")
- function_ids = set([result['nearest_neighbor_id'] for result in chunk])
+ function_ids = [result['nearest_neighbor_id'] for result in chunk]
  log_info(f"RevEng.AI | Cancelled: {self.cancelled.is_set()}")
  if self.cancelled.is_set():
  return []
 
  with self.config.create_api_client() as api_client:
- api_instance = revengai.FunctionsDataTypesApi(api_client)
- function_data_types_params = revengai.FunctionDataTypesParams.from_dict({"function_ids": function_ids}) 
- api_response = api_instance.generate_function_data_types_for_functions(function_data_types_params)
-
+ api_instance = revengai.DataTypesApi(api_client)
+ api_response = api_instance.v3_list_function_signatures(
+ function_ids=function_ids,
+ include_data_types=True,
+ ).to_dict()
+
  log_info(f"RevEng.AI | Cancelled: {self.cancelled.is_set()}")
  if self.cancelled.is_set():
  return []
- signatures = []
- items = []
- while True:
- if self.cancelled.is_set():
- return []
- with self.config.create_api_client() as api_client:
- api_instance = revengai.FunctionsDataTypesApi(api_client)
- api_response = api_instance.list_function_data_types_for_functions(function_ids=function_ids).to_dict()
 
- data = api_response.get("data", {})
- items = data.get("items", [])
- pending_count = sum(1 for item in items if item.get("status") == "pending")
- log_info(f"RevEng.AI | [Chunk {chunk_index}] {pending_count} items still pending...")
- if not pending_count:
- break
- time.sleep(3)
+ data_types = {}
+ for group in api_response.get("data_types") or []:
+ for entry in group.get("items") or []:
+ data_types[str(entry["data_type_id"])] = entry
 
- for item in items:
+ signatures = []
+ for item in api_response.get("items") or []:
  if self.cancelled.is_set():
  return []
- if item['status'] != "completed":
+ signature_data = build_signature_data(item, data_types)
+ if signature_data is None:
  continue
  for result in chunk:
  if result['nearest_neighbor_id'] == item['function_id']:
- signature = "N/A"
- item2 = item.get("data_types", {})
- func_types = item2.get("func_types", None)
- func_deps = item2.get("func_deps", [])
- if func_types is not None:
- fnc: Function = _art_from_dict(func_types)
- if fnc.name is None:
- log_info(f"Function {item['function_id']} has no name, skipping signature application.")
- continue
- log_info(f"Applying signature for {fnc.name}")
- signature = self.function_to_str(fnc)
- if signature != "N/A":
- signatures.append({"nearest_neighbor_id": result['nearest_neighbor_id'], "signature": signature, "data_types": item['data_types'], "signature_data": {"deps": func_deps, "function": fnc}})
+ fnc: Function = signature_data["function"]
+ log_info(f"Applying signature for {fnc.name}")
+ signature = self.function_to_str(fnc)
+ signatures.append({
+ "nearest_neighbor_id": result['nearest_neighbor_id'],
+ "signature": signature,
+ "data_types": data_types,
+ "signature_data": signature_data,
+ })
  break
 
- #log_info(f"RevEng.AI | Total count: {total_count}")
- #log_info(f"RevEng.AI | Total data types: {total_data_types}")
- #log_info(f"RevEng.AI | Items: {items}")
-
  return signatures
  except Exception as e:
  log_error(f"RevEng.AI | Error processing data type batch: {str(e)}")
  return []
-
+
  def function_arguments(self, fnc: Function) -> list[str]:
  args = []
  for k in fnc.header.args:
@@ -234,23 +219,6 @@ def function_to_str(self, fnc: Function) -> str:
  # convert the signature to a string representation
  return f"{fnc.type} {fnc.name}"\
  f"({', '.join(self.function_arguments(fnc))})"
-
- def make_signature(self, data_types: List[Dict]) -> str:
- try:
- #log_info(f"RevEng.AI | Making signature for {data_types}")
- signature = "("
- for _, arg in data_types['func_types'].get('header', {}).get('args', {}).items():
- #log_info(f"RevEng.AI | Arg: {arg}")
- signature += f"{arg.get('type', 'N/A')}, "
- signature = signature[:-2] if signature.endswith(", ") else signature
-
- signature += f") {data_types['func_types'].get('type', 'N/A')}"
-
- log_info(f"RevEng.AI | Signature: {signature}")
- return signature
- except Exception as e:
- log_error(f"RevEng.AI | Error making signature: {str(e)}")
- return "N/A"
 
  def fetch_data_types(self, bv: BinaryView, selected_results: List[Dict]) -> Tuple[bool, Dict[str, Any]]:
  try: