| #!/usr/bin/env python3 |
| ############################################################################################################### |
| # |
| # Copyright (C) 2019 Motorola Mobility LLC |
| # |
| # Redistribution and use in source and binary forms, with or without modification, are permitted provided that |
| # the following conditions are met: |
| # |
| # 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the |
| # following disclaimer. |
| # |
| # 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and |
| # the following disclaimer in the documentation and/or other materials provided with the distribution. |
| # |
| # 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or |
| # promote products derived from this software without specific prior written permission. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED |
| # WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A |
| # PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR |
| # ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED |
| # TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) |
| # HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING |
| # NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE |
| # POSSIBILITY OF SUCH DAMAGE. |
| # |
| ############################################################################################################### |
| ############################################################################################################### |
| # |
| # Bluetooth Virtual Sniffing for Android |
| # |
| # This script supports Bluetooth Virtual Sniffing via Live Import Feature of Frontline Bluetooth Sniffer(FTS) |
| # |
| # It extracts the HCI packets from Snoop logs and redirect it to FTS for live HCI sniffing. |
| # |
| # It uses liveimport.ini and LiveImportAPI.dll from FTS path to communicate with FTS sniffer software. |
| # |
| # It works on both Windows and Ubuntu. For Ubuntu, both FTS and Python should be installed on Wine. |
| # |
| # FTS_INI_PATH & FTS_DLL_PATH should be set to absolute path of liveimport.ini and LiveImportAPI.dll of FTS |
| # |
| # Example below - This may change per machine per FTS version in Windows vs Ubuntu (Wine), set accordingly. |
| # For Windows |
| # FTS_INI_PATH = 'C:\\Program Files (x86)\\Frontline Test System II\\Frontline 13.2\\' |
| # FTS_DLL_PATH = 'C:\\Program Files (x86)\\Frontline Test System II\\Frontline 13.2\\Executables\\Core\\' |
| # |
| # For Ubuntu - FTS path recognized by Wine (not Ubuntu path) |
| # FTS_INI_PATH = 'C:\\Program Files\\Frontline Test System II\\Frontline 13.2\\' |
| # FTS_DLL_PATH = 'C:\\Program Files\\Frontline Test System II\\Frontline 13.2\\Executables\\Core\\' |
| # |
| ############################################################################################################### |
| |
| import os |
| import platform |
| import socket |
| import struct |
| import subprocess |
| import sys |
| import time |
| if sys.version_info[0] >= 3: |
| import configparser |
| else: |
| import ConfigParser as configparser |
| |
| from calendar import timegm |
| from ctypes import byref, c_bool, c_longlong, CDLL |
| from _ctypes import FreeLibrary |
| from datetime import datetime |
| |
| # Update below to right path corresponding to your machine, FTS version and OS used. |
| FTS_INI_PATH = 'C:\\Program Files (x86)\\Frontline Test System II\\Frontline 15.12\\' |
| FTS_DLL_PATH = 'C:\\Program Files (x86)\\Frontline Test System II\\Frontline 15.12\\Executables\\Core\\' |
| |
| iniName = 'liveimport.ini' |
| if (platform.architecture()[0] == '32bit'): |
| dllName = 'LiveImportAPI.dll' |
| else: |
| dllName = 'LiveImportAPI_x64.dll' |
| |
| launchFtsCmd = '\"' + FTS_DLL_PATH + 'fts.exe\"' + ' \"/ComProbe Protocol Analysis System=Generic\"' + ' \"/oemkey=Virtual\"' |
| |
| # Unix Epoch delta since 01/01/1970 |
| FILETIME_EPOCH_DELTA = 116444736000000000 |
| HUNDREDS_OF_NANOSECONDS = 10000000 |
| |
| HOST = 'localhost' |
| PORT = 8872 |
| SNOOP_ID = 16 |
| SNOOP_HDR = 24 |
| |
| |
| def get_file_time(): |
| """ |
| Obtain current time in file time format for display |
| """ |
| date_time = datetime.now() |
| file_time = FILETIME_EPOCH_DELTA + ( |
| timegm(date_time.timetuple()) * HUNDREDS_OF_NANOSECONDS) |
| file_time = file_time + (date_time.microsecond * 10) |
| return file_time |
| |
| |
| def get_connection_string(): |
| """ |
| Read ConnectionString from liveimport.ini |
| """ |
| config = configparser.ConfigParser() |
| config.read(FTS_INI_PATH + iniName) |
| try: |
| conn_str = config.get('General', 'ConnectionString') |
| except (configparser.NoSectionError, configparser.NoOptionError): |
| return None |
| |
| return conn_str |
| |
| |
| def get_configuration_string(): |
| """ |
| Read Configuration string from liveimport.ini |
| """ |
| config_str = '' |
| config = configparser.ConfigParser() |
| config.read(FTS_INI_PATH + iniName) |
| try: |
| config_items = config.items('Configuration') |
| except (configparser.NoSectionError, configparser.NoOptionError): |
| return None |
| |
| if config_items is not None: |
| for item in config_items: |
| key, value = item |
| config_str += ("%s=%s\n" % (key, value)) |
| return config_str |
| else: |
| return None |
| |
| |
| def check_live_import_connection(live_import): |
| """ |
| Launch FTS app in Virtual Sniffing Mode |
| Check if FTS App is ready to start receiving the data. |
| If not, wait until 1 min and exit if FTS didn't start. |
| """ |
| is_connection_running = c_bool() |
| count = 0 |
| |
| status = live_import.IsAppReady(byref(is_connection_running)) |
| if (is_connection_running.value == True): |
| print("FTS is already launched, Start capture if not already started") |
| return True |
| |
| print("Launching FTS Virtual Sniffing") |
| try: |
| ftsProcess = subprocess.Popen((launchFtsCmd), stdout=subprocess.PIPE) |
| except: |
| print("Error in Launching FTS.. exiting") |
| return False |
| |
| while (is_connection_running.value == False and count < 12): |
| status = live_import.IsAppReady(byref(is_connection_running)) |
| if (status < 0): |
| print("Live Import Internal Error %d" % (status)) |
| return False |
| if (is_connection_running.value == False): |
| print("Waiting for 5 sec.. Open FTS Virtual Sniffing") |
| time.sleep(5) |
| count += 1 |
| if (is_connection_running.value == True): |
| print("FTS is ready to receive the data, Start capture now") |
| return True |
| else: |
| print("FTS Virtual Sniffing didn't start until 1 min.. exiting") |
| return False |
| |
| |
| def init_live_import(conn_str, config_str): |
| """ |
| Load DLL and Initialize the LiveImport module for FTS. |
| """ |
| success = c_bool() |
| try: |
| live_import = CDLL(FTS_DLL_PATH + dllName) |
| except: |
| return None |
| |
| if live_import is None: |
| print("Error: Path to LiveImportAPI.dll is incorrect.. exiting") |
| return None |
| |
| print(dllName + " loaded successfully") |
| result = live_import.InitializeLiveImport( |
| conn_str.encode('ascii', 'ignore'), config_str.encode( |
| 'ascii', 'ignore'), byref(success)) |
| if (result < 0): |
| print("Live Import Init failed") |
| return None |
| else: |
| print("Live Import Init success") |
| return live_import |
| |
| |
| def release_live_import(live_import): |
| """ |
| Cleanup and exit live import module. |
| """ |
| if live_import is not None: |
| live_import.ReleaseLiveImport() |
| FreeLibrary(live_import._handle) |
| |
| |
| def main(): |
| |
| print("Bluetooth Virtual Sniffing for Fluoride") |
| connection_str = get_connection_string() |
| if connection_str is None: |
| print("Error: path to liveimport.ini is incorrect.. exiting") |
| exit(0) |
| |
| configuration_str = get_configuration_string() |
| if configuration_str is None: |
| print("Error: path to liveimport.ini is incorrect.. exiting") |
| exit(0) |
| |
| live_import = init_live_import(connection_str, configuration_str) |
| if live_import is None: |
| print("Error: Path to LiveImportAPI.dll is incorrect.. exiting") |
| exit(0) |
| |
| if (check_live_import_connection(live_import) == False): |
| release_live_import(live_import) |
| exit(0) |
| |
| # Wait until the forward socket is ready |
| print("Waiting until adb is ready") |
| os.system('adb wait-for-device forward tcp:8872 tcp:8872') |
| |
| btsnoop_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| btsnoop_sock.connect((HOST, PORT)) |
| snoop_id = btsnoop_sock.recv(SNOOP_ID) |
| if not snoop_id.startswith(b"btsnoop"): |
| print("Error: Snoop ID wasn't received.. exiting") |
| release_live_import(live_import) |
| exit(0) |
| |
| while True: |
| try: |
| snoop_hdr = btsnoop_sock.recv(SNOOP_HDR) |
| if snoop_hdr is not None: |
| try: |
| olen, ilen, flags = struct.unpack(">LLL", snoop_hdr[0:12]) |
| except struct.error: |
| print("Error: Invalid data", repr(snoop_hdr)) |
| continue |
| |
| file_time = get_file_time() |
| timestamp = c_longlong(file_time) |
| |
| snoop_data = b'' |
| while (len(snoop_data) < olen): |
| data_frag = btsnoop_sock.recv(olen - len(snoop_data)) |
| if data_frag is not None: |
| snoop_data += data_frag |
| |
| print("Bytes received %d Olen %d ilen %d flags %d" % |
| (len(snoop_data), olen, ilen, flags)) |
| packet_type = struct.unpack(">B", snoop_data[0:1])[0] |
| if packet_type == 1: |
| drf = 1 |
| isend = 0 |
| elif packet_type == 2: |
| drf = 2 |
| if (flags & 0x01): |
| isend = 1 |
| else: |
| isend = 0 |
| elif packet_type == 3: |
| drf = 4 |
| if (flags & 0x01): |
| isend = 1 |
| else: |
| isend = 0 |
| elif packet_type == 4: |
| drf = 8 |
| isend = 1 |
| |
| result = live_import.SendFrame(olen - 1, olen - 1, |
| snoop_data[1:olen], drf, isend, |
| timestamp) |
| if (result < 0): |
| print("Send frame failed") |
| except KeyboardInterrupt: |
| print("Cleanup and exit") |
| release_live_import(live_import) |
| btsnoop_sock.close() |
| exit(0) |
| |
| |
| if __name__ == '__main__': |
| main() |
