1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
|
% Regression tests for obdscanner
~ vcan_socket needs_root linux not_pypy automotive_comm scanner
+ Configuration
~ conf
= Imports
with open(scapy_path("test/contrib/automotive/interface_mockup.py")) as f:
exec(f.read())
load_contrib("automotive.ecu", globals_dict=globals())
+ Usage tests
= Test wrong usage
print(sys.executable)
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
returncode = result.wait()
std_out, std_err = result.communicate()
assert returncode != 0
expected_output = plain_str(b'usage:')
assert expected_output in plain_str(std_err)
= Test show help
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py", "--help"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
assert result.wait() == 0
std_out, std_err = result.communicate()
expected_output = plain_str(b'Scan for all possible obd service classes and their subfunctions.')
assert expected_output in plain_str(std_out)
= Test wrong socket for Python2 or Windows
if six.PY2:
version = subprocess.Popen(["python2", "--version"], stdout=subprocess.PIPE)
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py", "-c", "vcan0", "-s", "0x600", "-d", "0x601", "-t", "0.001"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
assert result.wait() == 1
expected_output = plain_str(b'Please provide all required arguments.')
std_out, std_err = result.communicate()
assert expected_output in plain_str(std_err)
= Test Python2 call
if six.PY2:
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py", "-i", "socketcan", "-c", "vcan0", "-s", "0x600", "-d", "0x601", "-v", "-t", "0.001"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
returncode = result.wait()
std_out, std_err = result.communicate()
assert returncode == 0
expected_output = plain_str(b'Starting OBD-Scan...')
assert expected_output in plain_str(std_out)
= Test Python2 call with python-can args
if six.PY2:
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py", "-i", "socketcan", "-c", "vcan0", "-s", "0x600", "-d", "0x601", "-v", "-a", "bitrate=250000", "-t", "0.001"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
returncode = result.wait()
std_out, std_err = result.communicate()
assert returncode == 0
expected_output = plain_str(b'Starting OBD-Scan...')
assert expected_output in plain_str(std_out)
+ Scan tests
= Load contribution layer
from scapy.contrib.automotive.obd.obd import *
+ Simulate scanner
= Test DTC scan
drain_bus(iface0)
s3 = OBD()/OBD_S03_PR(dtcs=[OBD_DTC()])
example_responses = [EcuResponse(responses=s3)]
with new_can_socket0() as isocan, ISOTPSocket(isocan, 0x7e8, 0x7e0, basecls=OBD, padding=True) as ecu, \
new_can_socket0() as isocan2, ISOTPSocket(isocan2, 0x7e0, 0x7e8, basecls=OBD, padding=True) as tester:
conf.verb = -1
answering_machine = EcuAnsweringMachine(supported_responses=example_responses, main_socket=ecu, basecls=OBD, verbose=False)
sim = threading.Thread(target=answering_machine, kwargs={'verbose': False, 'timeout': 15, 'stop_filter': lambda p: bytes(p) == b"\x01\xff\xff\xff\xff"})
sim.start()
try:
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
std_out1, std_err1 = result.communicate()
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
std_out2, std_err2 = result.communicate()
except Exception as e:
print(e)
finally:
tester.send(b"\x01\xff\xff\xff\xff")
sim.join(timeout=10)
expected_output = b"1 requests were sent, 1 answered"
assert bytes_encode(expected_output) in bytes_encode(std_out1) or bytes_encode(expected_output) in bytes_encode(std_out2)
= Test supported PIDs scan
drain_bus(iface0)
s1_pid00 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID00(supported_pids="PID03+PID0B+PID0F")])
s6_mid00 = OBD()/OBD_S06_PR(data_records=[OBD_S06_PR_Record()/OBD_MID00(supported_mids="")])
s8_tid00 = OBD()/OBD_S08_PR(data_records=[OBD_S08_PR_Record()/OBD_TID00(supported_tids="")])
s9_iid00 = OBD()/OBD_S09_PR(data_records=[OBD_S09_PR_Record()/OBD_IID00(supported_iids="")])
s1_pid03 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID03(fuel_system1=0, fuel_system2=2)])
s1_pid0B = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID0B(data=100)])
s1_pid0F = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID0F(data=50)])
# Create answers for 'supported PIDs scan'
example_responses = \
[EcuResponse(responses=s3),
EcuResponse(responses=s1_pid00),
EcuResponse(responses=s6_mid00),
EcuResponse(responses=s8_tid00),
EcuResponse(responses=s9_iid00),
EcuResponse(responses=s1_pid03),
EcuResponse(responses=s1_pid0B),
EcuResponse(responses=s1_pid0F)]
with new_can_socket0() as isocan, ISOTPSocket(isocan, 0x7e8, 0x7e0, basecls=OBD, padding=True) as ecu, \
new_can_socket0() as isocan2, ISOTPSocket(isocan2, 0x7e0, 0x7e8, basecls=OBD, padding=True) as tester:
answering_machine = EcuAnsweringMachine(supported_responses=example_responses, main_socket=ecu, basecls=OBD, verbose=False)
sim = threading.Thread(target=answering_machine, kwargs={'verbose': False, 'timeout': 100, 'stop_filter': lambda p: bytes(p) == b"\x01\xff\xff\xff\xff"})
sim.start()
try:
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
std_out1, std_err1 = result.communicate()
print(std_out2, std_err2)
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
std_out2, std_err2 = result.communicate()
print(std_out2, std_err2)
except Exception as e:
print(e)
finally:
tester.send(b"\x01\xff\xff\xff\xff")
sim.join(timeout=10)
expected_output = ["supported_pids=PID0F+PID0B+PID03", "fuel_system1=OpenLoopInsufficientEngineTemperature fuel_system2=ClosedLoop", "data=100 kPa", "data=50.0 deg. C"]
for out in expected_output:
assert bytes_encode(out) in bytes_encode(std_out1) or bytes_encode(out) in bytes_encode(std_out2)
= Test only Service 01 PIDs scan
drain_bus(iface0)
s1_pid00 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID00(supported_pids="PID03+PID0B+PID0F")])
s6_mid00 = OBD()/OBD_S06_PR(data_records=[OBD_S06_PR_Record()/OBD_MID00(supported_mids="")])
s8_tid00 = OBD()/OBD_S08_PR(data_records=[OBD_S08_PR_Record()/OBD_TID00(supported_tids="")])
s9_iid00 = OBD()/OBD_S09_PR(data_records=[OBD_S09_PR_Record()/OBD_IID00(supported_iids="")])
s1_pid03 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID03(fuel_system1=0, fuel_system2=2)])
s1_pid0B = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID0B(data=100)])
s1_pid0F = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID0F(data=50)])
# Create answers for 'supported PIDs scan'
example_responses = \
[EcuResponse(responses=s3),
EcuResponse(responses=s1_pid00),
EcuResponse(responses=s6_mid00),
EcuResponse(responses=s8_tid00),
EcuResponse(responses=s9_iid00),
EcuResponse(responses=s1_pid03),
EcuResponse(responses=s1_pid0B),
EcuResponse(responses=s1_pid0F)]
with new_can_socket0() as isocan, ISOTPSocket(isocan, 0x7e8, 0x7e0, basecls=OBD, padding=True) as ecu, \
new_can_socket0() as isocan2, ISOTPSocket(isocan2, 0x7e0, 0x7e8, basecls=OBD, padding=True) as tester:
answering_machine = EcuAnsweringMachine(supported_responses=example_responses, main_socket=ecu, basecls=OBD, verbose=False)
sim = threading.Thread(target=answering_machine, kwargs={'verbose': False, 'timeout': 100, 'stop_filter': lambda p: bytes(p) == b"\x01\xff\xff\xff\xff"})
sim.start()
try:
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30", "-1"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
std_out1, std_err1 = result.communicate()
print(std_out1, std_err1)
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30", "-1"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
std_out2, std_err2 = result.communicate()
print(std_out2, std_err2)
except Exception as e:
print(e)
finally:
tester.send(b"\x01\xff\xff\xff\xff")
sim.join(timeout=10)
expected_output = ["supported_pids=PID0F+PID0B+PID03", "fuel_system1=OpenLoopInsufficientEngineTemperature fuel_system2=ClosedLoop", "data=100 kPa", "data=50.0 deg. C"]
for out in expected_output:
assert bytes_encode(out) in bytes_encode(std_out1) or bytes_encode(out) in bytes_encode(std_out2)
= Test full scan
drain_bus(iface0)
# Add unsupported PID
s1_pid01 = OBD()/OBD_S01_PR(data_records=[OBD_S01_PR_Record()/OBD_PID01()])
example_responses.append(EcuResponse(responses=s1_pid01))
with new_can_socket0() as isocan, ISOTPSocket(isocan, 0x7e8, 0x7e0, basecls=OBD, padding=True) as ecu, \
new_can_socket0() as isocan2, ISOTPSocket(isocan2, 0x7e0, 0x7e8, basecls=OBD, padding=True) as tester:
answering_machine = EcuAnsweringMachine(supported_responses=example_responses, main_socket=ecu, basecls=OBD, verbose=False)
sim = threading.Thread(target=answering_machine, kwargs={'verbose': False, 'timeout': 100, 'stop_filter': lambda p: bytes(p) == b"\x01\xff\xff\xff\xff"})
sim.start()
try:
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30", "-f", "-1", "-3"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
std_out1, std_err1 = result.communicate()
result = subprocess.Popen([sys.executable, "scapy/tools/automotive/obdscanner.py"] + can_socket_string_list + ["-s", "0x7e0", "-d", "0x7e8", "-t", "0.30", "-f", "-1", "-3"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
std_out2, std_err2 = result.communicate()
except Exception as e:
print(e)
finally:
tester.send(b"\x01\xff\xff\xff\xff")
sim.join(timeout=10)
expected_output = ["256 requests were sent", "1 requests were sent, 1 answered"]
for out in expected_output:
assert bytes_encode(out) in bytes_encode(std_out1) or bytes_encode(out) in bytes_encode(std_out2)
+ Cleanup
= Delete vcan interfaces
assert cleanup_interfaces()
|
