如何模糊测试SOME/IP协议（一）

汽车安全 1周前 admin

55 0 0

安全研究员onepwnman报告了在BMW车辆MGU 中使用SOME/IP的应用程序中的几个安全漏洞。因此，被列入宝马名人堂。在本文中，将介绍SOME/IP协议并分享如何模糊SOME/IP应用程序以查找漏洞。

What is SOME/IP Protocol?

传统上，在汽车行业中的车载网络主要采用基于信号的CAN通信。即使到现在，CAN依然是车辆内部广泛使用的协议之一。近年来，除了CAN通信之外，以太网为基础的通信越来越受到青睐。AUTOSAR标准中的一个以太网车载协议是SOME/IP（Scalable Service-Oriented MiddlewarE over IP），这是一种面向服务的协议。宝马集团使用的是vsomeip开源实现版本的SOME/IP协议。（https://github.com/COVESA/vsomeip）

下图是基于SOME/IP的车载网络通信方案。

如何模糊测试SOME/IP协议（一）

SOME/IP消息可以使用TCP或UDP协议。服务接口通过端口号来区分，因此相同的接口不能在同一端口上提供多个实例。

消息类型如下图所示，分为四种支持方式。

如何模糊测试SOME/IP协议（一）

Request/Response与Fire & Forget的区别在于是否有响应消息。Events是一种消息类型，当服务器发生事件时会回调。

关于Subscribe的更多信息，请参见SOME/IP-SD协议。我们将在第二部分中详细解释，首先让我们了解一下SOME/IP的数据包结构。

SOME/IP Packet Structure

首先，为了发送一个普通的SOME/IP消息，你需要了解SOME/IP头部的结构。下图是从标准中摘录的SOME/IP头部格式。

如何模糊测试SOME/IP协议（一）

• Message ID

• 每项服务都有一个唯一的Service ID。

• 每项服务可以有多种方法，每种方法都有自己的ID。

如何模糊测试SOME/IP协议（一）

• Length

• 从Request ID到有效负载末尾的长度。

• Request ID

• Client ID允许ECU区分对同一方法来自多个客户端的调用。

• Session ID允许区分来自同一发送者的顺序消息或请求。

如何模糊测试SOME/IP协议（一）

• Protocol Version

• 它通常有一个固定值。

• Interface Version

• 它通常有一个固定值。

• Message Type

• Message Type字段根据上述四种消息类型的不通而不同。例如，在事件消息的情况下，消息类型是0x2 notification，并且发送方不期望收到响应。

如何模糊测试SOME/IP协议（一）

• 返回代码正常消息的返回代码通常为0x00（E_OK）。

如何模糊测试SOME/IP协议（一）

来看一个例子。车载信息娱乐系统大多具有蓝牙功能，这使得智能手机能够连接到车辆。在这种情况下，假设有一个叫做DeviceManager的服务，用于管理连接到车辆的设备。DeviceManager可以有各种方法，比如将车辆连接到乘客的智能手机，检查连接状态，切换到另一个设备的连接，以及断开设备连接。此外车辆中的其他控制器可能也需要以事件格式接收设备连接状态，以及向车载信息娱乐系统发送请求/响应格式的数据请求。这样，一个服务通常由多种方法组成，并支持多种消息类型。

Payload Structure

SOME/IP的有效负载在传输前会被序列化，接收端则通过反序列化过程获取数据。因此，随机模糊测试（发送随机字节到有效负载）不太可能将所需的数据传送到应用层。

为了进行结构化的模糊测试，你需要理解有效负载的变量类型和序列化方式。
在宝马MGU 22的实际案例中，它使用了COVESA的SOME/IP相关库，每个数据类型的实现都在以下链接的代码中。(https://github.com/COVESA/vsomeip/wiki/vsomeip-in-10-minutes#prep)

这里有一个表格，展示了一些数据类型。

Data Type	Byte Count
UInt8	1
UInt16	2
UInt32	4
UInt64	8
Bool	1
Enum	1, 2
Utf8	Depends on the length of the string
Utf16	Depends on the length of the string
Array	Depends on the length of the string
Map	Depends on the length of the string
Struct	0

以下示例是为每种数据类型用Python实现的代码，返回值是一个十六进制字符串。

def _uint8(p):    return "{:02X}".format(p)
def _uint16(p):    return "{:04X}".format(p)
def _uint32(p):    return "{:08X}".format(p)
def _uint64(p):    return "{:016X}".format(p)
def _bool(p):    return "{:02X}".format(p)
def _enum(p, w=False):    if w:        return "{:04X}".format(p)    else:        return "{:02X}".format(p)
def _utf8(p):    data = "EFBBBF"    data += p.encode("utf-8").hex()    data += "00"    return "{:08X}".format(len(data)//2) + data
def _utf16(p):    data = "FEFF"    data += p.encode("utf-16-be").hex()    data += "0000"    return "{:08X}".format(len(data)//2) + data
def _array(p):    return "{:08X}".format(len(p)//2) + p
def _map(k, v):    data = k + v    return "{:08X}".format(len(data)//2) + data

例如，UTF-16字符序列“TEST”可以表示为如下形式。

print(_utf16("TEST"))
# the length of a string 0000000C# actual string content "FEFF" + "0054004500530054" + "0000"# return value 0000000CFEFF00540045005300540000

此外，有效负载是由多种类型的序列化数据组成的。序列化数据仅仅是一系列的数据类型。如果一个数据类型需要长度，那么长度会被加在这个数据之前。例如，我们考虑以下几种数据类型的序列化：

Struct{  UInt32,  Array  {    Struct    {      Utf8,      Utf8    },  },  Map  {    Enum,    Utf8  },  Bool}

以下数据类型按照以下顺序进行序列化：

Struct - UInt32 - Array - Struct - Utf8 - Utf8 - Map - Enum - Utf8 - Bool

在序列化过程中，Struct 被忽略。Struct 内部主要有三种数据类型：
Array、Map 和 Bool

Array 包含 Struct、Utf8 和 Utf8 类型。这里，Struct 也被忽略，在 Python 代码中，它变成

 _array(_utf8("str1") + _utf8("str2"))

对于 Map，也可以表示为

_map(_enum(data1), _utf8("str3"))

因此，整个有效负载可以用如下代码表示：

payload =  _uint32(data1)payload += _array(_utf8("str1") + _utf8("str2"))payload += _map(_enum(data2) + _utf8("str3"))payload += _bool(data3)

总之，发送序列化后的SOME/IP消息的代码如下：

import socket import datetime

class SomeIP:    def __init__(self, service_id, method_id, client_id, session_id, protocol_version, interface_version, message_type, return_code):        self.service_id = service_id            self.method_id = method_id        self.client_id = client_id        self.session_id = session_id        self.protocol_version = protocol_version        self.interface_version = interface_version        self.message_type = message_type        self.return_code = return_code
    def assemble_packet(self, payload):        self.header =  "{:04X}".format(self.service_id)        self.header += "{:04X}".format(self.method_id)         self.header += "{:08X}".format(len(payload) + 8)        self.header += "{:04X}".format(self.client_id)        self.header += "{:04X}".format(self.session_id)         self.header += "{:02X}".format(self.protocol_version)        self.header += "{:02X}".format(self.interface_version)        self.header += "{:02X}".format(self.message_type)        self.header += "{:02X}".format(self.return_code)        self.packet = bytes.fromhex(self.header) + payload

def main():    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)    sock.connect(("160.48.199.99", 32504))    sock.settimeout(0.1)
    def _uint8(p):        return "{:02X}".format(p)    def _uint16(p):        return "{:04X}".format(p)    def _uint32(p):        return "{:08X}".format(p)    def _uint64(p):        return "{:016X}".format(p)    def _bool(p):        return "{:02X}".format(p)    def _enum(p, w=False):        if w:            return "{:04X}".format(p)        else:            return "{:02X}".format(p)
    def _utf8(p):        data = "EFBBBF"        data += p.encode("utf-8").hex()        data += "00"        return "{:08X}".format(len(data)//2) + data
    def _utf16(p):        data = "FEFF"        data += p.encode("utf-16-be").hex()        data += "0000"        return "{:08X}".format(len(data)//2) + data
    def _array(p):        return "{:08X}".format(len(p)//2) + p
    def _map(k, v):        data = k + v        return "{:08X}".format(len(data)//2) + data
    def someip_func(data1, str1, str2, data2, str3, data3) -> SomeIP:        someip = SomeIP(45246, 1, 0, 0, 1, 1, 0, 0)        payload =  _uint32(data1)        payload =  _array(_utf8(str1) + _utf8(str2))        payload += _map(_enum(data2), _utf8(str3))        payload += _bool(data3)        someip.assemble_packet(bytes.fromhex(payload))        return someip
    def send_and_recv(someip):        try:            sock.send(someip.packet)            print(f"{datetime.datetime.now()}: {someip.packet.hex()}")            res = sock.recv(1024)            print(f"Recv: {res.hex()}")        except socket.timeout:            pass

    send_and_recv(someip_func(1, "aaa", "bbb", 2, "ccc", 3))

if __name__ == "__main__":    main()

上面的代码是向IP地址160.48.199.99的32504端口发送一个序列化消息。服务ID是45246，方法ID是1。消息模式是请求/响应。

在这种情况下，会话ID和客户端ID可以设置为0，协议版本固定为1。接口版本通常的取值范围是0到3。

上述代码的结果被编码为字节并带有以下十六进制值发送：

b0be00010000003300000000010100000000001600000007efbbbf6161610000000007efbbbf626262000000000c0200000007efbbbf6363630003

我们现在准备好了发送SOME/IP消息。但是，我们还不知道要发送的目标的IP地址和端口、协议类型（TCP或UDP）、服务ID和服务方法ID、消息类型（请求/响应还是事件）以及有效负载中序列化的数据类型。如果我们通过发送随机数据来进行模糊测试，覆盖率将会非常低。因此，我们需要了解SOME/IP-SD协议以知晓各个元素。

SOME/IP-SD指的是SOME/IP服务发现协议。它用于查找服务实例，检测服务实例是否正在运行，并实现发布/订阅处理。

在第二部分中，我们将学习SOME/IP-SD的工作原理，将PC连接到车辆网络，并分析SOME/IP-SD的数据包结构。

如何模糊测试SOME/IP协议（一）