From 94e3f9e715d401b4b4cf7828f9181ff4e185f4aa Mon Sep 17 00:00:00 2001 From: r0ny123 <49360849+r0ny123@users.noreply.github.com> Date: Tue, 8 Sep 2026 16:17:57 +0530 Subject: [PATCH] fix(data-types): migrate sync to v3 signatures API The plugin still calls deprecated v2 data-type operations while the current service exposes v3 signature endpoints. Add a compatibility adapter for the bundled SDK, preserve the existing IDA importer, update function headers through the v3 signature API, and skip unsupported stack-variable writes. --- .../services/data_types/data_types_service.py | 43 +- .../app/services/data_types/v3_data_types.py | 432 ++++++++++++++++++ .../variable_sync/variable_sync_service.py | 194 +++----- tests/unit/data_types/test_service.py | 158 +++---- tests/unit/data_types/test_v3_adapter.py | 99 ++++ tests/unit/variable_sync/test_service.py | 381 +++++---------- 6 files changed, 792 insertions(+), 515 deletions(-) create mode 100644 reai_toolkit/app/services/data_types/v3_data_types.py create mode 100644 tests/unit/data_types/test_v3_adapter.py diff --git a/reai_toolkit/app/services/data_types/data_types_service.py b/reai_toolkit/app/services/data_types/data_types_service.py index 58159fe..1dc5a83 100644 --- a/reai_toolkit/app/services/data_types/data_types_service.py +++ b/reai_toolkit/app/services/data_types/data_types_service.py @@ -7,19 +7,15 @@ from loguru import logger from reai_toolkit.app.core.netstore_service import SimpleNetStore -from revengai import ( - FunctionsDataTypesApi, - FunctionDataTypesList, - BaseResponseFunctionDataTypesList -) +from revengai import FunctionDataTypesList from reai_toolkit.app.interfaces.thread_service import IThreadService +from reai_toolkit.app.services.data_types.v3_data_types import ( + list_function_signatures, + to_legacy_function_data_types, +) from reai_toolkit.app.transformations.import_data_types import ImportDataTypes - -FUNCTION_IDS_BATCH_SIZE = 50 - - @dataclass class DataTypesImportResult: error: str | None = None @@ -60,12 +56,11 @@ def import_data_types( logger.error(f"RevEng.AI: failed to sync function data types: {e}") return DataTypesImportResult(error=f"Failed to sync function data types: {e}") - present_ids: set[int] = ( - {item.function_id for item in response.items if item.data_types is not None} - if response - else set() - ) - remote_absent_ids: set[int] = set(matched_function_ids) - present_ids + # The v3 API distinguishes "no extracted signature" from a missing + # function. A missing signature is not a reason to send the old v2 + # data-type blob back to the server, so do not turn it into a local + # push candidate. + remote_absent_ids: set[int] = set() apply_failed_ids: set[int] = set() if response: @@ -82,15 +77,11 @@ def _get_data_types(self, function_ids: list[int] | None = None) -> FunctionData if not function_ids: return None - items = [] with self.yield_api_client(sdk_config=self.sdk_config) as api_client: - client = FunctionsDataTypesApi(api_client=api_client) - for start in range(0, len(function_ids), FUNCTION_IDS_BATCH_SIZE): - chunk = function_ids[start:start + FUNCTION_IDS_BATCH_SIZE] - response: BaseResponseFunctionDataTypesList = ( - client.list_function_data_types_for_functions(function_ids=chunk) # type: ignore - ) - if response.status and response.data: - items.extend(response.data.items) - - return FunctionDataTypesList(items=items) + response = list_function_signatures( + api_client, + function_ids, + include_data_types=True, + ) + + return to_legacy_function_data_types(response) diff --git a/reai_toolkit/app/services/data_types/v3_data_types.py b/reai_toolkit/app/services/data_types/v3_data_types.py new file mode 100644 index 0000000..770faa3 --- /dev/null +++ b/reai_toolkit/app/services/data_types/v3_data_types.py @@ -0,0 +1,432 @@ +"""Small compatibility layer for the RevEng.AI v3 data-type API. + +The plugin release currently bundles the v2 SDK, while the service no longer +serves ``/v2/functions/data_types``. Keeping this adapter on top of the +bundled ``ApiClient`` lets the plugin use the v3 endpoints without replacing +the whole SDK (which would otherwise break the rest of the plugin's v2 model +imports). +""" + +from __future__ import annotations + +import json +from typing import Any, Iterable + +from revengai import ApiClient, ApiException +from revengai.models.argument import Argument +from revengai.models.enumeration import Enumeration +from revengai.models.function_data_types_list import FunctionDataTypesList +from revengai.models.function_data_types_list_item import FunctionDataTypesListItem +from revengai.models.structure import Structure +from revengai.models.structure_member import StructureMember +from revengai.models.type_definition import TypeDefinition +from revengai.models.v2_function_header import V2FunctionHeader +from revengai.models.v2_function_info import V2FunctionInfo +from revengai.models.v2_function_info_func_deps_inner import V2FunctionInfoFuncDepsInner +from revengai.models.v2_function_type import V2FunctionType + + +FUNCTION_IDS_BATCH_SIZE = 50 + + +def _as_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _request_json( + api_client: ApiClient, + method: str, + resource_path: str, + *, + query_params: Iterable[tuple[str, Any]] = (), + body: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Call a v3 endpoint using the SDK client shipped with the plugin.""" + + headers: dict[str, str] = {"Accept": "application/json"} + if body is not None: + headers["Content-Type"] = "application/json" + + request = api_client.param_serialize( + method=method, + resource_path=resource_path, + path_params={}, + query_params=list(query_params), + header_params=headers, + body=body, + post_params=[], + files={}, + auth_settings=["APIKey", "bearerAuth"], + collection_formats={"function_ids": "multi"}, + _host=None, + _request_auth=None, + ) + response = api_client.call_api(*request) + response.read() + + raw = response.data or b"" + text = raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else str(raw) + if not 200 <= response.status <= 299: + data: Any = None + try: + data = json.loads(text) if text else None + except json.JSONDecodeError: + pass + raise ApiException.from_response(http_resp=response, body=text, data=data) + + if not text: + return {} + decoded = json.loads(text) + if not isinstance(decoded, dict): + raise ValueError(f"Unexpected RevEng.AI response shape: {type(decoded).__name__}") + return decoded + + +def list_function_signatures( + api_client: ApiClient, + function_ids: list[int], + *, + include_data_types: bool = True, +) -> dict[str, Any]: + """Fetch v3 signatures, preserving the endpoint's grouped type data.""" + + if not function_ids: + return {"items": [], "data_types": []} + + all_items: list[dict[str, Any]] = [] + groups: dict[int, dict[str, Any]] = {} + for start in range(0, len(function_ids), FUNCTION_IDS_BATCH_SIZE): + chunk = function_ids[start : start + FUNCTION_IDS_BATCH_SIZE] + response = _request_json( + api_client, + "GET", + "/v3/functions/signatures", + query_params=[ + ("function_ids", chunk), + ("include_data_types", include_data_types), + ], + ) + all_items.extend(response.get("items") or []) + for group in response.get("data_types") or []: + analysis_id = _as_int(group.get("analysis_id"), -1) + if analysis_id < 0: + continue + existing = groups.setdefault(analysis_id, {"analysis_id": analysis_id, "items": []}) + existing["items"].extend(group.get("items") or []) + + return {"items": all_items, "data_types": list(groups.values())} + + +def update_function_signature( + api_client: ApiClient, + analysis_id: int, + function_id: int, + body: dict[str, Any], +) -> dict[str, Any]: + """Update one function signature through the v3 API.""" + + return _request_json( + api_client, + "PUT", + f"/v3/analyses/{int(analysis_id)}/functions/{int(function_id)}/signature", + body=body, + ) + + +class _TypeResolver: + """Resolve v3 data-type IDs into the C-like strings the IDA importer uses.""" + + def __init__(self, entries: Iterable[dict[str, Any]]) -> None: + self.entries: dict[int, dict[str, Any]] = { + _as_int(entry.get("data_type_id"), -1): entry + for entry in entries + if entry.get("data_type_id") is not None + } + self._cache: dict[int, str] = {} + self._resolving: set[int] = set() + + @staticmethod + def _qualified_name(entry: dict[str, Any]) -> str: + name = str(entry.get("name") or "").strip() + namespace = str(entry.get("namespace") or "").strip() + if namespace and name: + return f"{namespace}::{name}" + return name + + def size(self, data_type_id: Any) -> int: + entry = self.entries.get(_as_int(data_type_id, -1)) + return _as_int(entry.get("size"), 0) if entry else 0 + + def type_name(self, data_type_id: Any) -> str: + if data_type_id is None: + return "" + type_id = _as_int(data_type_id, -1) + if type_id < 0: + return "" + if type_id in self._cache: + return self._cache[type_id] + + entry = self.entries.get(type_id) + if entry is None: + return "" + if type_id in self._resolving: + return self._qualified_name(entry) or "void" + + self._resolving.add(type_id) + kind = str(entry.get("kind") or "").upper() + definition = entry.get("definition") or {} + name = self._qualified_name(entry) + + if kind == "POINTER": + target = definition.get("pointee_data_type_id") + target_name = self.type_name(target) or name or "void" + resolved = f"{target_name} *" + elif kind == "ARRAY": + element = self.type_name(definition.get("element_data_type_id")) or name or "void" + count = definition.get("count") + resolved = f"{element}[{count}]" if count is not None else f"{element}[]" + else: + resolved = name or "void" + + self._resolving.discard(type_id) + self._cache[type_id] = resolved + return resolved + + @staticmethod + def _dependency_name(entry: dict[str, Any]) -> str: + return _TypeResolver._qualified_name(entry) or "anonymous_type" + + def dependencies(self) -> list[V2FunctionInfoFuncDepsInner]: + dependencies: list[V2FunctionInfoFuncDepsInner] = [] + for entry in self.entries.values(): + kind = str(entry.get("kind") or "").upper() + name = self._dependency_name(entry) + definition = entry.get("definition") or {} + + if kind in {"STRUCT", "UNION"}: + members: dict[str, StructureMember] = {} + for index, member in enumerate(definition.get("members") or []): + offset = _as_int(member.get("offset"), index) + member_name = str(member.get("name") or f"field_{offset:x}") + members[str(offset)] = StructureMember.model_construct( + name=member_name, + offset=offset, + type=self.type_name(member.get("data_type_id")), + size=_as_int(member.get("size"), 0), + ) + dependency = Structure.model_construct( + name=name, + size=_as_int(entry.get("size"), 0), + members=members, + artifact_type="Structure", + ) + dependencies.append(V2FunctionInfoFuncDepsInner.model_construct(actual_instance=dependency)) + elif kind == "ENUM": + values: dict[str, int] = {} + for value in definition.get("values") or []: + value_name = str(value.get("name") or "") + if not value_name: + continue + values[value_name] = _as_int(value.get("value"), 0) + dependency = Enumeration.model_construct( + name=name, + members=values, + artifact_type="Enumeration", + ) + dependencies.append(V2FunctionInfoFuncDepsInner.model_construct(actual_instance=dependency)) + elif kind == "TYPEDEF": + dependency = TypeDefinition.model_construct( + name=name, + type=self.type_name(definition.get("target_data_type_id")), + artifact_type="Typedef", + ) + dependencies.append(V2FunctionInfoFuncDepsInner.model_construct(actual_instance=dependency)) + + return dependencies + + def ids_by_name(self) -> dict[str, int]: + result: dict[str, int] = {} + for type_id, entry in self.entries.items(): + names = { + self._qualified_name(entry), + str(entry.get("name") or "").strip(), + self.type_name(type_id), + } + for name in names: + if name: + result[name] = type_id + return result + + +def _legacy_function_info( + item: dict[str, Any], + resolver: _TypeResolver, + local_address: int = 0, +) -> V2FunctionInfo: + function_id = _as_int(item.get("function_id")) + function_name = str(item.get("function_name") or f"sub_{function_id:x}") + return_type = resolver.type_name(item.get("return_data_type_id")) or "void" + + args: dict[str, Argument] = {} + for parameter in item.get("parameters") or []: + ordinal = _as_int(parameter.get("ordinal"), len(args)) + data_type_id = parameter.get("data_type_id") + args[hex(ordinal)] = Argument.model_construct( + offset=ordinal, + name=str(parameter.get("name") or f"arg_{ordinal}"), + type=resolver.type_name(data_type_id), + size=resolver.size(data_type_id), + ) + + header = V2FunctionHeader.model_construct( + name=function_name, + addr=local_address, + type=return_type, + args=args, + ) + function_type = V2FunctionType.model_construct( + addr=local_address, + size=0, + header=header, + stack_vars=None, + name=function_name, + type=return_type, + artifact_type="Function", + ) + return V2FunctionInfo.model_construct( + func_types=function_type, + func_deps=resolver.dependencies(), + ) + + +def to_legacy_function_data_types( + response: dict[str, Any], + *, + local_addresses: dict[int, int] | None = None, +) -> FunctionDataTypesList: + """Adapt a v3 signature response to the plugin's existing IDA importer.""" + + local_addresses = local_addresses or {} + type_groups: dict[int, list[dict[str, Any]]] = { + _as_int(group.get("analysis_id"), -1): group.get("items") or [] + for group in response.get("data_types") or [] + } + + items: list[FunctionDataTypesListItem] = [] + with_types = 0 + for raw_item in response.get("items") or []: + function_id = _as_int(raw_item.get("function_id")) + has_signature = bool(raw_item.get("has_signature")) + analysis_id = _as_int(raw_item.get("analysis_id"), -1) + entries = type_groups.get(analysis_id, []) + resolver = _TypeResolver(entries) + data_types = ( + _legacy_function_info( + raw_item, + resolver, + local_address=_as_int(local_addresses.get(function_id), 0), + ) + if has_signature + else None + ) + if data_types is not None: + with_types += 1 + items.append( + FunctionDataTypesListItem.model_construct( + completed=True, + status="success" if has_signature else "not_available", + data_types=data_types, + data_types_version=None, + function_id=function_id, + ) + ) + + return FunctionDataTypesList.model_construct( + total_count=len(items), + total_data_types_count=with_types, + items=items, + ) + + +def resolver_for_function(response: dict[str, Any], function_id: int) -> tuple[dict[str, Any], _TypeResolver] | None: + """Return a raw v3 item and its analysis-scoped type resolver.""" + + item = next( + (candidate for candidate in response.get("items") or [] if _as_int(candidate.get("function_id")) == function_id), + None, + ) + if item is None: + return None + analysis_id = _as_int(item.get("analysis_id"), -1) + group = next( + (candidate for candidate in response.get("data_types") or [] if _as_int(candidate.get("analysis_id")) == analysis_id), + None, + ) + return item, _TypeResolver((group or {}).get("items") or []) + + +def normalise_type_text(value: str) -> str: + """Normalise equivalent C-like type spellings for ID lookup.""" + + text = " ".join(str(value or "").replace("*", " * ").split()) + for qualifier in ("const ", "volatile ", "struct ", "union ", "enum "): + text = text.replace(qualifier, "") + return text.replace(" *", "*").replace("* ", "*").strip() + + +def build_signature_update( + response: dict[str, Any], + function_id: int, + *, + function_info: Any, +) -> dict[str, Any] | None: + """Build a v3 signature update from the plugin's current local function.""" + + resolved = resolver_for_function(response, function_id) + if resolved is None: + return None + remote_item, resolver = resolved + if not remote_item.get("has_signature"): + return None + + ids_by_name = resolver.ids_by_name() + normalised_ids = {normalise_type_text(name): data_type_id for name, data_type_id in ids_by_name.items()} + function_type = getattr(function_info, "func_types", None) + header = getattr(function_type, "header", None) + if header is None: + return None + + def resolve_id(type_text: str | None) -> int | None: + normalised = normalise_type_text(type_text or "") + if not normalised or normalised == "void": + return None + return normalised_ids.get(normalised) + + parameters: list[dict[str, Any]] = [] + for index, argument in enumerate((header.args or {}).values()): + ordinal = _as_int(getattr(argument, "offset", index), index) + type_id = resolve_id(getattr(argument, "type", "")) + if type_id is None and getattr(argument, "type", ""): + return None + parameter: dict[str, Any] = {"ordinal": ordinal} + name = str(getattr(argument, "name", "") or "") + if name: + parameter["name"] = name + if type_id is not None: + parameter["data_type_id"] = type_id + parameters.append(parameter) + + body: dict[str, Any] = {"parameters": parameters} + return_type = str(getattr(header, "type", "") or "") + normalised_return_type = normalise_type_text(return_type) + if not normalised_return_type: + return None + if normalised_return_type != "void": + return_id = resolve_id(return_type) + if return_id is None: + return None + body["return_data_type_id"] = return_id + return body diff --git a/reai_toolkit/app/services/variable_sync/variable_sync_service.py b/reai_toolkit/app/services/variable_sync/variable_sync_service.py index 93aa089..71a82d3 100644 --- a/reai_toolkit/app/services/variable_sync/variable_sync_service.py +++ b/reai_toolkit/app/services/variable_sync/variable_sync_service.py @@ -10,14 +10,9 @@ from revengai import ( FunctionArgument, - BaseResponseFunctionDataTypesList, - BatchUpdateDataTypesInputBody, - BatchUpdateDataTypesItem, - BatchUpdateDataTypesOutputBody, Configuration, FunctionDependency, FunctionInfo, - FunctionsDataTypesApi, FunctionType, ) from revengai import ApiException @@ -26,6 +21,11 @@ from reai_toolkit.app.core.netstore_service import SimpleNetStore from reai_toolkit.app.interfaces.thread_service import IThreadService +from reai_toolkit.app.services.data_types.v3_data_types import ( + build_signature_update, + list_function_signatures, + update_function_signature, +) @execute_read @@ -62,8 +62,6 @@ class VariableSyncService(IThreadService): _q: queue.Queue[Tuple[int, int, object]] = queue.Queue() _last_ts: dict[tuple, float] = {} _debounce_ms: int = 400 - _max_retries: int = 3 - _type_batch_size: int = 50 def __init__(self, netstore_service: SimpleNetStore, sdk_config: Configuration): super().__init__(netstore_service=netstore_service, sdk_config=sdk_config) @@ -118,49 +116,44 @@ def _push_change(self, function_id: int, func_addr: int, artifact: object) -> No logger.debug("RevEng.AI: no analysis id; skipping data types push") return - for _ in range(self._max_retries): - fetched = self._fetch_function_info(function_id) - if fetched is None: - # First write for this function: build the object from current state. - info = self._build_function_info(func_addr) - if info is None: - logger.debug(f"RevEng.AI: could not build data types for function {function_id}; skipping push") - return - version = 0 - else: - info, version = fetched - if not self._patch(info, artifact): - logger.debug(f"RevEng.AI: no data types change for function {function_id}; skipping push") - return - - status: str = self._push(function_id, analysis_id, info, version) - # A concurrent edit moved the stored version on; re-fetch and reapply. - if status == "version_conflict": - time.sleep(0.2) - continue - if status == "updated": - logger.info(f"RevEng.AI: pushed data types for function {function_id}") - else: - logger.warning( - f"RevEng.AI: data types update for function {function_id} returned {status}" - ) - return - - def _fetch_function_info(self, function_id: int) -> Optional[Tuple[FunctionInfo, int]]: - with self.yield_api_client(sdk_config=self.sdk_config) as api_client: - client = FunctionsDataTypesApi(api_client) - resp: BaseResponseFunctionDataTypesList = ( - client.list_function_data_types_for_functions(function_ids=[function_id]) # type: ignore + # The v3 API has no equivalent of the old opaque function-data-types + # blob or its optimistic version field. It does support full function + # signature updates, but stack-variable edits have no v3 write + # endpoint. Do not send a false success for those edits. + if isinstance(artifact, StackVariable): + logger.debug( + "RevEng.AI: v3 API has no stack-variable write endpoint; " + f"skipping local stack variable sync for function {function_id}" ) + return - if not resp.status or not resp.data: - return None + info = self._build_function_info(func_addr) + if info is None: + logger.debug(f"RevEng.AI: could not build signature for function {function_id}; skipping push") + return - for item in resp.data.items: - if item.function_id == function_id and item.data_types is not None: - return item.data_types, (item.data_types_version or 0) + try: + with self.yield_api_client(sdk_config=self.sdk_config) as api_client: + response = list_function_signatures( + api_client, + [function_id], + include_data_types=True, + ) + body = build_signature_update(response, function_id, function_info=info) + if body is None: + logger.debug( + f"RevEng.AI: no v3 signature available for function {function_id}; skipping push" + ) + return + update_function_signature(api_client, analysis_id, function_id, body) + except ApiException as e: + logger.error(f"RevEng.AI: failed to push function signature: HTTP {e.status} {e.reason}") + return + except Exception as e: + logger.error(f"RevEng.AI: failed to push function signature: {e}") + return - return None + logger.info(f"RevEng.AI: pushed function signature for function {function_id}") def _build_function_info(self, func_addr: int) -> Optional[FunctionInfo]: if self._deci is None: @@ -259,27 +252,6 @@ def _base_type_name(type_str: Optional[str]) -> Optional[str]: tokens = [tok for tok in cleaned.split() if tok not in keywords] return tokens[-1] if tokens else None - def _push(self, function_id: int, analysis_id: int, info: FunctionInfo, version: int) -> str: - body = BatchUpdateDataTypesInputBody( - functions=[ - BatchUpdateDataTypesItem( - function_id=function_id, - data_types=info.to_dict(), - data_types_version=version, - ) - ] - ) - with self.yield_api_client(sdk_config=self.sdk_config) as api_client: - client = FunctionsDataTypesApi(api_client) - out: BatchUpdateDataTypesOutputBody = client.batch_update_function_data_types( - analysis_id=analysis_id, - batch_update_data_types_input_body=body, - ) - - if out.results: - return out.results[0].status - return "error" - def _patch(self, info: FunctionInfo, artifact: object) -> bool: ft = info.func_types if ft is None: @@ -339,74 +311,30 @@ def push_local_function_types_batch(self, targets: dict[int, int], analysis_id: return 0 base: int = self._deci.binary_base_addr - infos: dict[int, FunctionInfo] = {} + updated = 0 for function_id, ea in targets.items(): try: info = self._build_function_info(ea - base) - except Exception as e: - logger.debug(f"RevEng.AI: could not build local types for function {function_id}: {e}") - continue - if info is not None: - infos[function_id] = info - - if not infos: - return 0 - - versions: dict[int, int] = self._fetch_type_versions(list(infos.keys())) - pending: dict[int, FunctionInfo] = dict(infos) - updated: int = 0 - - for _ in range(self._max_retries): - if not pending: - break - conflicts: dict[int, FunctionInfo] = {} - items = list(pending.items()) - for start in range(0, len(items), self._type_batch_size): - chunk = items[start:start + self._type_batch_size] - for result in self._push_types_batch(analysis_id, chunk, versions): - if result.status == "updated": - updated += 1 - elif result.status == "version_conflict": - conflicts[result.function_id] = infos[result.function_id] - else: - logger.warning( - f"RevEng.AI: type push for function {result.function_id} returned {result.status}" - ) - pending = conflicts - if pending: - versions.update(self._fetch_type_versions(list(pending.keys()))) - - return updated - - def _fetch_type_versions(self, function_ids: list[int]) -> dict[int, int]: - versions: dict[int, int] = {} - with self.yield_api_client(sdk_config=self.sdk_config) as api_client: - client = FunctionsDataTypesApi(api_client) - for start in range(0, len(function_ids), self._type_batch_size): - chunk = function_ids[start:start + self._type_batch_size] - resp: BaseResponseFunctionDataTypesList = ( - client.list_function_data_types_for_functions(function_ids=chunk) # type: ignore + if info is None: + continue + with self.yield_api_client(sdk_config=self.sdk_config) as api_client: + response = list_function_signatures( + api_client, + [function_id], + include_data_types=True, + ) + body = build_signature_update(response, function_id, function_info=info) + if body is None: + continue + update_function_signature(api_client, analysis_id, function_id, body) + updated += 1 + except ApiException as e: + logger.warning( + f"RevEng.AI: skipped v3 signature push for function {function_id}: " + f"HTTP {e.status} {e.reason}" ) - if resp.status and resp.data: - for item in resp.data.items: - versions[item.function_id] = item.data_types_version or 0 - return versions - - def _push_types_batch(self, analysis_id: int, items: list, versions: dict[int, int]) -> list: - body = BatchUpdateDataTypesInputBody( - functions=[ - BatchUpdateDataTypesItem( - function_id=function_id, - data_types=info.to_dict(), - data_types_version=versions.get(function_id, 0), + except Exception as e: + logger.warning( + f"RevEng.AI: skipped v3 signature push for function {function_id}: {e}" ) - for function_id, info in items - ] - ) - with self.yield_api_client(sdk_config=self.sdk_config) as api_client: - client = FunctionsDataTypesApi(api_client) - out: BatchUpdateDataTypesOutputBody = client.batch_update_function_data_types( - analysis_id=analysis_id, - batch_update_data_types_input_body=body, - ) - return out.results or [] + return updated diff --git a/tests/unit/data_types/test_service.py b/tests/unit/data_types/test_service.py index f2b03b7..cd6f053 100644 --- a/tests/unit/data_types/test_service.py +++ b/tests/unit/data_types/test_service.py @@ -1,15 +1,11 @@ from unittest.mock import MagicMock import pytest +from revengai import FunctionDataTypesList from revengai.exceptions import ForbiddenException, NotFoundException -from revengai.models.function_data_types_list import FunctionDataTypesList -from revengai.models.function_data_types_list_item import FunctionDataTypesListItem from reai_toolkit.app.services.data_types import data_types_service as svc_mod -from reai_toolkit.app.services.data_types.data_types_service import ( - FUNCTION_IDS_BATCH_SIZE, - ImportDataTypesService, -) +from reai_toolkit.app.services.data_types.data_types_service import ImportDataTypesService @pytest.fixture @@ -19,106 +15,78 @@ def service(): @pytest.fixture def sdk(mocker): - mocker.patch.object(ImportDataTypesService, "yield_api_client") - api_class = mocker.patch.object(svc_mod, "FunctionsDataTypesApi") - api_inst = MagicMock() - api_class.return_value = api_inst - return api_inst - - -def _item(function_id: int) -> FunctionDataTypesListItem: - return FunctionDataTypesListItem.model_construct( - completed=True, - status="success", - data_types=None, - data_types_version=1, - function_id=function_id, - ) + context = mocker.patch.object(ImportDataTypesService, "yield_api_client") + api_client = MagicMock() + context.return_value.__enter__.return_value = api_client + return api_client -def _response(items): - resp = MagicMock() - resp.status = True - resp.data = FunctionDataTypesList.model_construct( - total_count=len(items), - total_data_types_count=len(items), - items=items, +def _legacy_response(items=None) -> FunctionDataTypesList: + return FunctionDataTypesList.model_construct( + total_count=len(items or []), + total_data_types_count=len(items or []), + items=items or [], ) - return resp -def test_single_batch_when_under_threshold(service, sdk): - ids = list(range(1, 11)) - sdk.list_function_data_types_for_functions.return_value = _response( - [_item(i) for i in ids] +def test_get_data_types_uses_v3_signatures_and_adapter(service, sdk, mocker): + raw_response = {"items": [], "data_types": []} + list_signatures = mocker.patch.object( + svc_mod, "list_function_signatures", return_value=raw_response ) + adapted = _legacy_response() + adapt = mocker.patch.object(svc_mod, "to_legacy_function_data_types", return_value=adapted) - result = service._get_data_types(ids) - - assert sdk.list_function_data_types_for_functions.call_count == 1 - sdk.list_function_data_types_for_functions.assert_called_once_with(function_ids=ids) - assert [item.function_id for item in result.items] == ids - - -def test_chunks_large_id_list_to_avoid_uri_too_large(service, sdk): - total = FUNCTION_IDS_BATCH_SIZE * 2 + 15 - ids = list(range(total)) - sdk.list_function_data_types_for_functions.side_effect = lambda function_ids: _response( - [_item(i) for i in function_ids] - ) + result = service._get_data_types([10, 11]) - result = service._get_data_types(ids) + assert result is adapted + list_signatures.assert_called_once_with(sdk, [10, 11], include_data_types=True) + adapt.assert_called_once_with(raw_response) - calls = sdk.list_function_data_types_for_functions.call_args_list - assert len(calls) == 3 - assert [len(c.kwargs["function_ids"]) for c in calls] == [ - FUNCTION_IDS_BATCH_SIZE, - FUNCTION_IDS_BATCH_SIZE, - 15, - ] - for c in calls: - assert len(c.kwargs["function_ids"]) <= FUNCTION_IDS_BATCH_SIZE - assert [item.function_id for item in result.items] == ids +def test_empty_list_returns_none_without_calling_sdk(service, sdk, mocker): + list_signatures = mocker.patch.object(svc_mod, "list_function_signatures") -def test_empty_list_returns_none_without_calling_sdk(service, sdk): assert service._get_data_types([]) is None - sdk.list_function_data_types_for_functions.assert_not_called() - + list_signatures.assert_not_called() -def test_skips_unsuccessful_chunks(service, sdk): - ok = _response([_item(1), _item(2)]) - bad = MagicMock(status=False, data=None) - sdk.list_function_data_types_for_functions.side_effect = [ok, bad, ok] - result = service._get_data_types( - list(range(FUNCTION_IDS_BATCH_SIZE * 2 + 1)) +def test_import_does_not_push_back_missing_v3_signature(service, sdk, mocker): + list_signatures = mocker.patch.object( + svc_mod, + "list_function_signatures", + return_value={ + "items": [{"analysis_id": 99, "function_id": 1, "has_signature": False}], + "data_types": [], + }, + ) + mocker.patch.object( + svc_mod, + "to_legacy_function_data_types", + return_value=_legacy_response(), ) - - assert [item.function_id for item in result.items] == [1, 2, 1, 2] - - -def test_import_data_types_computes_absent_when_remote_types_missing(service, sdk, mocker): - sdk.list_function_data_types_for_functions.return_value = _response([_item(1)]) apply = mocker.patch.object(svc_mod.ImportDataTypes, "execute", return_value=set()) result = service.import_data_types({1: 0x1000}) + list_signatures.assert_called_once() apply.assert_called_once() assert result.error is None - assert result.remote_absent_ids == {1} + assert result.remote_absent_ids == set() assert result.apply_failed_ids == set() def test_import_data_types_marks_apply_failures(service, sdk, mocker): - present = FunctionDataTypesListItem.model_construct( - completed=True, - status="success", - data_types=MagicMock(), - data_types_version=1, - function_id=1, + mocker.patch.object( + svc_mod, + "list_function_signatures", + return_value={"items": [], "data_types": []}, + ) + mocker.patch.object( + svc_mod, + "to_legacy_function_data_types", + return_value=_legacy_response(), ) - sdk.list_function_data_types_for_functions.return_value = _response([present]) mocker.patch.object(svc_mod.ImportDataTypes, "execute", return_value={1}) result = service.import_data_types({1: 0x1000}) @@ -127,35 +95,39 @@ def test_import_data_types_marks_apply_failures(service, sdk, mocker): assert result.apply_failed_ids == {1} -def test_import_data_types_empty_matches(service, sdk): +def test_import_data_types_empty_matches(service, sdk, mocker): + list_signatures = mocker.patch.object(svc_mod, "list_function_signatures") + result = service.import_data_types({}) assert result.error is None assert result.remote_absent_ids == set() assert result.apply_failed_ids == set() - sdk.list_function_data_types_for_functions.assert_not_called() + list_signatures.assert_not_called() def test_import_data_types_returns_error_on_forbidden(service, sdk, mocker): - sdk.list_function_data_types_for_functions.side_effect = ForbiddenException( - status=403, reason="Forbidden" + mocker.patch.object( + svc_mod, + "list_function_signatures", + side_effect=ForbiddenException(status=403, reason="Forbidden"), ) apply = mocker.patch.object(svc_mod.ImportDataTypes, "execute") - matches = {fid: fid * 16 for fid in range(FUNCTION_IDS_BATCH_SIZE * 3)} - result = service.import_data_types(matches) + result = service.import_data_types({1: 0x1000}) assert result.error is not None assert "403" in result.error assert "Forbidden" in result.error assert result.remote_absent_ids == set() apply.assert_not_called() - assert sdk.list_function_data_types_for_functions.call_count == 1 -def test_import_data_types_treats_not_found_as_all_absent(service, sdk, mocker): - sdk.list_function_data_types_for_functions.side_effect = NotFoundException( - status=404, reason="Not Found" +def test_import_data_types_treats_missing_v3_endpoint_as_absent(service, sdk, mocker): + mocker.patch.object( + svc_mod, + "list_function_signatures", + side_effect=NotFoundException(status=404, reason="Not Found"), ) apply = mocker.patch.object(svc_mod.ImportDataTypes, "execute") @@ -167,7 +139,11 @@ def test_import_data_types_treats_not_found_as_all_absent(service, sdk, mocker): def test_import_data_types_returns_error_on_unexpected_exception(service, sdk, mocker): - sdk.list_function_data_types_for_functions.side_effect = RuntimeError("boom") + mocker.patch.object( + svc_mod, + "list_function_signatures", + side_effect=RuntimeError("boom"), + ) apply = mocker.patch.object(svc_mod.ImportDataTypes, "execute") result = service.import_data_types({1: 0x1000}) diff --git a/tests/unit/data_types/test_v3_adapter.py b/tests/unit/data_types/test_v3_adapter.py new file mode 100644 index 0000000..8b986b0 --- /dev/null +++ b/tests/unit/data_types/test_v3_adapter.py @@ -0,0 +1,99 @@ +from types import SimpleNamespace + +from revengai import FunctionInfo, FunctionType + +from reai_toolkit.app.services.data_types.v3_data_types import ( + build_signature_update, + normalise_type_text, + to_legacy_function_data_types, +) + + +def _response(): + return { + "items": [ + { + "analysis_id": 99, + "function_id": 10, + "function_name": "target", + "has_signature": True, + "parameters": [ + {"ordinal": 0, "name": "ctx", "data_type_id": 3}, + {"ordinal": 1, "name": "kind", "data_type_id": 4}, + ], + "return_data_type_id": 1, + } + ], + "data_types": [ + { + "analysis_id": 99, + "items": [ + {"data_type_id": 1, "kind": "BASE", "name": "int", "size": 4}, + { + "data_type_id": 2, + "kind": "STRUCT", + "name": "Context", + "size": 8, + "definition": { + "members": [ + {"offset": 0, "name": "value", "data_type_id": 1, "size": 4} + ] + }, + }, + { + "data_type_id": 3, + "kind": "POINTER", + "name": "Context *", + "size": 8, + "definition": {"pointee_data_type_id": 2}, + }, + { + "data_type_id": 4, + "kind": "ENUM", + "name": "Kind", + "size": 4, + "definition": {"values": [{"name": "FIRST", "value": "1"}]}, + }, + ], + } + ], + } + + +def test_v3_response_is_adapted_to_existing_ida_import_models(): + result = to_legacy_function_data_types(_response()) + + item = result.items[0] + assert item.function_id == 10 + assert item.data_types is not None + assert item.data_types.func_types.header.type == "int" + assert item.data_types.func_types.header.args["0x0"].type == "Context *" + assert item.data_types.func_types.header.args["0x1"].type == "Kind" + assert len(item.data_types.func_deps) == 2 + + +def test_v3_type_text_normalisation_handles_c_pointer_spelling(): + assert normalise_type_text("const struct Context *") == "Context*" + assert normalise_type_text("Context*") == "Context*" + + +def test_v3_signature_update_uses_existing_remote_type_ids(): + function_type = FunctionType.model_construct( + header=SimpleNamespace( + type="int", + args={ + "0x0": SimpleNamespace(offset=0, name="ctx", type="Context *"), + "0x1": SimpleNamespace(offset=1, name="kind", type="Kind"), + }, + ) + ) + info = FunctionInfo.model_construct(func_types=function_type, func_deps=[]) + + assert build_signature_update(_response(), 10, function_info=info) == { + "parameters": [ + {"ordinal": 0, "name": "ctx", "data_type_id": 3}, + {"ordinal": 1, "name": "kind", "data_type_id": 4}, + ], + "return_data_type_id": 1, + } + diff --git a/tests/unit/variable_sync/test_service.py b/tests/unit/variable_sync/test_service.py index 4621024..6e00960 100644 --- a/tests/unit/variable_sync/test_service.py +++ b/tests/unit/variable_sync/test_service.py @@ -1,21 +1,14 @@ import queue -from types import SimpleNamespace from unittest.mock import MagicMock import pytest -from libbs.artifacts import Function, FunctionArgument, FunctionHeader, StackVariable -from revengai import ( - Argument, - FunctionInfo, - FunctionType, -) +from libbs.artifacts import FunctionArgument, FunctionHeader, StackVariable +from revengai import ApiException, Argument, FunctionInfo, FunctionType from revengai.models.function_header import FunctionHeader as SdkFunctionHeader from revengai.models.stack_variable import StackVariable as SdkStackVariable from reai_toolkit.app.services.variable_sync import variable_sync_service as svc_mod -from reai_toolkit.app.services.variable_sync.variable_sync_service import ( - VariableSyncService, -) +from reai_toolkit.app.services.variable_sync.variable_sync_service import VariableSyncService @pytest.fixture @@ -33,11 +26,10 @@ def service(netstore): @pytest.fixture def sdk(mocker): - mocker.patch.object(VariableSyncService, "yield_api_client") - api_class = mocker.patch.object(svc_mod, "FunctionsDataTypesApi") - api_inst = MagicMock() - api_class.return_value = api_inst - return api_inst + context = mocker.patch.object(VariableSyncService, "yield_api_client") + api_client = MagicMock() + context.return_value.__enter__.return_value = api_client + return api_client def _sdk_stack_var(offset: int, name: str, type_: str) -> SdkStackVariable: @@ -73,10 +65,38 @@ def _function_info(stack_vars=None, args=None, ret_type="void") -> FunctionInfo: return FunctionInfo.model_construct(func_types=func_types, func_deps=[]) +def _signature_response(function_id=42): + return { + "items": [ + { + "analysis_id": 7, + "function_id": function_id, + "function_name": "f", + "has_signature": True, + "parameters": [ + {"ordinal": 0, "name": "arg", "data_type_id": 1}, + ], + "return_data_type_id": 1, + } + ], + "data_types": [ + { + "analysis_id": 7, + "items": [ + {"data_type_id": 1, "kind": "BASE", "name": "int", "size": 4}, + ], + } + ], + } + + def test_patch_stack_var_matches_by_offset(service): info = _function_info(stack_vars={"-0x20": _sdk_stack_var(-32, "local_20", "char")}) - changed = service._patch(info, StackVariable(stack_offset=-32, name="counter", type_="int", size=4, addr=0x0)) + changed = service._patch( + info, + StackVariable(stack_offset=-32, name="counter", type_="int", size=4, addr=0x0), + ) assert changed is True entry = info.func_types.stack_vars["-0x20"] @@ -84,19 +104,13 @@ def test_patch_stack_var_matches_by_offset(service): assert entry.type == "int" -def test_patch_stack_var_no_offset_match_is_noop(service): - info = _function_info(stack_vars={"-0x20": _sdk_stack_var(-32, "local_20", "char")}) - - changed = service._patch(info, StackVariable(stack_offset=-99, name="x", type_="int", size=4, addr=0x0)) - - assert changed is False - assert info.func_types.stack_vars["-0x20"].name == "local_20" - - def test_patch_header_updates_arg_and_return_type(service): info = _function_info(args={"0x0": _sdk_arg(0, "a1", "int")}, ret_type="void") fheader = FunctionHeader( - name="f", addr=0x0, type_="int", args={0: FunctionArgument(offset=0, name="count", type_="size_t", size=8)} + name="f", + addr=0x0, + type_="int", + args={0: FunctionArgument(offset=0, name="count", type_="size_t", size=8)}, ) changed = service._patch(info, fheader) @@ -108,249 +122,101 @@ def test_patch_header_updates_arg_and_return_type(service): assert info.func_types.type == "int" -def test_push_local_types_batch_noop_without_targets_or_analysis(service): - service._deci = MagicMock() - assert service.push_local_function_types_batch({}, analysis_id=1) == 0 - assert service.push_local_function_types_batch({1: 0x1000}, analysis_id=None) == 0 - - -def test_push_local_types_batch_uses_version_zero_when_absent(service, sdk, mocker): - service._deci = MagicMock() - service._deci.binary_base_addr = 0x400000 - build = mocker.patch.object(service, "_build_function_info", return_value=_function_info()) - sdk.list_function_data_types_for_functions.return_value = MagicMock( - status=True, data=MagicMock(items=[]) - ) - out = MagicMock() - out.results = [SimpleNamespace(function_id=1, status="updated", data_types_version=1)] - sdk.batch_update_function_data_types.return_value = out - - updated = service.push_local_function_types_batch({1: 0x401000}, analysis_id=9) - - assert updated == 1 - build.assert_called_once_with(0x401000 - 0x400000) - body = sdk.batch_update_function_data_types.call_args.kwargs[ - "batch_update_data_types_input_body" - ] - assert body.functions[0].function_id == 1 - assert body.functions[0].data_types_version == 0 - - -def test_push_local_types_batch_overwrites_with_remote_version(service, sdk, mocker): - service._deci = MagicMock() - service._deci.binary_base_addr = 0 - mocker.patch.object(service, "_build_function_info", return_value=_function_info()) - sdk.list_function_data_types_for_functions.return_value = MagicMock( - status=True, - data=MagicMock(items=[SimpleNamespace(function_id=1, data_types_version=5)]), +def test_push_local_types_batch_uses_v3_signature_update(service, sdk, mocker): + service._deci = MagicMock(binary_base_addr=0x400000) + build = mocker.patch.object( + service, + "_build_function_info", + return_value=_function_info(args={"0x0": _sdk_arg(0, "arg", "int")}, ret_type="int"), ) - out = MagicMock() - out.results = [SimpleNamespace(function_id=1, status="updated", data_types_version=6)] - sdk.batch_update_function_data_types.return_value = out - - service.push_local_function_types_batch({1: 0x1000}, analysis_id=9) - - body = sdk.batch_update_function_data_types.call_args.kwargs[ - "batch_update_data_types_input_body" - ] - assert body.functions[0].data_types_version == 5 - - -def test_push_local_types_batch_retries_on_version_conflict(service, sdk, mocker): - service._deci = MagicMock() - service._deci.binary_base_addr = 0 - mocker.patch.object(service, "_build_function_info", return_value=_function_info()) - sdk.list_function_data_types_for_functions.side_effect = [ - MagicMock( - status=True, - data=MagicMock(items=[SimpleNamespace(function_id=1, data_types_version=1)]), - ), - MagicMock( - status=True, - data=MagicMock(items=[SimpleNamespace(function_id=1, data_types_version=2)]), - ), - ] - conflict = MagicMock() - conflict.results = [ - SimpleNamespace(function_id=1, status="version_conflict", data_types_version=None) - ] - ok = MagicMock() - ok.results = [SimpleNamespace(function_id=1, status="updated", data_types_version=3)] - sdk.batch_update_function_data_types.side_effect = [conflict, ok] - - updated = service.push_local_function_types_batch({1: 0x1000}, analysis_id=9) - - assert updated == 1 - assert sdk.batch_update_function_data_types.call_count == 2 - - -def test_resolve_type_builds_valid_function_info_dependency(service, mocker): - from libbs.artifacts import Typedef - from revengai import FunctionDependency, FunctionInfo - mocker.patch.object( - svc_mod, "_read_named_type", return_value=Typedef(name="u32", type_="unsigned int") + svc_mod, "list_function_signatures", return_value=_signature_response(1) ) + update = mocker.patch.object(svc_mod, "update_function_signature") - dep, referenced = service._resolve_type("u32") + updated = service.push_local_function_types_batch({1: 0x401000}, analysis_id=7) - assert isinstance(dep, FunctionDependency) - assert referenced == ["unsigned int"] - info = FunctionInfo(func_types=None, func_deps=[dep]) - assert info.to_dict()["func_deps"][0] == { - "artifact_type": "Typedef", - "name": "u32", - "type": "unsigned int", + assert updated == 1 + build.assert_called_once_with(0x1000) + update.assert_called_once() + assert update.call_args.args[1:3] == (7, 1) + assert update.call_args.args[3] == { + "parameters": [{"ordinal": 0, "name": "arg", "data_type_id": 1}], + "return_data_type_id": 1, } -def test_patch_stack_var_type_change_keeps_name(service): - info = _function_info(stack_vars={"-0x20": _sdk_stack_var(-32, "local_20", "char")}) - - changed = service._patch(info, StackVariable(stack_offset=-32, name=None, type_="int", size=4, addr=0x0)) +def test_push_local_types_batch_noop_without_targets_or_analysis(service, sdk, mocker): + list_signatures = mocker.patch.object(svc_mod, "list_function_signatures") - assert changed is True - entry = info.func_types.stack_vars["-0x20"] - assert entry.name == "local_20" - assert entry.type == "int" + service._deci = MagicMock() + assert service.push_local_function_types_batch({}, analysis_id=1) == 0 + assert service.push_local_function_types_batch({1: 0x1000}, analysis_id=None) == 0 + list_signatures.assert_not_called() -def test_patch_header_arg_type_change_keeps_name(service): - info = _function_info(args={"0x0": _sdk_arg(0, "oldfile", "char *")}, ret_type="int") - fheader = FunctionHeader( - name=None, addr=0x0, type_=None, args={0: FunctionArgument(offset=0, name=None, type_="wchar_t *", size=8)} +def test_push_change_updates_function_signature(service, sdk, netstore, mocker): + netstore.get_analysis_id.return_value = 7 + mocker.patch.object( + service, + "_build_function_info", + return_value=_function_info(args={"0x0": _sdk_arg(0, "arg", "int")}, ret_type="int"), ) - - changed = service._patch(info, fheader) - - assert changed is True - assert info.func_types.header.args["0x0"].name == "oldfile" - assert info.func_types.header.args["0x0"].type == "wchar_t *" - - -def test_patch_header_identical_is_noop(service): - info = _function_info(args={"0x0": _sdk_arg(0, "a1", "int")}, ret_type="void") - fheader = FunctionHeader( - name="f", addr=0x0, type_="void", args={0: FunctionArgument(offset=0, name="a1", type_="int", size=8)} + mocker.patch.object( + svc_mod, "list_function_signatures", return_value=_signature_response(42) ) + update = mocker.patch.object(svc_mod, "update_function_signature") - assert service._patch(info, fheader) is False - - -def test_push_change_fetches_patches_and_pushes(service, sdk, netstore): - netstore.get_analysis_id.return_value = 7 - info = _function_info(stack_vars={"-0x20": _sdk_stack_var(-32, "local_20", "char")}) - fetched = MagicMock() - fetched.status = True - fetched.data.items = [ - MagicMock(function_id=42, data_types=info, data_types_version=3) - ] - sdk.list_function_data_types_for_functions.return_value = fetched - sdk.batch_update_function_data_types.return_value = MagicMock( - results=[MagicMock(status="updated")] + service._push_change( + 42, + 0x2668, + FunctionHeader( + name=None, + addr=0x2668, + type_=None, + args={0: FunctionArgument(offset=0, name="arg", type_="int", size=4)}, + ), ) - service._push_change(42, 0x2668, StackVariable(stack_offset=-32, name="n", type_="int", size=4, addr=0x0)) - - sdk.batch_update_function_data_types.assert_called_once() - body = sdk.batch_update_function_data_types.call_args.kwargs[ - "batch_update_data_types_input_body" - ] - item = body.functions[0] - assert item.function_id == 42 - assert item.data_types_version == 3 + update.assert_called_once() + assert update.call_args.args[1] == 7 + assert update.call_args.args[2] == 42 -def test_push_change_skips_push_when_unchanged(service, sdk, netstore): +def test_push_change_skips_stack_variable_write_because_v3_has_no_endpoint( + service, sdk, netstore, mocker +): netstore.get_analysis_id.return_value = 7 - info = _function_info(stack_vars={"-0x20": _sdk_stack_var(-32, "local_20", "char")}) - fetched = MagicMock() - fetched.status = True - fetched.data.items = [ - MagicMock(function_id=42, data_types=info, data_types_version=3) - ] - sdk.list_function_data_types_for_functions.return_value = fetched - - service._push_change(42, 0x2668, StackVariable(stack_offset=-999, name="n", type_="int", size=4, addr=0x0)) - - sdk.batch_update_function_data_types.assert_not_called() - - -def test_push_change_retries_on_version_conflict(service, sdk, netstore): - netstore.get_analysis_id.return_value = 7 - - def fresh_response(): - info = _function_info(stack_vars={"-0x20": _sdk_stack_var(-32, "local_20", "char")}) - resp = MagicMock() - resp.status = True - resp.data.items = [MagicMock(function_id=42, data_types=info, data_types_version=3)] - return resp + build = mocker.patch.object(service, "_build_function_info") + list_signatures = mocker.patch.object(svc_mod, "list_function_signatures") + update = mocker.patch.object(svc_mod, "update_function_signature") - sdk.list_function_data_types_for_functions.side_effect = lambda function_ids: fresh_response() - sdk.batch_update_function_data_types.side_effect = [ - MagicMock(results=[MagicMock(status="version_conflict")]), - MagicMock(results=[MagicMock(status="updated")]), - ] - - service._push_change(42, 0x2668, StackVariable(stack_offset=-32, name="n", type_="int", size=4, addr=0x0)) + service._push_change( + 42, + 0x2668, + StackVariable(stack_offset=-32, name="n", type_="int", size=4, addr=0x0), + ) - assert sdk.batch_update_function_data_types.call_count == 2 + build.assert_not_called() + list_signatures.assert_not_called() + update.assert_not_called() -def test_push_change_builds_object_when_no_stored_types(service, sdk, netstore, mocker): +def test_push_change_swallows_v3_api_error(service, sdk, netstore, mocker): netstore.get_analysis_id.return_value = 7 - empty = MagicMock() - empty.status = True - empty.data.items = [] - sdk.list_function_data_types_for_functions.return_value = empty - sdk.batch_update_function_data_types.return_value = MagicMock( - results=[MagicMock(status="updated")] - ) - - func = Function( - addr=0x2668, - size=10, - name="relink", - header=FunctionHeader( - name="relink", - addr=0x2668, - type_="int", - args={0: FunctionArgument(offset=0, name="oldfile", type_="wchar_t *", size=8)}, - ), - stack_vars={-32: StackVariable(stack_offset=-32, name="v2", type_="dev_t", size=8, addr=0x2668)}, + mocker.patch.object(service, "_build_function_info", return_value=_function_info()) + mocker.patch.object( + svc_mod, + "list_function_signatures", + side_effect=ApiException(status=404, reason="Not Found"), ) - service.attach_decompiler(MagicMock()) - mocker.patch.object(svc_mod, "_read_decompiler_function", return_value=func) service._push_change( - 2015112411, + 42, 0x2668, - FunctionHeader(name=None, addr=0x2668, type_=None, args={0: FunctionArgument(offset=0, name=None, type_="wchar_t *", size=8)}), + FunctionHeader(name=None, addr=0x2668, type_=None, args={}), ) - sdk.batch_update_function_data_types.assert_called_once() - body = sdk.batch_update_function_data_types.call_args.kwargs[ - "batch_update_data_types_input_body" - ] - item = body.functions[0] - assert item.function_id == 2015112411 - assert item.data_types_version == 0 - assert item.data_types["func_types"]["header"]["args"]["0x0"]["name"] == "oldfile" - assert item.data_types["func_types"]["header"]["args"]["0x0"]["type"] == "wchar_t *" - assert item.data_types["func_types"]["stack_vars"]["-0x20"]["name"] == "v2" - - -def test_push_change_skips_build_when_no_decompiler(service, sdk, netstore): - netstore.get_analysis_id.return_value = 7 - empty = MagicMock() - empty.status = True - empty.data.items = [] - sdk.list_function_data_types_for_functions.return_value = empty - - service._push_change(42, 0x2668, StackVariable(stack_offset=-32, name="n", type_="int", size=4, addr=0x0)) - - sdk.batch_update_function_data_types.assert_not_called() - def test_collect_func_deps_resolves_typedef_chain(service, mocker): from libbs.artifacts import Typedef @@ -362,41 +228,26 @@ def test_collect_func_deps_resolves_typedef_chain(service, mocker): "dev_t": Typedef(name="dev_t", type_="__dev_t"), "__dev_t": Typedef(name="__dev_t", type_="unsigned long"), } - mocker.patch.object(svc_mod, "_read_named_type", side_effect=lambda deci, name: chain.get(name)) + mocker.patch.object( + svc_mod, "_read_named_type", side_effect=lambda deci, name: chain.get(name) + ) deps = service._collect_func_deps(func_type) assert sorted(d.name for d in deps) == ["__dev_t", "dev_t"] - assert all(d.artifact_type == "Typedef" for d in deps) -def test_collect_func_deps_resolves_struct_members(service, mocker): - from libbs.artifacts import Struct, StructMember, Typedef - - service.attach_decompiler(MagicMock()) - func_type = _function_info(args={"0x0": _sdk_arg(0, "p", "mystruct *")}, ret_type="int").func_types - - types = { - "mystruct": Struct( - name="mystruct", size=8, members={0: StructMember(name="x", offset=0, type_="myint", size=4)} - ), - "myint": Typedef(name="myint", type_="int"), - } - mocker.patch.object(svc_mod, "_read_named_type", side_effect=lambda deci, name: types.get(name)) - - deps = service._collect_func_deps(func_type) - - by_name = {d.name: d for d in deps} - assert set(by_name) == {"mystruct", "myint"} - assert by_name["mystruct"].members["0x0"]["type"] == "myint" - - -def test_push_change_no_analysis_id_does_nothing(service, sdk, netstore): +def test_push_change_no_analysis_id_does_nothing(service, sdk, netstore, mocker): netstore.get_analysis_id.return_value = None + list_signatures = mocker.patch.object(svc_mod, "list_function_signatures") - service._push_change(42, 0x2668, StackVariable(stack_offset=-32, name="n", type_="int", size=4, addr=0x0)) + service._push_change( + 42, + 0x2668, + StackVariable(stack_offset=-32, name="n", type_="int", size=4, addr=0x0), + ) - sdk.list_function_data_types_for_functions.assert_not_called() + list_signatures.assert_not_called() def test_enqueue_change_debounces_rapid_duplicates(service, mocker):