Added pre-commit, license, fixed pep8 violations

Author: csacca

.pre-commit-config.yaml

@@ -0,0 +1,25 @@
+repos:
+    - repo: https://github.com/pre-commit/pre-commit-hooks
+      rev: v3.4.0
+      hooks:
+          - id: check-yaml
+          - id: end-of-file-fixer
+            exclude: '.*.bin|requirements.txt'
+          - id: trailing-whitespace
+            exclude: '.*.bin'
+
+    - repo: https://github.com/pycqa/isort
+      rev: 5.7.0
+      hooks:
+          - id: isort
+
+    - repo: https://github.com/psf/black
+      rev: 20.8b1
+      hooks:
+          - id: black
+            language_version: python3
+
+    - repo: https://gitlab.com/pycqa/flake8
+      rev: 3.8.4
+      hooks:
+          - id: flake8

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2021 Christopher Sacca
+
+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.

README.md

@@ -1 +1,25 @@
-# Reverse engineering of the Horizon 7.4 AT (-02) Treadmill
+# Reverse Engineering of the Horizon 7.4 AT (-02) Treadmill
+
+## Analyzed software
+
+- Horizon 7.4 AT Treadmill Firmware: [7.4AT-02 (TM746B or TM499B) Treadmill V2.001](https://cdn.horizonfitness.rocks/content/5321/Horizon-7.4.zip)
+  - `7.4AT-02+V2.001.EFM.bin`
+  - SHA-256: `f7e6f57a15f69bdf68c484cef338a42265723f788b140950ee169e0a3761e8c3`
+- Android App: "AFG Pro Fitness" (com.xtremeprog.shell.treadmillv2)
+  - `AFG Pro Fitness_v1.5.8_apkpure.com.apk`
+  - SHA-256: `e296302f6766eb169b105b2406d0870124f015b2f2b55fb115409ffa480687ca`
+
+## Reverse engineering artifacts
+
+- IDA Pro 7.5 database: `ida/7.4AT-02+V2.001.EFM.bin.idb`
+- JEB 3.28 project: `jeb/AFG Pro Fitness_v1.5.8_apkpure.com.apk.jdb2`
+
+## Instantiating a virtual environment
+
+    python3 -m venv ./env
+    source ./env/bin/activate
+    pip3 install -r requirements.txt
+
+## Analyzing a Bluetooth HCI snoop log
+
+    ./parse_btsnoop.py btsnoop_hci.log

horizon/__init__.py

@@ -1,10 +1,10 @@
 # flake8: noqa
-from .state import State
 from .consts import HEADER_LENGTH
-from .workout_data import WorkoutData
-from .user_info import UserInfo
-from .message import Message
 from .device_info import DeviceInfo
-from .set_speed import SetSpeed
+from .message import Message
 from .set_incline import SetIncline
+from .set_speed import SetSpeed
+from .state import State
+from .user_info import UserInfo
+from .workout_data import WorkoutData
 from .workout_summary import WorkoutSummary

horizon/device_info.py

@@ -19,6 +19,8 @@ def __init__(self, msg: bytes) -> None:
     def format_body(self) -> str:
         return (
             "Device Info:\n"
-            f"  Machine Type:    {self.machineType:>3d}  Model Type: {self.modelType:>3d}\n"
-            f"  Security Switch: {self.securitySwitch:>3d}  Run Status: {self.runStatus:>3d}"
+            f"  Machine Type:    {self.machineType:>3d}"
+            f"  Model Type: {self.modelType:>3d}\n"
+            f"  Security Switch: {self.securitySwitch:>3d}"
+            f"  Run Status: {self.runStatus:>3d}"
         )

horizon/message.py

@@ -24,7 +24,8 @@ def __init__(self, msg: bytes) -> None:
     def format_header(self) -> str:
         return (
             "Header\n"
-            f"  signature:       0x{self.signature.hex()}  sequence number: 0x{self.seq_num:04x}\n"
+            f"  signature:       0x{self.signature.hex()}"
+            f"  sequence number: 0x{self.seq_num:04x}\n"
             f"  msg[4]: 0x{self.msg4:02x}\n"
             f"  msg[5]: 0x{self.msg5:02x}\n"
             f"  length:          0x{self.length:04x}\n"
@@ -49,7 +50,9 @@ def format_raw_msg(self) -> str:
         return string
 
     def __str__(self) -> str:
-        return self.format_type() + "\n" + self.format_header() + "\n" + self.format_body()
+        return (
+            self.format_type() + "\n" + self.format_header() + "\n" + self.format_body()
+        )
 
     def parse_msg_type(self) -> None:
         msg_type = ""
@@ -165,8 +168,10 @@ def parse_msg_type(self) -> None:
 
         if msg_type == "":
             msg_type = (
-                f"Unknown: msg[4]: {self.msg[4:5].hex()}  msg[5]: {self.msg[5:6].hex()}  "
-                f"msg[10]: {self.msg[10:11].hex()}  msg[11]: {self.msg[11:12].hex()}"
+                f"Unknown: msg[4]: {self.msg[4:5].hex()}"
+                f"  msg[5]: {self.msg[5:6].hex()}"
+                f"  msg[10]: {self.msg[10:11].hex()}"
+                f"  msg[11]: {self.msg[11:12].hex()}"
             )
 
         self.type = msg_type

horizon/parsers.py

@@ -1,8 +1,8 @@
-from .workout_data import WorkoutData
-from .user_info import UserInfo
-from .header import Header
 from .device_info import DeviceInfo
+from .header import Header
 from .set_speed import SetSpeed
+from .user_info import UserInfo
+from .workout_data import WorkoutData
 
 
 def parse_header(msg: bytes) -> Header:

horizon/set_incline.py

@@ -11,4 +11,7 @@ def __init__(self, msg: bytes) -> None:
         self.msg12 = msg[12]
 
     def format_body(self) -> str:
-        return f"Set Incline:\n  Incline: {self.incline/10.0:.1f}\n  msg[12]: {self.msg12:02x}"
+        return (
+            f"Set Incline:\n  Incline: {self.incline/10.0:.1f}\n"
+            f"  msg[12]: {self.msg12:02x}"
+        )

horizon/user_info.py

@@ -55,15 +55,16 @@ def format_body(self) -> str:
             f"  User Slot: {self.userSlot:d}\n"
             f"  User Name: {self.userName!r}\n"
             f"  User Weight: {self.userWeight:d}\n"
-            f"  User Birth Year: {self.userBirthYear:d}  User Birth Month: {self.userBirthMonth:d}  "
+            f"  User Birth Year: {self.userBirthYear:d}"
+            f"  User Birth Month: {self.userBirthMonth:d}  "
             f"User Birth Day: {self.userBirthDay:d}\n"
             f"  Units: {self.units:d}\n"
             f"  Custom Program CRC: {self.customProgramCRC:04x}  Custom Heartrate CRC: "
             f"{self.customHeartrateCRC:04x}\n"
             f"  My First 5k Week: {self.myFirst5kWeek:d}  My First 5k Workout: "
             f"{self.myFirst5kWorkout:d}\n"
-            f"  My First 5k Walk Speed: {self.myFirst5kWalkSpeed:d}  My First 5k Jog Speed: "
-            f"{self.myFirst5kJogSpeed:d}\n"
+            f"  My First 5k Walk Speed: {self.myFirst5kWalkSpeed:d}"
+            f"  My First 5k Jog Speed: {self.myFirst5kJogSpeed:d}\n"
             f"  My First 5k Reset Counter: {self.myFirst5kResetCounter:d}\n"
             f"  Map My Fitness Token ID: {self.MMFTokenId!r}\n"
             f"  My Fitness Pal Token ID: {self.MFPTokenId!r}\n"

horizon/workout_summary.py

@@ -1,6 +1,7 @@
-from .message import Message
 from typing import List
 
+from .message import Message
+
 
 class WorkoutSummary(Message):
     userName: str = ""
@@ -45,7 +46,7 @@ def parse_workout_list(self, msg: bytes, workoutCounter: int) -> List:
     def format_body(self) -> str:
         s = (
             "Workout Summary:\n"
-            f"  User Name: \'{self.userName:s}\'  User Id: {self.userId:d}\n"
+            f"  User Name: '{self.userName:s}'  User Id: {self.userId:d}\n"
             f"  MMFTokenId: {self.MMFTokenId:s}\n"
             f"  MFPTokenId: {self.MFPTokenId:s}\n"
             f"  MFPUserId: {self.MFPUserId:s}\n"
@@ -95,7 +96,8 @@ def __init__(self, msg: bytes, start: int):
 
     def __str__(self) -> str:
         return (
-            f"    Time: {self.time:d}  Calories: {self.calories:d}  Distance: {self.distance:d}\n"
+            f"    Time: {self.time:d}  Calories: {self.calories:d}"
+            f"  Distance: {self.distance:d}\n"
             f"    Max Speed: {self.maxSpeed:d}  Average Speed: {self.averageSpeed:d}\n"
             f"    Max HR: {self.maxHR:d}  Average HR: {self.averageHR:d}\n"
             f"    mem10: {self.mem10:d}  Units: {self.units:d}\n"

message_type_notes.txt

@@ -58,4 +58,4 @@ msg[12] (0,1)
 
 0307 setResistance
 msg[10] resistance
-msg[11] (0,1)
+msg[11] (0,1)

parse_btsnoop.py

@@ -89,7 +89,9 @@ def main():
                 and int(pkt.bthci_evt.le_meta_subevent, 16) == 0x02
                 and int(pkt.bthci_evt.le_advts_event_type, 16) == 0x04
                 and "btcommon_eir_ad_entry_device_name" in pkt.bthci_evt.field_names
-                and pkt.bthci_evt.btcommon_eir_ad_entry_device_name.lower().startswith("horizon")
+                and pkt.bthci_evt.btcommon_eir_ad_entry_device_name.lower().startswith(
+                    "horizon"
+                )
             ):
                 # pkt.pretty_print()
                 mac = pkt.bthci_evt.bd_addr

requirements.txt

@@ -1,16 +1,12 @@
+from capstone import CS_ARCH_ARM, CS_MODE_THUMB, Cs
 from unicorn import (
     UC_ARCH_ARM,
-    UC_ERR_MAP,
-    UC_HOOK_BLOCK,
     UC_HOOK_CODE,
     UC_HOOK_MEM_FETCH_UNMAPPED,
-    UC_HOOK_MEM_READ,
-    UC_HOOK_MEM_READ_AFTER,
     UC_HOOK_MEM_READ_UNMAPPED,
     UC_HOOK_MEM_WRITE_UNMAPPED,
     UC_MODE_THUMB,
     Uc,
-    UcError,
 )
 from unicorn.arm_const import (
     UC_ARM_REG_PC,
@@ -22,9 +18,6 @@
     UC_ARM_REG_SP,
 )
 
-from capstone import Cs, CS_ARCH_ARM, CS_MODE_THUMB
-from unicorn.unicorn_const import UC_HOOK_MEM_WRITE
-
 cs = Cs(CS_ARCH_ARM, CS_MODE_THUMB)
 
 PAGE_SIZE = 0x1000
@@ -46,15 +39,17 @@ def hook_mem_read(uc, access, address, size, value, user_data):
 def hook_mem_read_after(uc, access, address, size, value, user_data):
     if address < 0x38000:
         print(
-            ">>> Memory READ at 0x%x, data size = %u, data value = 0x%x" % (address, size, value)
+            ">>> Memory READ at 0x%x, data size = %u, data value = 0x%x"
+            % (address, size, value)
         )
     return True
 
 
 def hook_mem_write(uc, access, address, size, value, user_data):
     if address < 0xF0000000:
         print(
-            ">>> Memory WRITE at 0x%x, data size = %u, data value = 0x%x" % (address, size, value)
+            ">>> Memory WRITE at 0x%x, data size = %u, data value = 0x%x"
+            % (address, size, value)
         )
     return True
 
@@ -101,15 +96,18 @@ def hook_instr(mu: Uc, address, size, user_data):
         # input()
         pass
     if address >= 0x0369E0 and address <= 0x36A00:
-        print(">>> Tracing instruction at 0x%X, instruction size = 0x%X" % (address, size))
+        print(
+            ">>> Tracing instruction at 0x%X, instruction size = 0x%X" % (address, size)
+        )
         R0 = mu.reg_read(UC_ARM_REG_R0)
         R1 = mu.reg_read(UC_ARM_REG_R1)
         R2 = mu.reg_read(UC_ARM_REG_R2)
         R3 = mu.reg_read(UC_ARM_REG_R3)
         R4 = mu.reg_read(UC_ARM_REG_R4)
         PC = mu.reg_read(UC_ARM_REG_PC)
         print(
-            f"R0: {R0:08X}  R1: {R1:08X}  R2: {R2:08X}  R3: {R3:08X}  R4: {R4:08X}  PC: {PC:08X}"
+            f"R0: {R0:08X}  R1: {R1:08X}  R2: {R2:08X}  "
+            f"R3: {R3:08X}  R4: {R4:08X}  PC: {PC:08X}"
         )
         mem = mu.mem_read(address, size)
         for i in cs.disasm(mem, address):
@@ -120,15 +118,18 @@ def hook_instr(mu: Uc, address, size, user_data):
         print()
         # dump_hex_buf(mu, R0, R1)
 
-    mu.last_instr = f">>> Tracing instruction at 0x{address:08X}, instruction size = 0x{size:X}\n"
+    mu.last_instr = (
+        f">>> Tracing instruction at 0x{address:08X}, instruction size = 0x{size:X}\n"
+    )
     R0 = mu.reg_read(UC_ARM_REG_R0)
     R1 = mu.reg_read(UC_ARM_REG_R1)
     R2 = mu.reg_read(UC_ARM_REG_R2)
     R3 = mu.reg_read(UC_ARM_REG_R3)
     R4 = mu.reg_read(UC_ARM_REG_R4)
     PC = mu.reg_read(UC_ARM_REG_PC)
     mu.last_instr += (
-        f"R0: {R0:08X}  R1: {R1:08X}  R2: {R2:08X}  R3: {R3:08X}  R4: {R4:08X}  PC: {PC:08X}\n"
+        f"R0: {R0:08X}  R1: {R1:08X}  R2: {R2:08X}  "
+        f"R3: {R3:08X}  R4: {R4:08X}  PC: {PC:08X}\n"
     )
 
     # branch = False