体組成計とスマホ間のBluetooth通信をハッキングしてみる

###経緯

2年程前にスマホで測定結果を管理できる体組成計を購入しました。

当時は物珍しいこともあり、毎日欠かさずに測定していました。

ところがある日を境に測定をやめてしまいました。

理由は、測定時にいちいちスマホアプリを起動しなければならず、面倒になったからです。

測定するときは大抵が風呂上がりで、湿った手でスマホを触るのも嫌でした。

そこでスマホアプリを起動せずに何とか測定結果を管理できないか考えてみました。

体組成計はBluetoothでデータをやり取りします。

家には常時起動のミニサーバ(Raspberry Pi 3)があり、Bluetoothを標準装備しています。

ミニサーバが体組成計の起動を自動で検知して、測定データをまとめてくれたら嬉しいです。

少し調べてみたら、どうやらPythonからBluetoothを弄れそうだと判明し、ハッキングを決意しました。

###ハッキング対象

対象は、DIKIの体組成計です。Amazonで2,500円程のお手頃なものです。
http://amzn.asia/d/9dIJnMN

###通信の観察

スマホはアンドロイドを利用しています。

アンドロイドにはBluetoothのスヌープログを出力する機能があり、ログをWiresharkで開けばパケットを観察できます。

早速、設定の開発者向けオプションで「Bluetooth HCIスヌープログを有効にする」にチェックを入れます。

そして測定アプリを開き、体組成計と接続して自分の体重を計ります。

その後、スマホをPCにUSB接続します。

コマンドプロンプトを開き、スヌープログの在りかを確認します。

adb shell cat /etc/bluetooth/bt_stack.conf

僕の環境では、以下にありました。

/sdcard/btsnoop_hci.log

これをスマホから採取します。

adb pull /sdcard/btsnoop_hci.log

PCでWiresharkを起動し、スヌープログを開きます。

数分程眺めていると、ATTというプロトコルでデータをやり取りしていることが分かりました。

そして通信の最後の方に以下のようなデータがありました。

ac02ff000221120904161b1a02af00dd00b20000
010007020205f6001c02591e00b001011eb00216

他のパケットのデータと違って長いのでこれが測定結果かもしれません。

16進数の文字列です。

スマホアプリの測定結果と照らし合わせて、どこかにデータが隠れてないか探したところ、以下が判明しました。

ac02 ff00 0221 1209 0416 1b1a 02af 00dd 00b2 0000
                              ↑体重 ↑BMI ↑体脂肪率        

0100 07           0202   05f6     001c  0259   1e       00b0     0101 1eb0 0216
     ↑内臓脂肪指数 ↑筋肉率 ↑基礎代謝 ↑骨量 ↑体水分 ↑身体年齢 ↑タンパク質率

測定結果がどのように送付されてくるのかが分かりました。

###通信の解析

続いて、その他のデータにも着目してみます。

体組成計である以上、スマホ側から性別や身長データを送っているはずです。

何度か測定し、見比べたりしながら特徴を見極めます。

目を凝らして見てみると、どうやら次のようなプロトコルでデータをやり取りしていそうです。

□スマホ⇒体組成計
△体組成計⇒スマホ

１．コマンド送信開始(□)
　　ac02f7000000ccc3

２．コマンド取得開始(△)
　　ac02f70c020eccdf

３．ユーザリスト送信開始(□)
　　ac02ff000000cfce

４．ユーザリスト取得開始(△)
　　ac02fe000000cfcd

５．ユーザリスト送信(□)
　　1回の送信で2ユーザまで送信可能

ac02fd00 01       01          1e    af    02ad       0255       02        02         1c    9e    02ad       0241
ac02fd00 03       02          1a    9a    0000       0000       00        00         00    00    0000       0000
         ↑ユーザNo ↑男(1)女(2) ↑年齢 ↑身長  ↑前回の体重 ↑前回の何か ↑ユーザNo ↑男(1)女(2) ↑年齢 ↑身長 ↑前回の体重 ↑前回の何か

６．ユーザリスト送信完了(□)
　　ac02fd020000cfce

７．ユーザリスト取得完了(△)
　　ac02fc000000cfcb

８．送信待ち(□)
　　一定時間の待ち時間が常にある。約0.5秒

９．今回の測定ユーザ情報１送信(□)

　　ac02fa 03       0000ccc9
　　        ↑ユーザNo

１０．今回の測定ユーザ情報２送信(□)

　　ac02fb  02          1a    9a     cc7d 
　　        ↑男(1)女(2)  ↑年齢 ↑身長

１１．ユーザ情報取得完了(△)
　　ac02fe090000ccd3

１２．コマンド送信完了(□)
　　ac02fe060000ccd0

１３．体重データ送信(△)
　　※測定中の体重データを連続で送信してきます

１４．測定データ送信(△)
　　※上記の体重、BMI、体脂肪などすべてのデータを最後に送信してきます。

１５．測定データ送信完了(△)
　　ac02fe010000cccb

項目名は、データから推測して勝手に名付けました。

###Raspberry PiのPythonでBluetoothを操作する

他ページが詳しいので割愛します。

★環境整備／サンプルプログラム
https://qiita.com/Peeeeepei/items/58cd6eae81784f17205f

★リファレンス
https://www.ratoc-e2estore.com/blog/bluepy_peripheral

###体組成計からデータを取得するコード

ラズパイで実行するPythonコードを記します。

体組成計のBDアドレスはスヌープログから取得しました。

UUIDは、スマホ⇒体組成計のパケットのものを利用します。

上記の「１１．ユーザ情報取得完了(△)」のパケットは、なぜか作成プログラムでは受信できなかったので無視することにしました。

WeightScan.py

# -*- coding: utf-8 -*-

from bluepy.btle import Peripheral
import bluepy.btle as btle
import binascii
from time import sleep

# 体組成計のBDアドレス
BD_ADDR = "02:B3:EC:02:9F:CF"

# UUID
UUID = "0000ffb1-0000-1000-8000-00805f9b34fb"

# ユーザデータ
userdata = {
    "user_list": "\xac\x02\xfd\x00\x01\x01\x1e\xaf\x02\xaf\x02\x16\x00\x00\x00\x00\x00\x00\x00\x00",
    "user_info1": "\xac\x02\xfa\x01\x00\x00\xcc\xc7",
    "user_info2": "\xac\x02\xfb\x01\x1e\xaf\xcc\x96"
    }

# 送信データ
sendcom = {
    "send_start": "\xac\x02\xf7\x00\x00\x00\xcc\xc3",
    "send_ul_start": "\xac\x02\xff\x00\x00\x00\xcf\xce",
    "send_ul_end": "\xac\x02\xfd\x02\x00\x00\xcf\xce",
    "send_end": "\xac\x02\xfe\x06\x00\x00\xcc\xd0"
    }

# 送信データに対応するACK
recvcom = {
    "recv_start": "ac02f70c020eccdf",
    "recv_ul_start": "ac02fe000000cfcd",
    "recv_ul_end": "ac02fc000000cfcb",
    "recv_end": "ac02fe010000cccb"
    }

# 状態管理変数
status = ""
result1 = ""
result2 = ""

# データを受信した時に呼ばれる
class MyDelegate(btle.DefaultDelegate):
    def __init__(self, params):
        btle.DefaultDelegate.__init__(self)

    def handleNotification(self, cHandle, data):
        global status
        global result1
        global result2
        c_data = binascii.b2a_hex(data)

        if c_data.find(recvcom["recv_start"]) > -1:
            print("recv_start")
            status = "recv_start"
        elif c_data.find(recvcom["recv_ul_start"]) > -1:
            print("recv_ul_start")
            status = "recv_ul_start"
        elif c_data.find(recvcom["recv_ul_end"]) > -1:
            print("recv_ul_end")
            status = "recv_ul_end"
        elif len(c_data) == 40:
            result1 = result2
            result2 = c_data
        elif c_data.find(recvcom["recv_end"]) > -1:
            print("recv_end")
            status = "recv_end"

# 体組成計クラス
class WeightScan(Peripheral):
    def __init__(self, addr):
        Peripheral.__init__(self, addr)
        self.result = 1

def main():
    # 初期設定
    print("Waiting...")
    while True:
        try:
            ws = WeightScan(BD_ADDR)
        except Exception as e:
            print("Waiting...")
        else:
            break
    ws.setDelegate(MyDelegate(btle.DefaultDelegate))
    print("Delegate ready")

    # UUIDからハンドルを取得
    charlist = ws.getCharacteristics()
    handle = 0
    for c in charlist:
        if str(c.uuid) == UUID:
            handle = c.handle + 1

    # コマンド送信開始
    ws.writeCharacteristic(handle, sendcom["send_start"])
    print("send_start")

    # データ受信待ち
    while True:
        if ws.waitForNotifications(1.0):
            if status == "recv_start":
                break

    # ユーザリスト送信開始
    ws.writeCharacteristic(handle, sendcom["send_ul_start"])
    print("send_ul_start")

    # データ受信待ち
    while True:
        if ws.waitForNotifications(1.0):
            if status == "recv_ul_start":
                break

    # ユーザリスト送信
    ws.writeCharacteristic(handle, userdata["user_list"])
    print("user_list")

    # ユーザリスト送信完了
    ws.writeCharacteristic(handle, sendcom["send_ul_end"])
    print("send_ul_end")

    # データ受信待ち
    while True:
        if ws.waitForNotifications(1.0):
            if status == "recv_ul_end":
                break

    # 送信待ち
    sleep(0.55)

    # 測定者情報送信
    ws.writeCharacteristic(handle, userdata["user_info1"])
    print("user_info1")

    ws.writeCharacteristic(handle, userdata["user_info2"])
    print("user_info2")

    # データ受信待ち
    while True:
        if ws.waitForNotifications(1.0):
            break

    # コマンド送信完了
    ws.writeCharacteristic(handle, sendcom["send_end"])
    print("send_end")

    # データ受信待ち
    while True:
        if ws.waitForNotifications(1.0):
            if status == "recv_end":
                break
            continue

    # 結果表示
    wr = WeightResult(result1, result2)
    print("***************************************")
    print("測定結果")
    print("***************************************")
    print("体重        ：\t" + str(wr.getWeight())) + "kg"
    print("BMI         ：\t" + str(wr.getBMI()))
    print("体脂肪率    ：\t" + str(wr.getFatper())) + "%"
    print("内臓脂肪指数：\t" + str(wr.getVisceralfat()))
    print("筋肉率      ：\t" + str(wr.getMuscleper())) + "%"
    print("基礎代謝    ：\t" + str(wr.getBMR())) + "kcal"
    print("骨量        ：\t" + str(wr.getBonemass())) + "kg"
    print("体水分率    ：\t" + str(wr.getBodywaterper())) + "%"
    print("身体年齢    ：\t" + str(wr.getBodyage())) + "歳"
    print("***************************************")

# データ解析クラス
class WeightResult:
    def __init__(self, result1, result2):
        self.result1 = result1
        self.result2 = result2

    # 体重
    def getWeight(self):
        substr = result1[24:28]
        ret = float(int(substr, 16)) / 10
        return ret

    # BMI
    def getBMI(self):
        substr = result1[28:32]
        ret = float(int(substr, 16)) / 10
        return ret

    # 体脂肪率
    def getFatper(self):
        substr = result1[32:36]
        ret = float(int(substr, 16)) / 10
        return ret

    # 内臓脂肪指数
    def getVisceralfat(self):
        substr = result2[4:6]
        ret = int(substr, 16)
        return ret

    # 筋肉率
    def getMuscleper(self):
        substr = result2[6:10]
        ret = float(int(substr, 16)) / 10
        return ret

    # 基礎代謝
    def getBMR(self):
        substr = result2[10:14]
        ret = int(substr, 16)
        return ret

    # 骨量
    def getBonemass(self):
        substr = result2[14:18]
        ret = float(int(substr, 16)) / 10
        return ret

    # 体水分率
    def getBodywaterper(self):
        substr = result2[18:22]
        ret = float(int(substr, 16)) / 10
        return ret

    # 身体年齢
    def getBodyage(self):
        substr = result2[22:24]
        ret = int(substr, 16)
        return ret

    # タンパク質率
    def getProteinper(self):
        substr = result2[24:28]
        ret = float(int(substr, 16)) / 10
        return ret

    # 前回の何か
    def getLastmark(self):
        return result2[36:40]

if __name__ == "__main__":
    main()

データ解析クラスで「前回の何か」というデータを取得できるようにしたのは、

体組成計が前回の測定結果を覚えているようで、「前回の何か」を照合している気がするからです。

何度か測定していると突然測定できなくなります。

そんな時は電池を抜き差しすれば「前回の何か」がリセットされ再び測れるようになります。

未検証ですが、ユーザリストの「前回の何か」の部分に入れてやれば改善すると思われます。

###実行結果

実行結果を示します。

運動します。。。

Waiting...
Waiting...
Delegate ready
send_start
recv_start
send_ul_start
recv_ul_start
user_list
send_ul_end
recv_ul_end
user_info1
user_info2
send_end
recv_end
***************************************
測定結果
***************************************
体重        ：  68.4kg
BMI         ：  22.3
体脂肪率    ：  19.2%
内臓脂肪指数：  7
筋肉率      ：  50.3%
基礎代謝    ：  1498kcal
骨量        ：  2.8kg
体水分率    ：  59.2%
身体年齢    ：  30歳
***************************************

###まとめ

今回作成したプログラムを修正すれば、目的のミニサーバでの体組成データ管理機能は実現できそうです。

収集したデータをグラフで見れるようにしてしまえば完璧です。

ただ、誰が体組成計に乗ったかは判別できないため、誰か一人専用になってしまいますね。

まあ安いしコンパクトなので問題ないですが。