% Regression tests for nativecansocket ~ not_pypy vcan_socket needs_root linux # More information at http://www.secdev.org/projects/UTscapy/ ############ ############ + Configuration of CAN virtual sockets ~ conf = Load module load_layer("can", globals_dict=globals()) conf.contribs['CANSocket'] = {'use-python-can': False} from scapy.contrib.cansocket_native import * conf.contribs['CAN'] = {'swap-bytes': False, 'remove-padding': True} = Setup string for vcan bashCommand = "/bin/bash -c 'sudo modprobe vcan; sudo ip link add name vcan0 type vcan; sudo ip link set dev vcan0 up'" = Load os import os import threading from time import sleep from subprocess import call = Setup vcan0 assert 0 == os.system(bashCommand) + Basic Packet Tests() = CAN Packet init canframe = CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08') assert bytes(canframe) == b'\x00\x00\x07\xff\x08\x00\x00\x00\x01\x02\x03\x04\x05\x06\x07\x08' + Basic Socket Tests() = CAN Socket Init sock1 = CANSocket(channel="vcan0") = CAN Socket send recv small packet without remove padding conf.contribs['CAN'] = {'swap-bytes': False, 'remove-padding': False} sock2 = CANSocket(channel="vcan0") sock2.send(CAN(identifier=0x7ff,length=1,data=b'\x01')) sock2.close() rx = sock1.recv() print(repr(rx)) assert rx == CAN(identifier=0x7ff,length=1,data=b'\x01') / Padding(b"\x00" * 7) = CAN Socket send recv small packet conf.contribs['CAN'] = {'swap-bytes': False, 'remove-padding': True} sock2 = CANSocket(channel="vcan0") sock2.send(CAN(identifier=0x7ff,length=1,data=b'\x01')) sock2.close() rx = sock1.recv() print(repr(rx)) assert rx == CAN(identifier=0x7ff,length=1,data=b'\x01') = CAN Socket send recv sock2 = CANSocket(channel="vcan0") sock2.send(CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.close() rx = sock1.recv() assert rx == CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08') = CAN Socket basecls test sock2 = CANSocket(channel="vcan0") sock2.send(CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.close() sock1.basecls = Raw rx = sock1.recv() assert rx == Raw(bytes(CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08'))) sock1.basecls = CAN + Advanced Socket Tests() = CAN Socket sr1 tx = CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08') = CAN Socket sr1 init time assert tx.sent_time == None sock2 = CANSocket(channel="vcan0") sock2.send(tx) sock2.close() rx = None rx = sock1.sr1(tx, verbose=False, timeout=3) assert rx == tx sock1.close() = CAN Socket sr1 time check assert abs(tx.sent_time - rx.time) < 0.1 assert rx.time > 0 = sr can tx = CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08') = sr can check init time assert tx.sent_time == None sock1 = CANSocket(channel="vcan0") sock2 = CANSocket(channel="vcan0") sock2.send(tx) sock2.close() rx = None rx = sock1.sr(tx, timeout=1, verbose=False) rx = rx[0][0][1] assert tx == rx = srcan check init time basecls sock1 = CANSocket(channel="vcan0", basecls=Raw) sock2 = CANSocket(channel="vcan0") sock2.send(tx) sock2.close() rx = None rx = sock1.sr(tx, timeout=1, verbose=False) rx = rx[0][0][1] assert Raw(bytes(tx)) == rx sock1.close() = sr can check rx and tx assert tx.sent_time > 0 and rx.time > 0 = sniff with filtermask 0x7ff sock1 = CANSocket(channel='vcan0', can_filters=[{'can_id': 0x200, 'can_mask': 0x7ff}]) sock2 = CANSocket(channel="vcan0") sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x300, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x300, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x100, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.close() packets = sock1.sniff(timeout=0.1, verbose=False, count=3) assert len(packets) == 3 sock1.close() = sniff with filtermask 0x700 sock1 = CANSocket(channel='vcan0', can_filters=[{'can_id': 0x200, 'can_mask': 0x700}]) sock2 = CANSocket(channel="vcan0") sock2.send(CAN(identifier=0x212, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x300, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x2ff, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x1ff, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x2aa, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.close() packets = sock1.sniff(timeout=0.1, verbose=False, count=4) assert len(packets) == 4 sock1.close() = sniff with filtermask 0x0ff sock1 = CANSocket(channel='vcan0', can_filters=[{'can_id': 0x200, 'can_mask': 0x0ff}]) sock2 = CANSocket(channel="vcan0") sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x301, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x300, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x1ff, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x700, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x100, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.close() packets = sock1.sniff(timeout=0.1, verbose=False, count=4) assert len(packets) == 4 sock1.close() = sniff with multiple filters sock1 = 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}]) sock2 = CANSocket(channel="vcan0") sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x300, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x400, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x500, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x600, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x700, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x7ff, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.close() packets = sock1.sniff(timeout=0.1, verbose=False, count=4) assert len(packets) == 4 sock1.close() = sniff with filtermask 0x7ff and inverse filter sock1 = CANSocket(channel='vcan0', can_filters=[{'can_id': 0x200 | CAN_INV_FILTER, 'can_mask': 0x7ff}]) sock2 = CANSocket(channel="vcan0") sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x300, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x100, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(identifier=0x200, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.close() packets = sock1.sniff(timeout=0.1, verbose=False, count=2) assert len(packets) == 2 sock1.close() = sniff with filtermask 0x1FFFFFFF sock1 = CANSocket(channel='vcan0', can_filters=[{'can_id': 0x10000000, 'can_mask': 0x1fffffff}]) sock2 = CANSocket(channel="vcan0") sock2.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10020000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10030000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10040000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.close() packets = sock1.sniff(timeout=0.1, verbose=False, count=2) assert len(packets) == 2 sock1.close() = sniff with filtermask 0x1FFFFFFF and inverse filter sock1 = CANSocket(channel='vcan0', can_filters=[{'can_id': 0x10000000 | CAN_INV_FILTER, 'can_mask': 0x1fffffff}]) sock2 = CANSocket(channel="vcan0") sock2.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10020000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10030000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10040000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock2.close() packets = sock1.sniff(timeout=0.1, verbose=False, count=4) assert len(packets) == 4 sock1.close() = CAN Socket sr1 with receive own messages sock1 = CANSocket(channel="vcan0", receive_own_messages=True) tx = CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08') rx = None rx = sock1.sr1(tx, verbose=False, timeout=3) assert tx == rx assert tx.sent_time < rx.time and tx == rx and rx.time > 0 sock1.close() = sr can sock1 = CANSocket(channel="vcan0", receive_own_messages=True) tx = CAN(identifier=0x7ff,length=8,data=b'\x01\x02\x03\x04\x05\x06\x07\x08') rx = None rx = sock1.sr(tx, timeout=0.1, verbose=False) assert tx == rx[0][0][1] + bridge and sniff tests = bridge and sniff setup vcan1 package forwarding bashCommand = "/bin/bash -c 'sudo ip link add name vcan1 type vcan; sudo ip link set dev vcan1 up'" assert 0 == os.system(bashCommand) sock0 = CANSocket(channel='vcan0') sock1 = CANSocket(channel='vcan1') bridgeStarted = threading.Event() def bridge(): global bridgeStarted bSock0 = CANSocket(channel="vcan0") bSock1 = CANSocket(channel='vcan1') def pnr(pkt): return pkt bridgeStarted.set() bridge_and_sniff(if1=bSock0, if2=bSock1, xfrm12=pnr, xfrm21=pnr, timeout=0.2, verbose=False, count=6) bSock0.close() bSock1.close() threadBridge = threading.Thread(target=bridge) threadBridge.start() bridgeStarted.wait(timeout=5) sock0.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10020000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10030000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10040000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) packetsVCan1 = sock1.sniff(timeout=0.1, verbose=False, count=6) assert len(packetsVCan1) == 6 threadBridge.join(timeout=5) assert not threadBridge.is_alive() sock1.close() sock0.close() = bridge and sniff setup vcan0 package forwarding sock0 = CANSocket(channel='vcan0') sock1 = CANSocket(channel='vcan1') bridgeStarted = threading.Event() def bridge(): global bridgeStarted bSock0 = CANSocket(channel="vcan0") bSock1 = CANSocket(channel='vcan1') def pnr(pkt): return pkt bridgeStarted.set() bridge_and_sniff(if1=bSock0, if2=bSock1, xfrm12=pnr, xfrm21=pnr, timeout=0.2, verbose=False, count=4) bSock0.close() bSock1.close() threadBridge = threading.Thread(target=bridge) threadBridge.start() bridgeStarted.wait(timeout=5) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x80, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) packetsVCan0 = sock0.sniff(timeout=0.1, verbose=False, count=4) assert len(packetsVCan0) == 4 sock0.close() sock1.close() threadBridge.join(timeout=5) assert not threadBridge.is_alive() =bridge and sniff setup vcan0 vcan1 package forwarding both directions sock0 = CANSocket(channel='vcan0') sock1 = CANSocket(channel='vcan1') bridgeStarted = threading.Event() def bridge(): global bridgeStarted bSock0 = CANSocket(channel="vcan0") bSock1 = CANSocket(channel='vcan1') def pnr(pkt): return pkt bridgeStarted.set() bridge_and_sniff(if1=bSock0, if2=bSock1, xfrm12=pnr, xfrm21=pnr, timeout=0.2, verbose=False, count=10) bSock0.close() bSock1.close() threadBridge = threading.Thread(target=bridge) threadBridge.start() bridgeStarted.wait(timeout=5) sock0.send(CAN(flags='extended', identifier=0x25, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x20, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x25, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x25, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x20, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x30, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock1.send(CAN(flags='extended', identifier=0x40, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x40, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x80, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x40, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) packetsVCan0 = sock0.sniff(timeout=0.1, count=4, verbose=False) packetsVCan1 = sock1.sniff(timeout=0.1, verbose=False, count=6) assert len(packetsVCan0) == 4 assert len(packetsVCan1) == 6 sock0.close() sock1.close() threadBridge.join(timeout=5) assert not threadBridge.is_alive() =bridge and sniff setup vcan1 package change sock0 = CANSocket(channel='vcan0') sock1 = CANSocket(channel='vcan1', can_filters=[{'can_id': 0x10010000, 'can_mask': 0x1fffffff}]) bridgeStarted = threading.Event() def bridgeWithPackageChangeVCan0ToVCan1(): global bridgeStarted bSock0 = CANSocket(channel="vcan0") bSock1 = CANSocket(channel="vcan1") def pnr(pkt): pkt.data = b'\x08\x07\x06\x05\x04\x03\x02\x01' pkt.identifier = 0x10010000 return pkt bridgeStarted.set() bridge_and_sniff(if1=bSock0, if2=bSock1, xfrm12=pnr, timeout=0.2, verbose=False, count=6) bSock0.close() bSock1.close() threadBridge = threading.Thread(target=bridgeWithPackageChangeVCan0ToVCan1) threadBridge.start() bridgeStarted.wait(timeout=5) sock0.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10020000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10030000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10040000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) packetsVCan1 = sock1.sniff(timeout=0.1, verbose=False, count=6) assert len(packetsVCan1) == 6 sock0.close() sock1.close() threadBridge.join(timeout=5) assert not threadBridge.is_alive() =bridge and sniff setup vcan0 package change sock0 = CANSocket(channel='vcan0', can_filters=[{'can_id': 0x10010000, 'can_mask': 0x1fffffff}]) sock1 = CANSocket(channel='vcan1') bridgeStarted = threading.Event() def bridgeWithPackageChangeVCan1ToVCan0(): global bridgeStarted bSock0 = CANSocket(channel="vcan0") bSock1 = CANSocket(channel="vcan1") def pnr(pkt): pkt.data = b'\x08\x07\x06\x05\x04\x03\x02\x01' pkt.identifier = 0x10010000 return pkt bridgeStarted.set() bridge_and_sniff(if1=bSock0, if2=bSock1, xfrm21=pnr, timeout=0.2, verbose=False, count=4) bSock0.close() bSock1.close() threadBridge = threading.Thread(target=bridgeWithPackageChangeVCan1ToVCan0) threadBridge.start() bridgeStarted.wait(timeout=5) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10050000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) packetsVCan0 = sock0.sniff(timeout=0.1, verbose=False, count=4) assert len(packetsVCan0) == 4 sock0.close() sock1.close() threadBridge.join(timeout=5) =bridge and sniff setup vcan0 and vcan1 package change in both directions sock0 = CANSocket(channel='vcan0', can_filters=[{'can_id': 0x10010000, 'can_mask': 0x1fffffff}]) sock1 = CANSocket(channel='vcan1', can_filters=[{'can_id': 0x10010000, 'can_mask': 0x1fffffff}]) bridgeStarted = threading.Event() def bridgeWithPackageChangeBothDirections(): global bridgeStarted bSock0 = CANSocket(channel="vcan0") bSock1 = CANSocket(channel="vcan1") def pnr(pkt): pkt.data = b'\x08\x07\x06\x05\x04\x03\x02\x01' pkt.identifier = 0x10010000 return pkt bridgeStarted.set() bridge_and_sniff(if1=bSock0, if2=bSock1, xfrm12=pnr, xfrm21=pnr, timeout=0.2, verbose=False, count=10) bSock0.close() bSock1.close() threadBridge = threading.Thread(target=bridgeWithPackageChangeBothDirections) threadBridge.start() bridgeStarted.wait(timeout=5) sock0.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10020000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10030000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10040000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10050000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) packetsVCan0 = sock0.sniff(timeout=0.1, verbose=False, count=4) packetsVCan1 = sock1.sniff(timeout=0.1, verbose=False, count=6) assert len(packetsVCan0) == 4 assert len(packetsVCan1) == 6 sock0.close() sock1.close() threadBridge.join(timeout=5) =bridge and sniff setup vcan0 package remove sock0 = CANSocket(channel='vcan0') sock1 = CANSocket(channel='vcan1') bridgeStarted = threading.Event() def bridgeWithRemovePackageFromVCan0ToVCan1(): global bridgeStarted bSock0 = CANSocket(channel="vcan0") bSock1 = CANSocket(channel="vcan1") def pnr(pkt): if(pkt.identifier == 0x10020000): pkt = None else: pkt = pkt return pkt bridgeStarted.set() bridge_and_sniff(if1=bSock0, if2=bSock1, xfrm12=pnr, timeout=0.2, verbose=False, count=6) bSock0.close() bSock1.close() threadBridge = threading.Thread(target=bridgeWithRemovePackageFromVCan0ToVCan1) threadBridge.start() bridgeStarted.wait(timeout=5) sock0.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10020000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10030000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10040000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) packetsVCan1 = sock1.sniff(timeout=0.1, verbose=False, count=5) assert len(packetsVCan1) == 5 sock0.close() sock1.close() threadBridge.join(timeout=5) =bridge and sniff setup vcan1 package remove sock0 = CANSocket(channel='vcan0') sock1 = CANSocket(channel='vcan1') bridgeStarted = threading.Event() def bridgeWithRemovePackageFromVCan1ToVCan0(): global bridgeStarted bSock0 = CANSocket(channel="vcan0") bSock1 = CANSocket(channel="vcan1") def pnr(pkt): if(pkt.identifier == 0x10050000): pkt = None else: pkt = pkt return pkt bridgeStarted.set() bridge_and_sniff(if1=bSock0, if2=bSock1, xfrm21=pnr, timeout=0.2, verbose=False, count=4) bSock0.close() bSock1.close() threadBridge = threading.Thread(target=bridgeWithRemovePackageFromVCan1ToVCan0) threadBridge.start() bridgeStarted.wait(timeout=5) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10050000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) packetsVCan0 = sock0.sniff(timeout=0.1, verbose=False, count=3) assert len(packetsVCan0) == 3 sock0.close() sock1.close() threadBridge.join(timeout=5) =bridge and sniff setup vcan0 and vcan1 package remove both directions sock0 = CANSocket(channel="vcan0") sock1 = CANSocket(channel="vcan1") bridgeStarted = threading.Event() def bridgeWithRemovePackageInBothDirections(): global bridgeStarted bSock0 = CANSocket(channel="vcan0") bSock1 = CANSocket(channel="vcan1") def pnrA(pkt): if(pkt.identifier == 0x10020000): pkt = None else: pkt = pkt return pkt def pnrB(pkt): if (pkt.identifier == 0x10050000): pkt = None else: pkt = pkt return pkt bridgeStarted.set() bridge_and_sniff(if1=bSock0, if2=bSock1, xfrm12=pnrA, xfrm21=pnrB, timeout=0.2, verbose=False, count=10) bSock0.close() bSock1.close() threadBridge = threading.Thread(target=bridgeWithRemovePackageInBothDirections) threadBridge.start() bridgeStarted.wait(timeout=5) sock0.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10020000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10030000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10040000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock0.send(CAN(flags='extended', identifier=0x10000000, length=8, data=b'\x01\x02\x03\x04\x05\x06\x07\x08')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10050000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) sock1.send(CAN(flags='extended', identifier=0x10010000, length=8, data=b'\x01\x02\x03\x04\x05\x04\x05\x06')) packetsVCan0 = sock0.sniff(timeout=0.1, verbose=False, count=3) packetsVCan1 = sock1.sniff(timeout=0.1, verbose=False, count=5) assert len(packetsVCan0) == 3 assert len(packetsVCan1) == 5 sock0.close() sock1.close() threadBridge.join(timeout=5) = Delete vcan interfaces if 0 != call(["sudo", "ip", "link", "delete", "vcan0"]): raise Exception("vcan0 could not be deleted") if 0 != call(["sudo", "ip", "link", "delete", "vcan1"]): raise Exception("vcan1 could not be deleted")