diff options
Diffstat (limited to 'python')
| -rw-r--r-- | python/__init__.py | 1 | ||||
| -rw-r--r-- | python/basedetection.py | 321 | ||||
| -rw-r--r-- | python/binaryview.py | 4 | ||||
| -rw-r--r-- | python/examples/raw_binary_base_detection.py | 101 | ||||
| -rw-r--r-- | python/highlevelil.py | 12 | ||||
| -rw-r--r-- | python/interaction.py | 19 | ||||
| -rw-r--r-- | python/websocketprovider.py | 11 |
7 files changed, 454 insertions, 15 deletions
diff --git a/python/__init__.py b/python/__init__.py index 498475f4..b7702569 100644 --- a/python/__init__.py +++ b/python/__init__.py @@ -74,6 +74,7 @@ from .typearchive import * from .typecontainer import * from .exceptions import * from .project import * +from .basedetection import * # We import each of these by name to prevent conflicts between # log.py and the function 'log' which we don't import below from .log import ( diff --git a/python/basedetection.py b/python/basedetection.py new file mode 100644 index 00000000..655f017b --- /dev/null +++ b/python/basedetection.py @@ -0,0 +1,321 @@ +# coding=utf-8 +# Copyright (c) 2015-2024 Vector 35 Inc +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. + +import os +import ctypes +from typing import Optional, Union, Literal +from dataclasses import dataclass +from .enums import BaseAddressDetectionPOIType, BaseAddressDetectionConfidence, BaseAddressDetectionPOISetting +from .binaryview import BinaryView +from . import _binaryninjacore as core + + +@dataclass +class BaseAddressDetectionReason: + """``class BaseAddressDetectionReason`` is a class that stores information used to understand why a base address + is a candidate. It consists of a pointer, the offset of the point-of-interest that the pointer aligns with, and the + type of point-of-interest (string, function, or data variable)""" + + pointer: int + offset: int + type: BaseAddressDetectionPOIType + + +class BaseAddressDetection: + """ + ``class BaseAddressDetection`` is a class that is used to detect candidate base addresses for position-dependent + raw binaries + + :Example: + + >>> from binaryninja import * + >>> bad = BaseAddressDetection("firmware.bin") + >>> bad.detect_base_address() + True + >>> hex(bad.preferred_base_address) + '0x4000000' + """ + + def __init__(self, view: Union[str, os.PathLike, BinaryView]) -> None: + if isinstance(view, str) or isinstance(view, os.PathLike): + view = BinaryView.load(str(view), update_analysis=False) + + _handle = core.BNCreateBaseAddressDetection(view.handle) + assert _handle is not None, "core.BNCreateBaseAddressDetection returned None" + self._handle = _handle + self._view_arch = view.arch + + self._scores = list() + self._confidence = 0 + self._last_tested_base_address = None + + def __del__(self): + if core is not None: + core.BNFreeBaseAddressDetection(self._handle) + + @property + def scores(self) -> list[tuple[int, int]]: + """ + ``scores`` returns a list of candidate base addresses and their scores + + .. note:: The score is set to the number of times a pointer pointed to a point-of-interest at that base address + + :Example: + + >>> from binaryninja import * + >>> bad = BaseAddressDetection("firmware.bin") + >>> bad.detect_base_address() + True + >>> for addr, score in bad.scores: + ... print(f"0x{addr:x}: {score}") + ... + 0x4000000: 7 + 0x400dc00: 1 + 0x400d800: 1 + 0x400cc00: 1 + 0x400c400: 1 + 0x400bc00: 1 + 0x400b800: 1 + 0x3fffc00: 1 + + :return: list of tuples containing each base address and score + :rtype: list[tuple[int, int]] + """ + + return self._scores + + @property + def confidence(self) -> BaseAddressDetectionConfidence: + """ + ``confidence`` returns an enum that indicates confidence the preferred candidate base address is correct + + :return: confidence of the base address detection results + :rtype: BaseAddressDetectionConfidence + """ + + return self._confidence + + @property + def last_tested_base_address(self) -> int: + """ + ``last_tested_base_address`` returns the last candidate base address that was tested + + .. note:: This is useful for situations where the user aborts the analysis and wants to restart from the last \ + tested base address by setting the ``low_boundary`` parameter in :py:func:`BaseAddressDetection.detect_base_address` + + :return: last candidate base address tested + :rtype: int + """ + + return self._last_tested_base_address + + @property + def preferred_base_address(self) -> Optional[int]: + """ + ``preferred_base_address`` returns the candidate base address which contains the most amount of pointers that + align with discovered points-of-interest in the binary + + .. note:: :py:attr:`BaseAddressDetection.confidence` reports a confidence level that the preferred base is correct + + .. note:: :py:attr:`BaseAddressDetection.scores` returns a list of the top 10 candidate base addresses and their \ + scores and can be used to discover other potential candidates + + :return: preferred candidate base address + :rtype: int + """ + + if not self._scores: + return None + + return self._scores[0][0] + + @property + def aborted(self) -> bool: + """ + ``aborted`` indicates whether or not base address detection analysis was aborted early + + :return: True if the analysis was aborted, False otherwise + :rtype: bool + """ + + return core.BNIsBaseAddressDetectionAborted(self._handle) + + def detect_base_address( + self, + arch: Optional[str] = "", + analysis: Optional[str] = Literal["basic", "controlFlow", "full"], + min_strlen: Optional[int] = 10, + alignment: Optional[int] = 1024, + low_boundary: Optional[int] = 0, + high_boundary: Optional[int] = 0xFFFFFFFFFFFFFFFF, + poi_analysis: Optional[BaseAddressDetectionPOISetting] = BaseAddressDetectionPOISetting.POIAnalysisAll, + max_pointers: Optional[int] = 128, + ) -> bool: + """ + ``detect_base_address`` runs initial analysis and attempts to identify candidate base addresses + + .. note:: This operation can take a long time to complete depending on the size and complexity of the binary \ + and the settings used + + :param str arch: CPU architecture of the binary (defaults to using auto-detection) + :param str analysis: analysis mode (``basic``, ``controlFlow``, or ``full``) + :param int min_strlen: minimum length of a string to be considered a point-of-interest + :param int alignment: byte boundary to align the base address to while brute-forcing + :param int low_boundary: lower boundary of the base address range to test + :param int high_boundary: upper boundary of the base address range to test + :param BaseAddressDetectionPOISetting poi_analysis: specifies types of points-of-interest to use for analysis + :param int max_pointers: maximum number of candidate pointers to collect per pointer cluster + :return: True if initial analysis completed with results, False otherwise + :rtype: bool + """ + + if not arch and self._view_arch: + arch = str(self._view_arch) + + if analysis not in ["basic", "controlFlow", "full"]: + raise ValueError("invalid analysis setting") + + if alignment <= 0: + raise ValueError("alignment must be greater than 0") + + if max_pointers < 2: + raise ValueError("max pointers must be at least 2") + + if high_boundary < low_boundary: + raise ValueError("upper boundary must be greater than lower boundary") + + settings = core.BNBaseAddressDetectionSettings( + arch.encode(), + analysis.encode(), + min_strlen, + alignment, + low_boundary, + high_boundary, + poi_analysis, + max_pointers, + ) + + if not core.BNDetectBaseAddress(self._handle, settings): + return False + + max_candidates = 10 + scores = (core.BNBaseAddressDetectionScore * max_candidates)() + confidence = core.BaseAddressDetectionConfidenceEnum() + last_base = ctypes.c_ulonglong() + num_candidates = core.BNGetBaseAddressDetectionScores( + self._handle, scores, max_candidates, ctypes.byref(confidence), ctypes.byref(last_base) + ) + + if num_candidates == 0: + return False + + self._scores.clear() + for i in range(num_candidates): + self._scores.append((scores[i].BaseAddress, scores[i].Score)) + + self._confidence = confidence.value + self._last_tested_base_address = last_base.value + return True + + def abort(self) -> None: + """ + ``abort`` aborts base address detection analysis + + .. note:: ``abort`` does not stop base address detection until after initial analysis has completed and it is \ + in the base address enumeration phase + + :rtype: None + """ + + core.BNAbortBaseAddressDetection(self._handle) + + def get_reasons(self, base_address: int) -> list[BaseAddressDetectionReason]: + """ + ``get_reasons`` returns a list of reasons that can be used to determine why a base address is a candidate + + :param int base_address: base address to get reasons for + :return: list of reasons for the specified base address + :rtype: list[BaseAddressDetectionReason] + """ + + count = ctypes.c_size_t() + reasons = core.BNGetBaseAddressDetectionReasons(self._handle, base_address, ctypes.byref(count)) + if count.value == 0: + return [] + + try: + result = list() + for i in range(count.value): + result.append(BaseAddressDetectionReason(reasons[i].Pointer, reasons[i].POIOffset, reasons[i].POIType)) + return result + finally: + core.BNFreeBaseAddressDetectionReasons(reasons) + + def _get_data_hits_by_type(self, base_address: int, poi_type: int) -> int: + reasons = self.get_reasons(base_address) + if not reasons: + return 0 + + hits = 0 + for reason in reasons: + if reason.type == poi_type: + hits += 1 + + return hits + + def get_string_hits(self, base_address: int) -> int: + """ + ``get_string_hits`` returns the number of times a pointer pointed to a string at the specified + base address + + .. note:: Data variables are only used as points-of-interest if analysis doesn't discover enough strings and \ + functions + + :param int base_address: base address to get string hits for + :return: number of string hits for the specified base address + :rtype: int + """ + + return self._get_data_hits_by_type(base_address, BaseAddressDetectionPOIType.POIString) + + def get_function_hits(self, base_address: int) -> int: + """ + ``get_function_hits`` returns the number of times a pointer pointed to a function at the + specified base address + + :param int base_address: base address to get function hits for + :return: number of function hits for the specified base address + :rtype: int + """ + + return self._get_data_hits_by_type(base_address, BaseAddressDetectionPOIType.POIFunction) + + def get_data_hits(self, base_address: int) -> int: + """ + ``get_data_hits`` returns the number of times a pointer pointed to a data variable at the + specified base address + + :param int base_address: base address to get data hits for + :return: number of data hits for the specified base address + :rtype: int + """ + + return self._get_data_hits_by_type(base_address, BaseAddressDetectionPOIType.POIDataVariable) diff --git a/python/binaryview.py b/python/binaryview.py index 9ef7a913..43eda87f 100644 --- a/python/binaryview.py +++ b/python/binaryview.py @@ -8833,7 +8833,7 @@ class BinaryView: ``show_graph_report`` displays a :py:class:`FlowGraph` object `graph` in a new tab with ``title``. :param title: Title of the graph - :type title: Plain text string title + :type title: Text string title of the tab :param graph: The graph you wish to display :type graph: :py:class:`FlowGraph` object """ @@ -8844,7 +8844,7 @@ class BinaryView: ``get_address_input`` Gets a virtual address via a prompt displayed to the user :param prompt: Prompt for the dialog - :param title: Display title, if displayed via the UI + :param title: Window title, if used in the UI :param current_address: Optional current address, for relative inputs :return: The value entered by the user, if one was entered """ diff --git a/python/examples/raw_binary_base_detection.py b/python/examples/raw_binary_base_detection.py new file mode 100644 index 00000000..76268fbe --- /dev/null +++ b/python/examples/raw_binary_base_detection.py @@ -0,0 +1,101 @@ +# Copyright (c) 2015-2024 Vector 35 Inc +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. + +"""Headless script for demonstrating Binary Ninja automated base address detection for +raw position-dependent firmware binaries +""" + +import argparse +import json +from os import walk, path +from binaryninja import BaseAddressDetection, log_to_stderr, LogLevel, log_info, log_error + + +def _get_directory_listing(_path: str) -> list[str]: + if path.isfile(_path): + return [_path] + + if not path.isdir(_path): + raise FileNotFoundError(f"Path '{_path}' is not a file or directory") + + files = [] + for dirpath, _, filenames in walk(_path): + for filename in filenames: + files.append(path.join(dirpath, filename)) + return files + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="detect base address of position-dependent raw firmware binaries") + parser.add_argument("path", help="path to the position-dependent raw firmware binary or directory") + parser.add_argument("--debug", action="store_true", help="enable debug logging") + parser.add_argument("--reasons", action="store_true", help="show reasons for base address selection") + parser.add_argument("--analysis", type=str, help="analysis level", default="basic") + parser.add_argument("--arch", type=str, default="", help="architecture of the binary") + return parser.parse_args() + + +def _setup_logger(debug: bool) -> None: + if debug: + log_to_stderr(LogLevel.DebugLog) + else: + log_to_stderr(LogLevel.InfoLog) + + +def main() -> None: + """Run the program""" + args = _parse_args() + _setup_logger(args.debug) + + files = _get_directory_listing(args.path) + for _file in files: + log_info(f"Running base address detection analysis on '{_file}'...") + bad = BaseAddressDetection(_file) + if not bad.detect_base_address(analysis=args.analysis, arch=args.arch): + log_error("Base address detection analysis failed") + continue + + json_dict = dict() + json_dict["filename"] = path.basename(_file) + json_dict["preferred_candidate"] = dict() + json_dict["preferred_candidate"]["address"] = f"0x{bad.preferred_base_address:x}" + json_dict["preferred_candidate"]["confidence"] = bad.confidence + json_dict["aborted"] = bad.aborted + json_dict["last_tested"] = f"0x{bad.last_tested_base_address:x}" + json_dict["candidates"] = dict() + for baseaddr, score in bad.scores: + json_dict["candidates"][f"0x{baseaddr:x}"] = dict() + json_dict["candidates"][f"0x{baseaddr:x}"]["score"] = score + json_dict["candidates"][f"0x{baseaddr:x}"]["function hits"] = bad.get_function_hits(baseaddr) + json_dict["candidates"][f"0x{baseaddr:x}"]["string hits"] = bad.get_string_hits(baseaddr) + json_dict["candidates"][f"0x{baseaddr:x}"]["data hits"] = bad.get_data_hits(baseaddr) + if args.reasons: + json_dict["candidates"][f"0x{baseaddr:x}"]["reasons"] = dict() + for reason in bad.get_reasons(baseaddr): + json_dict["candidates"][f"0x{baseaddr:x}"]["reasons"][f"0x{reason.pointer:x}"] = { + "poi_offset": f"0x{reason.offset:x}", + "poi_type": reason.type, + } + + print(json.dumps(json_dict, indent=4)) + + +if __name__ == "__main__": + main() diff --git a/python/highlevelil.py b/python/highlevelil.py index 79ac7366..afdc2b2e 100644 --- a/python/highlevelil.py +++ b/python/highlevelil.py @@ -798,10 +798,11 @@ class HighLevelILInstruction(BaseILInstruction): :Example: >>> def get_constant_less_than_value(inst: HighLevelILInstruction, value: int) -> int: - >>> if isinstance(inst, Constant) and inst.constant < value: - >>> return inst.constant + ... if isinstance(inst, Constant) and inst.constant < value: + ... return inst.constant >>> - >>> list(inst.traverse(get_constant_less_than_value, 10)) + >>> for result in inst.traverse(get_constant_less_than_value, 10): + ... print(f"Found a constant {result} < 10 in {repr(inst)}") """ if (result := cb(self, *args, **kwargs)) is not None: yield result @@ -2556,7 +2557,7 @@ class HighLevelILFunction: def traverse(self, cb: Callable[['HighLevelILInstruction', Any], Any], *args: Any, **kwargs: Any) -> Iterator[Any]: """ - ``traverse`` iterates through all the instructions in the HighLevelILInstruction and calls the callback function for + ``traverse`` iterates through all the instructions in the HighLevelILFunction and calls the callback function for each instruction and sub-instruction. See the `Developer Docs <https://docs.binary.ninja/dev/concepts.html#walking-ils>`_ for more examples. :param Callable[[HighLevelILInstruction, Any], Any] cb: The callback function to call for each node in the HighLevelILInstruction @@ -2572,7 +2573,8 @@ class HighLevelILFunction: ... case Localcall(dest=Constant(constant=c), params=[_, _, p]) if c == target and not isinstance(p, Constant): ... return i >>> target_address = bv.get_symbol_by_raw_name('_memcpy').address - >>> list(current_il_function.traverse(find_non_constant_memcpy, target_address)) + >>> for result in current_il_function.traverse(find_non_constant_memcpy, target_address): + ... print(f"Found suspicious memcpy: {repr(i)}") """ root = self.root if root is None: diff --git a/python/interaction.py b/python/interaction.py index c767af7c..c3bfb3a2 100644 --- a/python/interaction.py +++ b/python/interaction.py @@ -1062,8 +1062,8 @@ def show_plain_text_report(title, contents): .. note:: This API functions differently on the command-line vs the UI. In the UI, a pop-up is used. On the command-line, \ a simple text prompt is used. - :param str title: title to display in the UI pop-up - :param str contents: plaintext contents to display + :param str title: Title to display in the tab + :param str contents: Plaintext contents to display :rtype: None :Example: >>> show_plain_text_report("title", "contents") @@ -1081,6 +1081,7 @@ def show_markdown_report(title, contents, plaintext=""): .. note:: This API function differently on the command-line vs the UI. In the UI a pop-up is used. On the command-line \ a simple text prompt is used. + :param str title: title to display in the tab :param str contents: markdown contents to display :param str plaintext: Plain text version to display (used on the command-line) :rtype: None @@ -1097,6 +1098,7 @@ def show_html_report(title, contents, plaintext=""): applications. This API doesn't support hyperlinking into the BinaryView, use the :py:meth:`BinaryView.show_html_report` \ API if hyperlinking is needed. + :param str title: Title to display in the tab :param str contents: HTML contents to display :param str plaintext: Plain text version to display (used on the command-line) :rtype: None @@ -1115,6 +1117,7 @@ def show_graph_report(title, graph): .. note:: This API function will have no effect outside the UI. + :param str title: Title to display in the tab :param FlowGraph graph: Flow graph to display :rtype: None """ @@ -1144,9 +1147,9 @@ def get_text_line_input(prompt, title): .. note:: This API function differently on the command-line vs the UI. In the UI a pop-up is used. On the command-line \ a simple text prompt is used. - :param str prompt: String to prompt with. - :param str title: Title of the window when executed in the UI. - :rtype: str containing the input without trailing newline character. + :param str prompt: String to prompt with + :param str title: Title of the window when executed in the UI + :rtype: str containing the input without trailing newline character :Example: >>> get_text_line_input("PROMPT>", "getinfo") PROMPT> Input! @@ -1167,9 +1170,9 @@ def get_int_input(prompt, title): .. note:: This API function differently on the command-line vs the UI. In the UI a pop-up is used. On the command-line \ a simple text prompt is used. - :param str prompt: String to prompt with. - :param str title: Title of the window when executed in the UI. - :rtype: integer value input by the user. + :param str prompt: String to prompt with + :param str title: Title of the window when executed in the UI + :rtype: integer value input by the user :Example: >>> get_int_input("PROMPT>", "getinfo") PROMPT> 10 diff --git a/python/websocketprovider.py b/python/websocketprovider.py index 93e88eb8..420f43c6 100644 --- a/python/websocketprovider.py +++ b/python/websocketprovider.py @@ -46,6 +46,9 @@ def to_bytes(field): class WebsocketClient(object): + """ + This class implements a websocket client. See :py:func:`~WebsocketClient.connect` for more details. + """ _registered_clients = [] def __init__(self, provider, handle=None): @@ -157,12 +160,20 @@ class WebsocketClient(object): :param function(bytes) -> bool on_data: function to call when data is read from the websocket :return: if the connection has started, but not necessarily if it succeeded :rtype: bool + + :Example: + >>> provider = list(WebsocketProvider)[0] + >>> client = provider.create_instance() + >>> client.connect("ws://localhost:8080", {}) + True """ if self._connected: raise RuntimeError("Cannot use connect() twice on the same WebsocketClient") self._connected = True + if headers is None: + headers = {} header_keys = (ctypes.c_char_p * len(headers))() header_values = (ctypes.c_char_p * len(headers))() for (i, item) in enumerate(headers.items()): |