SCAPY官方教程十一

以下示例假设 Scapy 会话中的 CAN 层已加载。如果不是，可以在 Scapy 会话中使用以下命令加载 CAN 层：

>>> load_layer("can")

创建标准 CAN 帧：

>>> frame = CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')

创建扩展 CAN 帧：

frame = CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')
>>> frame.show()
###[ CAN ]###
  flags= extended
  identifier= 0x10010000
  length= 8
  reserved= 0
  data= '\x01\x02\x03\x04\x05\x06\x07\x08'

 CAN 帧输入和输出

CAN 帧可以从pcap文件中写入和读取：

x = CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08')
wrpcap('/tmp/scapyPcapTest.pcap', x, append=False)
y = rdpcap('/tmp/scapyPcapTest.pcap', 1)

此外，可以从candump输出和日志文件中导入 CAN 帧。类CandumpReader可以像使用socket对象一样使用。这允许您使用sniffScapy 的其他功能：

with CandumpReader("candump.log") as sock:
    can_msgs = sniff(count=50, opened_socket=sock)

DBC 文件格式和 CAN 信号

为了支持 DBC 文件格式， Scapy 中添加SignalFields了SignalPacket 类。SignalFields只能在SignalPacket. 多路复用器字段 (MUX) 可以通过ConditionalFields. 以下示例演示了用法：

DBC Example:

BO_ 4 muxTestFrame: 7 TEST_ECU
 SG_ myMuxer M : 53|3@1+ (1,0) [0|0] ""  CCL_TEST
 SG_ muxSig4 m0 : 25|7@1- (1,0) [0|0] ""  CCL_TEST
 SG_ muxSig3 m0 : 16|9@1+ (1,0) [0|0] ""  CCL_TEST
 SG_ muxSig2 m0 : 15|8@0- (1,0) [0|0] ""  CCL_TEST
 SG_ muxSig1 m0 : 0|8@1- (1,0) [0|0] ""  CCL_TEST
 SG_ muxSig5 m1 : 22|7@1- (0.01,0) [0|0] ""  CCL_TEST
 SG_ muxSig6 m1 : 32|9@1+ (2,10) [0|0] "mV"  CCL_TEST
 SG_ muxSig7 m1 : 2|8@0- (0.5,0) [0|0] ""  CCL_TEST
 SG_ muxSig8 m1 : 0|6@1- (10,0) [0|0] ""  CCL_TEST
 SG_ muxSig9 : 40|8@1- (100,-5) [0|0] "V"  CCL_TEST

BO_ 3 testFrameFloat: 8 TEST_ECU
 SG_ floatSignal2 : 32|32@1- (1,0) [0|0] ""  CCL_TEST
 SG_ floatSignal1 : 7|32@0- (1,0) [0|0] ""  CCL_TEST

此 DBC 描述的 Scapy 实现：

class muxTestFrame(SignalPacket):
    fields_desc = [
        LEUnsignedSignalField("myMuxer", default=0, start=53, size=3),
        ConditionalField(LESignedSignalField("muxSig4", default=0, start=25, size=7), lambda p: p.myMuxer == 0),
        ConditionalField(LEUnsignedSignalField("muxSig3", default=0, start=16, size=9), lambda p: p.myMuxer == 0),
        ConditionalField(BESignedSignalField("muxSig2", default=0, start=15, size=8), lambda p: p.myMuxer == 0),
        ConditionalField(LESignedSignalField("muxSig1", default=0, start=0, size=8), lambda p: p.myMuxer == 0),
        ConditionalField(LESignedSignalField("muxSig5", default=0, start=22, size=7, scaling=0.01), lambda p: p.myMuxer == 1),
        ConditionalField(LEUnsignedSignalField("muxSig6", default=0, start=32, size=9, scaling=2, offset=10, unit="mV"), lambda p: p.myMuxer == 1),
        ConditionalField(BESignedSignalField("muxSig7", default=0, start=2, size=8, scaling=0.5), lambda p: p.myMuxer == 1),
        ConditionalField(LESignedSignalField("muxSig8", default=0, start=3, size=3, scaling=10), lambda p: p.myMuxer == 1),
        LESignedSignalField("muxSig9", default=0, start=41, size=7, scaling=100, offset=-5, unit="V"),
    ]

class testFrameFloat(SignalPacket):
    fields_desc = [
        LEFloatSignalField("floatSignal2", default=0, start=32),
        BEFloatSignalField("floatSignal1", default=0, start=7)
    ]

bind_layers(SignalHeader, muxTestFrame, identifier=0x123)
bind_layers(SignalHeader, testFrameFloat, identifier=0x321)

dbc_sock = CANSocket("can0", basecls=SignalHeader)

pkt = SignalHeader()/testFrameFloat(floatSignal2=3.4)

dbc_sock.send(pkt)

此示例使用类SignalHeader作为标题。有效负载由 individual 指定SignalPackets。 bind_layers将标头与取决于 CAN 标识符的有效载荷组合在一起。如果您想直接SignalPackets从您的 接收，请将CANSocket参数提供basecls给您的init函数CANSocket。

② CANSockets

Linux SocketCAN

虚拟 CAN 设置

Linux SocketCAN 支持虚拟 CAN 接口。这些接口是在 CAN 总线上执行一些初步步骤的简单方法，无需特殊硬件。除此之外，虚拟 CAN 接口在 Scapy 单元测试中大量使用，用于与汽车相关的贡献。

虚拟 CAN 套接字需要一个特殊的 Linux 内核模块。以下 shell 命令加载所需的模块：

sudo modprobe vcan

为了使用虚拟 CAN 接口，需要一些额外的设置命令。此代码段选择vcan0虚拟 CAN 接口的名称。可以在这里选择任何名称：

sudo ip link add name vcan0 type vcan
sudo ip link set dev vcan0 up

可以从 Scapy 执行相同的命令，如下所示：

from scapy.layers.can import *
import os

bashCommand = "/bin/bash -c 'sudo modprobe vcan; sudo ip link add name vcan0 type vcan; sudo ip link set dev vcan0 up'"
os.system(bashCommand)

如果需要，可以使用命令将 CAN 接口设置为listen-onlyorloopback模式：ip link set

ip link set vcan0 type can help  # shows additional information

Linux can-utils

下面的例子展示了 Linux can-utils 的基本功能。这些实用程序对于从命令行快速检查、转储、发送或记录 CAN 消息非常方便。

Scapy CANSocket

在 Scapy 中，实现了两种 CANSocket。一种实现称为Native CANSocket，另一种实现称为Python-can CANSocket。

由于 Python 3 支持PF_CAN套接字，因此本机 CANSockets可以在具有 Python 3 或更高版本的基于 Linux 的系统上使用。这些套接字具有性能优势，因为select它们是可调用的。这在 MITM 场景中具有很大的影响。

原生 CANSocket

创建一个简单的原生 CANSocket：

conf.contribs['CANSocket'] = {'use-python-can': False} #(default)
load_contrib('cansocket')

# Simple Socket
socket = CANSocket(channel="vcan0")

创建本机 CANSocket 仅侦听 Id == 0x200 的消息：

socket = CANSocket(channel="vcan0", can_filters=[{'can_id': 0x200, 'can_mask': 0x7FF}])

创建本机 CANSocket 仅侦听 Id >= 0x200 和 Id <= 0x2ff 的消息：

socket = CANSocket(channel="vcan0", can_filters=[{'can_id': 0x200, 'can_mask': 0x700}])

创建本机 CANSocket 仅侦听 Id != 0x200 的消息：

socket = CANSocket(channel="vcan0", can_filters=[{'can_id': 0x200 | CAN_INV_FILTER, 'can_mask': 0x7FF}])

创建具有多个 can_filters 的本机 CANSocket：

socket = CANSocket(channel='vcan0', can_filters=[{'can_id': 0x200, 'can_mask': 0x7ff},
                                               {'can_id': 0x400, 'can_mask': 0x7ff},
                                               {'can_id': 0x600, 'can_mask': 0x7ff},
                                               {'can_id': 0x7ff, 'can_mask': 0x7ff}])

创建一个本机 CANSocket，它也接收自己的消息：

socket = CANSocket(channel="vcan0", receive_own_messages=True)

嗅探 CANSocket：

 CANSocket python-can

在 Windows、OSX 或 Linux 上使用各种 CAN 接口需要 python-can。python-can 库通过 CANSocket 对象使用。要创建 python-can CANSocket 对象，interface.Bus必须使用 python-can 对象的所有参数来初始化 CANSocket。

创建python-can CANSocket的方法：

conf.contribs['CANSocket'] = {'use-python-can': True}
load_contrib('cansocket')

创建一个简单的python-can CANSocket：

socket = CANSocket(bustype='socketcan', channel='vcan0', bitrate=250000)

创建一个带有多个过滤器的 python-can CANSocket：

socket = CANSocket(bustype='socketcan', channel='vcan0', bitrate=250000,
                can_filters=[{'can_id': 0x200, 'can_mask': 0x7ff},
                            {'can_id': 0x400, 'can_mask': 0x7ff},
                            {'can_id': 0x600, 'can_mask': 0x7ff},
                            {'can_id': 0x7ff, 'can_mask': 0x7ff}])

CANSocket MITM 攻击与桥接和嗅探

这个例子展示了如何在虚拟 CAN 接口上使用桥接和嗅探。对于现实世界的应用，请使用真正的 CAN 接口。在 Linux 终端上设置两个 vcan：

sudo modprobe vcan
sudo ip link add name vcan0 type vcan
sudo ip link add name vcan1 type vcan
sudo ip link set dev vcan0 up
sudo ip link set dev vcan1 up

导入模块：

import threading
load_contrib('cansocket')
load_layer("can")

创建可以攻击的套接字：

socket0 = CANSocket(channel='vcan0')
socket1 = CANSocket(channel='vcan1')

创建一个函数以使用线程发送数据包：

def sendPacket():
    sleep(0.2)
    socket0.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08'))

创建转发或更改数据包的功能：

def forwarding(pkt):
    return pkt

创建一个在两个套接字之间桥接和嗅探的函数：

def bridge():
    bSocket0 = CANSocket(channel='vcan0')
    bSocket1 = CANSocket(channel='vcan1')
    bridge_and_sniff(if1=bSocket0, if2=bSocket1, xfrm12=forwarding, xfrm21=forwarding, timeout=1)
    bSocket0.close()
    bSocket1.close()

创建用于发送数据包以及桥接和嗅探的线程：

threadBridge = threading.Thread(target=bridge)
threadSender = threading.Thread(target=sendMessage)

启动线程：

threadBridge.start()
threadSender.start()

嗅探数据包：

packets = socket1.sniff(timeout=0.3)

关闭套接字：

socket0.close()
socket1.close()

 ③ CAN 校准协议 (CCP)

CCP源自CAN。CAN 报头是 CCP 帧的一部分。CCP 有两种类型的消息对象。一种称为命令接收对象（CRO），另一种称为数据传输对象（DTO）。通常 CRO 被发送到 Ecu，而 DTO 从 Ecu 接收。如果一个 DTO 回答 CRO，则该信息是通过计数器字段 (ctr) 实现的。如果两个对象具有相同的计数器值，则可以从关联的 CRO 对象的命令中解释 DTO 对象的有效负载。

创建 CRO 消息：

load_contrib('automotive.ccp')
CCP(identifier=0x700)/CRO(ctr=1)/CONNECT(station_address=0x02)
CCP(identifier=0x711)/CRO(ctr=2)/GET_SEED(resource=2)
CCP(identifier=0x711)/CRO(ctr=3)/UNLOCK(key=b"123456")

如果我们对 Ecu 的 DTO 不感兴趣，我们可以像这样发送 CRO 消息：发送 CRO 消息：

pkt = CCP(identifier=0x700)/CRO(ctr=1)/CONNECT(station_address=0x02)
sock = CANSocket(bustype='socketcan', channel='vcan0')
sock.send(pkt)

如果我们对 Ecu 的 DTO 感兴趣，我们需要将 CANSocket 的 basecls 参数设置为 CCP，我们需要使用 sr1：发送 CRO 消息：

cro = CCP(identifier=0x700)/CRO(ctr=0x53)/PROGRAM_6(data=b"\x10\x11\x12\x10\x11\x12")
sock = CANSocket(bustype='socketcan', channel='vcan0', basecls=CCP)
dto = sock.sr1(cro)
dto.show()
###[ CAN Calibration Protocol ]###
  flags=
  identifier= 0x700
  length= 8
  reserved= 0
###[ DTO ]###
     packet_id= 0xff
     return_code= acknowledge / no error
     ctr= 83
###[ PROGRAM_6_DTO ]###
        MTA0_extension= 2
        MTA0_address= 0x34002006

由于 sr1 调用了 answers 函数，我们的 DTO 对象的有效负载被我们的 CRO 对象的命令解释。

④ 通用校准和测量协议 (XCP)

XCP 是 CCP 的继承者。它可用于多种协议。Scapy 包括 CAN、UDP 和 TCP。XCP 有两种消息类型：命令传输对象（CTO）和数据传输对象（DTO）。CTO 发送给 Ecu 的是请求（命令），Ecu 必须以肯定响应或错误来回复。此外，Ecu 可以发送 CTO 通知主设备异步事件 (EV) 或请求服务执行 (SERV)。Ecu 发送的 DTO 称为 DAQ（数据采集），包括测量值。Ecu 接收的 DTO 用于周期性刺激，称为 STIM（刺激）。

创建 CTO 消息：

CTORequest() / Connect()
CTORequest() / GetDaqResolutionInfo()
CTORequest() / GetSeed(mode=0x01, resource=0x00)

要通过 CAN 发送消息，必须添加标头：

pkt = XCPOnCAN(identifier=0x700) / CTORequest() / Connect()
sock = CANSocket(iface=can.interface.Bus(bustype='socketcan', channel='vcan0'))
sock.send(pkt)

如果我们对 Ecu 的响应感兴趣，我们需要将 CANSocket 的 basecls 参数设置为 XCPonCAN 并且我们需要使用 sr1：发送 CTO 消息：

sock = CANSocket(bustype='socketcan', channel='vcan0', basecls=XCPonCAN)
dto = sock.sr1(pkt)

由于 sr1 调用了 answers 函数，我们的 XCP 响应对象的有效负载被我们的 CTO 对象的命令解释。否则无法解释。第一条消息应该始终是“CONNECT”消息，Ecu 的响应决定了消息的读取方式。例如：字节顺序。否则，必须在 contrib 配置中设置地址粒度以及 DTO 和 CTO 的最大大小：

conf.contribs['XCP']['Address_Granularity_Byte'] = 1  # Can be 1, 2 or 4
conf.contribs['XCP']['MAX_CTO'] = 8
conf.contribs['XCP']['MAX_DTO'] = 8

如果您不希望在收到消息后设置此项，您也可以禁用该功能：

conf.contribs['XCP']['allow_byte_order_change'] = False
conf.contribs['XCP']['allow_ag_change'] = False
conf.contribs['XCP']['allow_cto_and_dto_change'] = False

要通过 TCP 或 UDP 发送 pkt，必须使用另一个标头。TCP：

prt1, prt2 = 12345, 54321
XCPOnTCP(sport=prt1, dport=prt2) / CTORequest() / Connect()

UDP：

XCPOnUDP(sport=prt1, dport=prt2) / CTORequest() / Connect()