diff --git a/.gitignore b/.gitignore index 966032f..c829b83 100755 --- a/.gitignore +++ b/.gitignore @@ -54,4 +54,8 @@ Thumbs.db reai_toolkit/vendor/ reai_toolkit/vendor/* -plugin_binary_ninja.egg-info/* \ No newline at end of file +plugin_binary_ninja.egg-info/* + +.dev-sdk/ +# Written by `npx @openapitools/openapi-generator-cli version-manager set`. +openapitools.json \ No newline at end of file diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..f7c8a89 --- /dev/null +++ b/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.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. diff --git a/pyproject.toml b/pyproject.toml index c50fb47..1cc4f0d 100644 --- a/pyproject.toml +++ b/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", diff --git a/reai_toolkit/features/match_current_function/match_current_function.py b/reai_toolkit/features/match_current_function/match_current_function.py index 5b73277..75653fd 100755 --- a/reai_toolkit/features/match_current_function/match_current_function.py +++ b/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" diff --git a/reai_toolkit/features/match_functions/match_functions.py b/reai_toolkit/features/match_functions/match_functions.py index 2f8ff58..1e25da7 100755 --- a/reai_toolkit/features/match_functions/match_functions.py +++ b/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" diff --git a/reai_toolkit/features/view_function_in_portal/view_function_in_portal.py b/reai_toolkit/features/view_function_in_portal/view_function_in_portal.py index 409562e..80e54a5 100755 --- a/reai_toolkit/features/view_function_in_portal/view_function_in_portal.py +++ b/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) diff --git a/reai_toolkit/utils/core/binary_ninja.py b/reai_toolkit/utils/core/binary_ninja.py index a68ab30..5dda2a2 100755 --- a/reai_toolkit/utils/core/binary_ninja.py +++ b/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}") diff --git a/reai_toolkit/utils/features/datatypes.py b/reai_toolkit/utils/features/datatypes.py index c26678d..22cb3e7 100755 --- a/reai_toolkit/utils/features/datatypes.py +++ b/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}") \ No newline at end of file + 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())} diff --git a/reai_toolkit/utils/features/matching.py b/reai_toolkit/utils/features/matching.py index 8e58afc..bc08180 100755 --- a/reai_toolkit/utils/features/matching.py +++ b/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: diff --git a/scripts/gen_dev_sdk.sh b/scripts/gen_dev_sdk.sh new file mode 100755 index 0000000..fb0ac2a --- /dev/null +++ b/scripts/gen_dev_sdk.sh @@ -0,0 +1,71 @@ +#!/usr/bin/env bash +# Generates a local, gitignored `revengai` SDK build (.dev-sdk/) from an OpenAPI spec, using +# sdk-python's own generator config fetched fresh from GitHub at `main`. This is a stopgap for +# endpoints that exist server-side but haven't reached a published `revengai` release yet +# (https://docs.reveng.ai/openapi.json only reflects what's currently live). +# +# Usage: +# scripts/gen_dev_sdk.sh +# scripts/gen_dev_sdk.sh --url +# +# Requires `openapi-generator-cli` (via `npx @openapitools/openapi-generator-cli`, which in +# turn requires a Java runtime) and `git`. +set -euo pipefail + +SDK_PYTHON_REPO="https://github.com/RevEngAI/sdk-python.git" +# version-manager wants the bare Maven version, not the GitHub release tag's "v" prefix. +GENERATOR_VERSION="7.17.0" +OUTPUT_DIR=".dev-sdk" + +usage() { + echo "Usage: $0 | --url " >&2 + exit 1 +} + +if [[ $# -ne 1 && $# -ne 2 ]]; then + usage +fi + +if [[ "$1" == "--url" ]]; then + [[ $# -eq 2 ]] || usage + SPEC="$2" +else + [[ $# -eq 1 ]] || usage + SPEC="$(cd "$(dirname "$1")" && pwd)/$(basename "$1")" +fi + +WORK_DIR="$(mktemp -d)" +trap 'rm -rf "$WORK_DIR"' EXIT + +echo "Fetching sdk-python's generator config from $SDK_PYTHON_REPO@main..." +git clone --quiet --depth 1 --branch main "$SDK_PYTHON_REPO" "$WORK_DIR/sdk-python" + +# So the generated package's own version compares >= pyproject.toml's `revengai` pin (schema-pin +# tests check this) - mirrors sdk-python's own check.yaml, which stamps packageVersion from the +# spec's info.version the same way. +SPEC_VERSION="$(python3 -c " +import json, urllib.request +spec = '$SPEC' +if spec.startswith('http://') or spec.startswith('https://'): + data = json.load(urllib.request.urlopen(spec)) +else: + data = json.load(open(spec)) +print(data['info']['version']) +")" + +rm -rf "$OUTPUT_DIR" + +npx --yes @openapitools/openapi-generator-cli version-manager set "$GENERATOR_VERSION" + +npx --yes @openapitools/openapi-generator-cli generate \ + --generator-name python \ + --input-spec "$SPEC" \ + --config "$WORK_DIR/sdk-python/config.yml" \ + --template-dir "$WORK_DIR/sdk-python/templates" \ + --additional-properties="packageVersion=$SPEC_VERSION" \ + --output "$OUTPUT_DIR" + +echo +echo "Dev SDK generated at $OUTPUT_DIR." +echo "Install it over the pinned revengai version with:" +echo " uv pip install -e $OUTPUT_DIR --python .venv" diff --git a/tests/headless/test_match_current_function.py b/tests/headless/test_match_current_function.py index e710545..3ff53ad 100644 --- a/tests/headless/test_match_current_function.py +++ b/tests/headless/test_match_current_function.py @@ -18,21 +18,6 @@ def test_match_current_function_pipeline(bv, mocker): feature = mcf_mod.MatchCurrentFunction(config) mocker.patch.object(mcf_mod.time, "sleep") - meta_api = mocker.patch.object( - mcf_mod.revengai, "AnalysesResultsMetadataApi" - ).return_value - meta_api.get_functions_list.return_value.to_dict.return_value = { - "data": { - "functions": [ - { - "function_id": 100, - "function_vaddr": func.start, - "function_name": func.name, - } - ] - } - } - matched = MagicMock() matched.function_name = "match_name" matched.mangled_name = "_Z10match_namev" @@ -45,6 +30,15 @@ def test_match_current_function_pipeline(bv, mocker): by_distance.function_id = 100 by_distance.matched_functions = [matched] core_api = mocker.patch.object(mcf_mod.revengai, "FunctionsCoreApi").return_value + core_api.list_analysis_functions.return_value.to_dict.return_value = { + "functions": [ + { + "function_id": 100, + "function_vaddr": func.start, + "function_name": func.name, + } + ] + } core_api.get_functions_matching_status.return_value = MagicMock( status="COMPLETED", step_index=1, steps_total=1 ) diff --git a/tests/unit/binary_ninja/test_service.py b/tests/unit/binary_ninja/test_service.py new file mode 100644 index 0000000..f4d201b --- /dev/null +++ b/tests/unit/binary_ninja/test_service.py @@ -0,0 +1,59 @@ +from unittest.mock import MagicMock + +import pytest + +pytest.importorskip("binaryninja") + +import revengai + +from reai_toolkit.utils.core import binary_ninja as bn_mod + + +def test_get_function_id_by_addr_uses_list_analysis_functions(mocker): + config = MagicMock() + config.get_analysis_id.return_value = 42 + old_api = mocker.patch.object(revengai, "AnalysesResultsMetadataApi").return_value + old_api.get_functions_list.side_effect = AssertionError( + "get_function_id_by_addr must look up functions via " + "FunctionsCoreApi.list_analysis_functions, not the retired " + "AnalysesResultsMetadataApi.get_functions_list" + ) + functions_api = mocker.patch.object(revengai, "FunctionsCoreApi").return_value + entry = MagicMock(function_vaddr=0x1000, function_id=7) + functions_api.list_analysis_functions.return_value.functions = [entry] + bv = MagicMock() + + function_id = bn_mod.get_function_id_by_addr(bv, 0x1000, config) + + functions_api.list_analysis_functions.assert_called_once_with(analysis_id=42) + assert function_id == 7 + + +def test_get_function_id_by_addr_raises_when_address_not_found(mocker): + config = MagicMock() + config.get_analysis_id.return_value = 42 + old_api = mocker.patch.object(revengai, "AnalysesResultsMetadataApi").return_value + old_api.get_functions_list.return_value.data.functions = [] + functions_api = mocker.patch.object(revengai, "FunctionsCoreApi").return_value + functions_api.list_analysis_functions.return_value.functions = [] + bv = MagicMock() + + with pytest.raises(Exception, match="Function not found at address"): + bn_mod.get_function_id_by_addr(bv, 0x1000, config) + + +def test_rename_in_portal_uses_rename_function(mocker): + config = MagicMock() + renaming_api = mocker.patch.object( + revengai, "FunctionsRenamingHistoryApi" + ).return_value + + bn_mod._rename_in_portal(config, 7, "new_name", "new_mangled_name") + + renaming_api.rename_function.assert_called_once() + call_kwargs = renaming_api.rename_function.call_args.kwargs + assert call_kwargs["function_id"] == 7 + body = call_kwargs["rename_input_body"] + assert isinstance(body, revengai.RenameInputBody) + assert body.new_name == "new_name" + assert body.new_mangled_name == "new_mangled_name" diff --git a/tests/unit/datatypes/test_service.py b/tests/unit/datatypes/test_service.py new file mode 100644 index 0000000..cd1e3f5 --- /dev/null +++ b/tests/unit/datatypes/test_service.py @@ -0,0 +1,173 @@ +import pytest + +pytest.importorskip("binaryninja") + +from libbs.artifacts import Enum, Struct, Typedef, _art_from_dict + +from reai_toolkit.utils.features import datatypes as dt_mod + +PRIMITIVE_DATA_TYPES = { + "1": {"data_type_id": 1, "kind": "BASE", "name": "int"}, + "2": {"data_type_id": 2, "kind": "BASE", "name": "char"}, +} + +STRUCT_DATA_TYPES = { + "1": {"data_type_id": 1, "kind": "BASE", "name": "int"}, + "2": { + "data_type_id": 2, + "kind": "STRUCT", + "name": "point_t", + "size": 8, + "definition": { + "members": [ + {"name": "x", "offset": 0, "size": 4, "data_type_id": 1}, + {"name": "y", "offset": 4, "size": 4, "data_type_id": 1}, + ] + }, + }, +} + +ENUM_DATA_TYPES = { + "1": {"data_type_id": 1, "kind": "BASE", "name": "int"}, + "3": { + "data_type_id": 3, + "kind": "ENUM", + "name": "color_t", + "definition": { + "values": [ + {"name": "RED", "value": "0"}, + {"name": "GREEN", "value": "1"}, + ] + }, + }, +} + +TYPEDEF_OF_STRUCT_DATA_TYPES = { + **STRUCT_DATA_TYPES, + "4": { + "data_type_id": 4, + "kind": "TYPEDEF", + "name": "point_ptr_t", + "definition": {"target_data_type_id": 2}, + }, +} + + +def test_build_signature_data_resolves_primitive_parameters_and_return_type(): + signature = { + "function_id": 100, + "function_name": "count_labels", + "has_signature": True, + "calling_convention": "cdecl", + "parameters": [ + {"name": "count", "ordinal": 0, "data_type_id": 1}, + {"name": "label", "ordinal": 1, "data_type_id": 2}, + ], + "return_data_type_id": 1, + "source_function_id": 55, + "source_type": "similar_function", + } + + result = dt_mod.build_signature_data(signature, PRIMITIVE_DATA_TYPES) + + assert result is not None + function = result["function"] + assert function.name == "count_labels" + assert function.type == "int" + assert function.args[0].name == "count" + assert function.args[0].type == "int" + assert function.args[1].name == "label" + assert function.args[1].type == "char" + assert result["deps"] == [] + + +def test_build_signature_data_returns_none_when_signature_unavailable(): + signature = { + "function_id": 100, + "has_signature": False, + "parameters": [], + "return_data_type_id": None, + } + + assert dt_mod.build_signature_data(signature, {}) is None + + +def test_build_signature_data_includes_struct_dependency_for_struct_parameter(): + signature = { + "function_id": 101, + "function_name": "consume_point", + "has_signature": True, + "parameters": [{"name": "p", "ordinal": 0, "data_type_id": 2}], + "return_data_type_id": 1, + } + + result = dt_mod.build_signature_data(signature, STRUCT_DATA_TYPES) + + assert result["function"].args[0].type == "point_t" + assert len(result["deps"]) == 1 + + dependency = _art_from_dict(result["deps"][0]) + assert isinstance(dependency, Struct) + assert dependency.name == "point_t" + assert {member.name for member in dependency.members.values()} == {"x", "y"} + + +def test_build_signature_data_includes_enum_dependency_for_enum_return_type(): + signature = { + "function_id": 102, + "function_name": "current_color", + "has_signature": True, + "parameters": [], + "return_data_type_id": 3, + } + + result = dt_mod.build_signature_data(signature, ENUM_DATA_TYPES) + + assert result["function"].type == "color_t" + assert len(result["deps"]) == 1 + + dependency = _art_from_dict(result["deps"][0]) + assert isinstance(dependency, Enum) + assert dependency.name == "color_t" + assert dict(dependency.members) == {"RED": 0, "GREEN": 1} + + +def test_build_signature_data_recursively_resolves_typedef_of_struct(): + signature = { + "function_id": 103, + "function_name": "consume_point_ptr", + "has_signature": True, + "parameters": [{"name": "p", "ordinal": 0, "data_type_id": 4}], + "return_data_type_id": 1, + } + + result = dt_mod.build_signature_data(signature, TYPEDEF_OF_STRUCT_DATA_TYPES) + + assert result["function"].args[0].type == "point_ptr_t" + + dependencies = [_art_from_dict(dep) for dep in result["deps"]] + assert {type(dependency) for dependency in dependencies} == {Typedef, Struct} + + typedef = next(d for d in dependencies if isinstance(d, Typedef)) + assert typedef.name == "point_ptr_t" + assert typedef.type == "point_t" + + struct = next(d for d in dependencies if isinstance(d, Struct)) + assert struct.name == "point_t" + + +def test_build_signature_data_dedupes_shared_dependency(): + signature = { + "function_id": 104, + "function_name": "consume_two_points", + "has_signature": True, + "parameters": [ + {"name": "a", "ordinal": 0, "data_type_id": 2}, + {"name": "b", "ordinal": 1, "data_type_id": 2}, + ], + "return_data_type_id": 1, + } + + result = dt_mod.build_signature_data(signature, STRUCT_DATA_TYPES) + + assert len(result["deps"]) == 1 diff --git a/tests/unit/match_current_function/test_service.py b/tests/unit/match_current_function/test_service.py new file mode 100644 index 0000000..d834ce7 --- /dev/null +++ b/tests/unit/match_current_function/test_service.py @@ -0,0 +1,49 @@ +from unittest.mock import MagicMock + +import pytest + +pytest.importorskip("binaryninja") + +import revengai + +from reai_toolkit.features.match_current_function import ( + match_current_function as mcf_mod, +) + + +def test_match_current_function_lists_analyzed_functions_via_list_analysis_functions( + mocker, +): + mocker.patch("time.sleep") + config = MagicMock() + config.get_analysis_id.return_value = 42 + + old_api = mocker.patch.object(revengai, "AnalysesResultsMetadataApi").return_value + old_api.get_functions_list.side_effect = AssertionError( + "match_functions must list analyzed functions via " + "FunctionsCoreApi.list_analysis_functions, not the retired " + "AnalysesResultsMetadataApi.get_functions_list" + ) + + entries = [{"function_id": 7, "function_vaddr": 0x1000, "function_name": "foo"}] + functions_api = mocker.patch.object(revengai, "FunctionsCoreApi").return_value + functions_api.list_analysis_functions.return_value.functions = entries + functions_api.list_analysis_functions.return_value.to_dict.return_value = { + "functions": entries + } + status = MagicMock(step_index=1, steps_total=1) + status.status = revengai.TaskStatus.COMPLETED + functions_api.get_functions_matching_status.return_value = status + functions_api.get_functions_matches.return_value = MagicMock(matches=[]) + + bv = MagicMock() + target_function = MagicMock(start=0x1000, name="foo") + bv.get_functions_containing.return_value = [target_function] + + feature = mcf_mod.MatchCurrentFunction(config) + ok, result = feature.match_functions(bv, {"function": 0x1000}) + + functions_api.list_analysis_functions.assert_called_once() + call = functions_api.list_analysis_functions.call_args + assert 42 in call.args or call.kwargs.get("analysis_id") == 42 + assert ok is True diff --git a/tests/unit/match_functions/test_service.py b/tests/unit/match_functions/test_service.py index 09e9a85..0ac90fd 100644 --- a/tests/unit/match_functions/test_service.py +++ b/tests/unit/match_functions/test_service.py @@ -4,6 +4,8 @@ pytest.importorskip("binaryninja") +import revengai + from reai_toolkit.features.match_functions import match_functions as mf_mod @@ -43,3 +45,42 @@ def test_process_rename_batch_skips_invalid_address(mocker): renamed, _ = feature._process_rename_batch(MagicMock(), chunk, MagicMock()) assert renamed == 0 + + +def test_match_functions_lists_analyzed_functions_via_list_analysis_functions(mocker): + mocker.patch("time.sleep") + config = MagicMock() + config.get_analysis_id.return_value = 42 + + old_api = mocker.patch.object(revengai, "AnalysesResultsMetadataApi").return_value + old_api.get_functions_list.side_effect = AssertionError( + "match_functions must list analyzed functions via " + "FunctionsCoreApi.list_analysis_functions, not the retired " + "AnalysesResultsMetadataApi.get_functions_list" + ) + + entries = [{"function_id": 7, "function_vaddr": 0x1000, "function_name": "foo"}] + functions_api = mocker.patch.object(revengai, "FunctionsCoreApi").return_value + functions_api.list_analysis_functions.return_value.functions = entries + functions_api.list_analysis_functions.return_value.to_dict.return_value = { + "functions": entries + } + + core_api = mocker.patch.object(revengai, "AnalysesCoreApi").return_value + status = MagicMock(step_index=1, steps_total=1) + status.status = revengai.TaskStatus.COMPLETED + core_api.get_analysis_function_matching_status.return_value = status + core_api.get_analysis_function_matches.return_value = MagicMock(matches=[]) + + bv = MagicMock() + matched_function = MagicMock(start=0x1000) + bv.functions = [matched_function] + + feature = mf_mod.MatchFunctions(config) + ok, result = feature.match_functions(bv, {}) + + functions_api.list_analysis_functions.assert_called_once() + call = functions_api.list_analysis_functions.call_args + assert 42 in call.args or call.kwargs.get("analysis_id") == 42 + assert ok is True + assert result["skipped"] == 0 diff --git a/tests/unit/sdk/test_sdk_schemas.py b/tests/unit/sdk/test_sdk_schemas.py index 2038b80..3e49bc3 100644 --- a/tests/unit/sdk/test_sdk_schemas.py +++ b/tests/unit/sdk/test_sdk_schemas.py @@ -3,10 +3,8 @@ import pytest import revengai +from revengai.models.analysis_function_entry import AnalysisFunctionEntry from revengai.models.analysis_function_mapping import AnalysisFunctionMapping -from revengai.models.app_api_rest_v2_functions_types_function import ( - AppApiRestV2FunctionsTypesFunction, -) from revengai.models.basic import Basic from revengai.models.binary_search_result import BinarySearchResult from revengai.models.comments_data import CommentsData @@ -16,12 +14,13 @@ from revengai.models.function_match import FunctionMatch from revengai.models.inline_comment import InlineComment from revengai.models.matched_function import MatchedFunction +from revengai.models.rename_input_body import RenameInputBody from revengai.models.status_output import StatusOutput from revengai.models.summary_data import SummaryData from revengai.models.task_status import TaskStatus from revengai.models.workflow_progress import WorkflowProgress -PINNED = (3, 123, 0) +PINNED = (4, 4, 0) API_METHODS = { "ConfigApi": ["get_config"], @@ -38,17 +37,14 @@ "get_analysis_function_matches", ], "ModelsApi": ["get_models"], - "AnalysesResultsMetadataApi": ["get_functions_list"], "FunctionsCoreApi": [ + "list_analysis_functions", "start_functions_matching", "get_functions_matching_status", "get_functions_matches", ], - "FunctionsRenamingHistoryApi": ["rename_function_id"], - "FunctionsDataTypesApi": [ - "generate_function_data_types_for_functions", - "list_function_data_types_for_functions", - ], + "FunctionsRenamingHistoryApi": ["rename_function"], + "DataTypesApi": ["v3_list_function_signatures"], "FunctionsAIDecompilationApi": [ "get_ai_decompilation_status", "create_ai_decompilation", @@ -127,14 +123,28 @@ def test_upload_file_accepts_plugin_kwargs(): def test_create_analysis_accepts_request_kwarg(): - assert "analysis_create_request" in _params(revengai.AnalysesCoreApi.create_analysis) + assert "analysis_create_request" in _params( + revengai.AnalysesCoreApi.create_analysis + ) def test_analysis_id_methods_accept_analysis_id(): - for method in ("get_analysis_status", "get_analysis_basic_info", "get_analysis_function_map"): + for method in ( + "get_analysis_status", + "get_analysis_basic_info", + "get_analysis_function_map", + ): assert "analysis_id" in _params(getattr(revengai.AnalysesCoreApi, method)) +def test_list_analysis_functions_accepts_analysis_id(): + assert "analysis_id" in _params(revengai.FunctionsCoreApi.list_analysis_functions) + + +def test_list_analysis_functions_output_has_no_envelope(): + assert {"functions"} <= set(revengai.ListAnalysisFunctionsOutputBody.model_fields) + + def test_start_functions_matching_accepts_request_kwargs(): params = _params(revengai.FunctionsCoreApi.start_functions_matching) assert "start_matching_for_functions_input_body" in params @@ -151,13 +161,16 @@ def test_start_analysis_function_matching_accepts_request_kwargs(): def test_analysis_matching_status_and_results_accept_analysis_id(): - for method in ("get_analysis_function_matching_status", "get_analysis_function_matches"): + for method in ( + "get_analysis_function_matching_status", + "get_analysis_function_matches", + ): assert "analysis_id" in _params(getattr(revengai.AnalysesCoreApi, method)) -def test_rename_function_id_accepts_kwargs(): - params = _params(revengai.FunctionsRenamingHistoryApi.rename_function_id) - assert {"function_id", "function_rename"} <= params +def test_rename_function_accepts_kwargs(): + params = _params(revengai.FunctionsRenamingHistoryApi.rename_function) + assert {"function_id", "rename_input_body"} <= params def test_ai_decompilation_methods_accept_function_id(): @@ -168,7 +181,9 @@ def test_ai_decompilation_methods_accept_function_id(): "get_ai_decompilation_summary", "get_ai_decompilation_inline_comments", ): - assert "function_id" in _params(getattr(revengai.FunctionsAIDecompilationApi, method)) + assert "function_id" in _params( + getattr(revengai.FunctionsAIDecompilationApi, method) + ) def test_analysis_create_request_has_plugin_fields(): @@ -188,8 +203,8 @@ def test_match_filters_has_plugin_fields(): ) -def test_function_rename_has_plugin_fields(): - assert {"new_name", "new_mangled_name"} <= set(revengai.FunctionRename.model_fields) +def test_rename_input_body_has_plugin_fields(): + assert {"new_name", "new_mangled_name"} <= set(RenameInputBody.model_fields) def test_tag_has_name_field(): @@ -241,7 +256,7 @@ def test_function_mapping_has_plugin_fields(): def test_analysis_functions_item_has_plugin_fields(): assert {"function_id", "function_vaddr", "function_name"} <= set( - AppApiRestV2FunctionsTypesFunction.model_fields + AnalysisFunctionEntry.model_fields ) @@ -343,3 +358,40 @@ def test_match_current_function_request_shape_is_honored(): assert payload["function_ids"] == [42] assert payload["filters"]["collection_ids"] == [1] assert payload["filters"]["binary_ids"] == [2] + + +def test_v3_list_function_signatures_accepts_plugin_kwargs(): + assert {"function_ids", "include_data_types"} <= _params( + revengai.DataTypesApi.v3_list_function_signatures + ) + + +def test_batch_function_signature_entry_has_plugin_fields(): + from revengai.models.batch_function_signature_entry import ( + BatchFunctionSignatureEntry, + ) + + assert { + "function_id", + "function_name", + "has_signature", + "calling_convention", + "parameters", + "return_data_type_id", + "source_function_id", + "source_type", + } <= set(BatchFunctionSignatureEntry.model_fields) + + +def test_list_function_signatures_output_has_items_and_data_types_catalogue(): + from revengai.models.list_function_signatures_output_body import ( + ListFunctionSignaturesOutputBody, + ) + + assert {"items", "data_types"} <= set(ListFunctionSignaturesOutputBody.model_fields) + + +def test_analysis_data_types_group_has_analysis_id_and_items(): + from revengai.models.analysis_data_types_group import AnalysisDataTypesGroup + + assert {"analysis_id", "items"} <= set(AnalysisDataTypesGroup.model_fields) diff --git a/tests/unit/view_function_in_portal/test_service.py b/tests/unit/view_function_in_portal/test_service.py new file mode 100644 index 0000000..ed6dbc1 --- /dev/null +++ b/tests/unit/view_function_in_portal/test_service.py @@ -0,0 +1,44 @@ +from unittest.mock import MagicMock + +import pytest + +pytest.importorskip("binaryninja") + +import revengai + +from reai_toolkit.features.view_function_in_portal import ( + view_function_in_portal as vfp_mod, +) + + +def test_view_function_in_portal_uses_list_analysis_functions(mocker): + config = MagicMock() + config.get_analysis_id.return_value = 42 + config.portal_url = "https://portal.reveng.ai" + + old_api = mocker.patch.object(revengai, "AnalysesResultsMetadataApi").return_value + old_api.get_functions_list.side_effect = AssertionError( + "view_function_in_portal must look up functions via " + "FunctionsCoreApi.list_analysis_functions, not the retired " + "AnalysesResultsMetadataApi.get_functions_list" + ) + functions_api = mocker.patch.object(revengai, "FunctionsCoreApi").return_value + entry = MagicMock(function_vaddr=0x2000, function_id=99) + functions_api.list_analysis_functions.return_value.functions = [entry] + + interaction_cls = mocker.patch.object(vfp_mod, "InteractionHandler") + + bv = MagicMock() + target_function = MagicMock(start=0x2000) + bv.get_functions_containing.return_value = [target_function] + + feature = vfp_mod.ViewFunctionInPortal(config) + ok, _ = feature.view_function_in_portal(bv, {"function": 0x2000}) + + functions_api.list_analysis_functions.assert_called_once() + call = functions_api.list_analysis_functions.call_args + assert 42 in call.args or call.kwargs.get("analysis_id") == 42 + assert ok is True + interaction_cls.return_value.open_url.assert_called_once_with( + "https://portal.reveng.ai/analyses/42?fn=99&view=matching&matchingMode=single" + ) diff --git a/uv.lock b/uv.lock index 23d6d5e..899cd71 100644 --- a/uv.lock +++ b/uv.lock @@ -345,7 +345,7 @@ requires-dist = [ { name = "pycparser", specifier = ">=2.22,<3" }, { name = "pydantic", specifier = ">=2.12.5" }, { name = "requests", specifier = ">=2.32" }, - { name = "revengai", specifier = ">=3.123.0" }, + { name = "revengai", specifier = ">=4.4.0" }, { name = "urllib3", specifier = ">=2.0.0,<2.3.0" }, ] @@ -674,7 +674,7 @@ wheels = [ [[package]] name = "revengai" -version = "3.123.0" +version = "4.4.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "lazy-imports" }, @@ -683,9 +683,9 @@ dependencies = [ { name = "typing-extensions" }, { name = "urllib3" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/e4/11/c263645fd280bb0d4c0a3cb3057880570bd8d6581b71be0f688f7adc6dee/revengai-3.123.0.tar.gz", hash = "sha256:871199b75675ac54543005179e7f92b845cfceba79ee49c8f11db0d77613aa7f", size = 369727, upload-time = "2026-07-23T08:49:18.527Z" } +sdist = { url = "https://files.pythonhosted.org/packages/c9/76/64dffda3361860419abde2aafaddb92ec03e2e35facab25b60b0f7ec8411/revengai-4.4.0.tar.gz", hash = "sha256:880b2994b4cb803d2339fd9fd69c62e779222b34a8a018197f44be9e4e456728", size = 407106, upload-time = "2026-08-24T07:50:06.817Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/91/e1/5a01283b4eb34f54879ce7a91a2db81d27f987c0c1a5e33ce2e81c6ae297/revengai-3.123.0-py3-none-any.whl", hash = "sha256:f488e653bd88b9c8d432d062d11417710502b9bb10966c88c2c77f69e588fdb1", size = 966561, upload-time = "2026-07-23T08:49:17.023Z" }, + { url = "https://files.pythonhosted.org/packages/45/a3/3a441a0fd5cd2cd9603890ec64d7c307b866ef46228c0517ba79a0f05886/revengai-4.4.0-py3-none-any.whl", hash = "sha256:d88dd348fdfb041802f13d542a1f91b86c6712ddef252aa4bf6a9b0d73461847", size = 1048901, upload-time = "2026-08-24T07:50:05.311Z" }, ] [[package]]