mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-21 00:03:45 +08:00
Merge commit 'c251b312d87d26b5ed347b267f6f1570793f9b91' as 'panda'
This commit is contained in:
Vehicle Researcher
Executable
+54
@@ -0,0 +1,54 @@
|
||||
#!/usr/bin/env python
|
||||
from __future__ import print_function
|
||||
import binascii
|
||||
import csv
|
||||
import sys
|
||||
from panda import Panda
|
||||
|
||||
def can_logger():
|
||||
|
||||
try:
|
||||
print("Trying to connect to Panda over USB...")
|
||||
p = Panda()
|
||||
|
||||
except AssertionError:
|
||||
print("USB connection failed. Trying WiFi...")
|
||||
|
||||
try:
|
||||
p = Panda("WIFI")
|
||||
except:
|
||||
print("WiFi connection timed out. Please make sure your Panda is connected and try again.")
|
||||
sys.exit(0)
|
||||
|
||||
try:
|
||||
outputfile = open('output.csv', 'wb')
|
||||
csvwriter = csv.writer(outputfile)
|
||||
#Write Header
|
||||
csvwriter.writerow(['Bus', 'MessageID', 'Message', 'MessageLength'])
|
||||
print("Writing csv file output.csv. Press Ctrl-C to exit...\n")
|
||||
|
||||
bus0_msg_cnt = 0
|
||||
bus1_msg_cnt = 0
|
||||
bus2_msg_cnt = 0
|
||||
|
||||
while True:
|
||||
can_recv = p.can_recv()
|
||||
|
||||
for address, _, dat, src in can_recv:
|
||||
csvwriter.writerow([str(src), str(hex(address)), "0x" + binascii.hexlify(dat), len(dat)])
|
||||
|
||||
if src == 0:
|
||||
bus0_msg_cnt += 1
|
||||
elif src == 1:
|
||||
bus1_msg_cnt += 1
|
||||
elif src == 2:
|
||||
bus2_msg_cnt += 1
|
||||
|
||||
print("Message Counts... Bus 0: " + str(bus0_msg_cnt) + " Bus 1: " + str(bus1_msg_cnt) + " Bus 2: " + str(bus2_msg_cnt), end='\r')
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\nNow exiting. Final message Counts... Bus 0: " + str(bus0_msg_cnt) + " Bus 1: " + str(bus1_msg_cnt) + " Bus 2: " + str(bus2_msg_cnt))
|
||||
outputfile.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
can_logger()
|
||||
@@ -0,0 +1,103 @@
|
||||
# How to use can_unique.py to reverse engineer a single bit field
|
||||
|
||||
Let's say our goal is to find the CAN message indicating that the driver's door is either open or closed.
|
||||
The following process is great for simple single-bit messages.
|
||||
However for frequently changing values, such as RPM or speed, Cabana's graphical plots are probably better to use.
|
||||
|
||||
|
||||
First record a few minutes of background CAN messages with all the doors closed and save it in background.csv:
|
||||
```
|
||||
./can_logger.py
|
||||
mv output.csv background.csv
|
||||
```
|
||||
Then run can_logger.py for a few seconds while performing the action you're interested, such as opening and then closing the
|
||||
front-left door and save it as door-fl-1.csv
|
||||
Repeat the process and save it as door-f1-2.csv to have an easy way to confirm any suspicions.
|
||||
|
||||
Now we'll use can_unique.py to look for unique bits:
|
||||
```
|
||||
$ ./can_unique.py door-fl-1.csv background*
|
||||
id 820 new one at byte 2 bitmask 2
|
||||
id 520 new one at byte 3 bitmask 7
|
||||
id 520 new zero at byte 3 bitmask 8
|
||||
id 520 new one at byte 5 bitmask 6
|
||||
id 520 new zero at byte 5 bitmask 9
|
||||
id 559 new zero at byte 6 bitmask 4
|
||||
id 804 new one at byte 5 bitmask 2
|
||||
id 804 new zero at byte 5 bitmask 1
|
||||
|
||||
$ ./can_unique.py door-fl-2.csv background*
|
||||
id 672 new one at byte 3 bitmask 3
|
||||
id 820 new one at byte 2 bitmask 2
|
||||
id 520 new one at byte 3 bitmask 7
|
||||
id 520 new zero at byte 3 bitmask 8
|
||||
id 520 new one at byte 5 bitmask 6
|
||||
id 520 new zero at byte 5 bitmask 9
|
||||
id 559 new zero at byte 6 bitmask 4
|
||||
```
|
||||
|
||||
One of these bits hopefully indicates that the driver's door is open.
|
||||
Let's go through each message ID to figure out which one is correct.
|
||||
We expect any correct bits to have changed in both runs.
|
||||
We can rule out 804 because it only occurred in the first run.
|
||||
We can rule out 672 because it only occurred in the second run.
|
||||
That leaves us with these message IDs: 820, 520, 559. Let's take a closer look at each one.
|
||||
|
||||
```
|
||||
$ fgrep ,559, door-fl-1.csv |head
|
||||
0,559,00ff0000000024f0
|
||||
0,559,00ff000000004464
|
||||
0,559,00ff0000000054a9
|
||||
0,559,00ff0000000064e3
|
||||
0,559,00ff00000000742e
|
||||
0,559,00ff000000008451
|
||||
0,559,00ff00000000949c
|
||||
0,559,00ff00000000a4d6
|
||||
0,559,00ff00000000b41b
|
||||
0,559,00ff00000000c442
|
||||
```
|
||||
Message ID 559 looks like an incrementing value, so it's not what we're looking for.
|
||||
|
||||
```
|
||||
$ fgrep ,520, door-fl-2.csv
|
||||
0,520,26ff00f8a1890000
|
||||
0,520,26ff00f8a2890000
|
||||
0,520,26ff00f8a2890000
|
||||
0,520,26ff00f8a1890000
|
||||
0,520,26ff00f8a2890000
|
||||
0,520,26ff00f8a1890000
|
||||
0,520,26ff00f8a2890000
|
||||
0,520,26ff00f8a1890000
|
||||
0,520,26ff00f8a2890000
|
||||
0,520,26ff00f8a1890000
|
||||
0,520,26ff00f8a2890000
|
||||
0,520,26ff00f8a1890000
|
||||
```
|
||||
Message ID 520 oscillates between two values. However I only opened and closed the door once, so this is probably not it.
|
||||
|
||||
```
|
||||
$ fgrep ,820, door-fl-1.csv
|
||||
0,820,44000100a500c802
|
||||
0,820,44000100a500c803
|
||||
0,820,44000300a500c803
|
||||
0,820,44000300a500c802
|
||||
0,820,44000300a500c802
|
||||
0,820,44000300a500c802
|
||||
0,820,44000100a500c802
|
||||
0,820,44000100a500c802
|
||||
0,820,44000100a500c802
|
||||
```
|
||||
Message ID 820 looks promising! It starts off at 44000100a500c802 when the door is closed.
|
||||
When the door is open it goes to 44000300a500c802.
|
||||
Then when the door is closed again, it goes back to 44000100a500c802.
|
||||
Let's confirm by looking at the data from our other run:
|
||||
```
|
||||
$ fgrep ,820, door-fl-2.csv
|
||||
0,820,44000100a500c802
|
||||
0,820,44000300a500c802
|
||||
0,820,44000100a500c802
|
||||
```
|
||||
Perfect! We now know that message id 820 at byte 2 bitmask 2 is set if the driver's door is open.
|
||||
If we repeat the process with the front passenger's door,
|
||||
then we'll find that message id 820 at byte 2 bitmask 4 is set if the front-right door is open.
|
||||
This confirms our finding because it's common for similar signals to be near each other.
|
||||
Executable
+91
@@ -0,0 +1,91 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
# Given an interesting CSV file of CAN messages and a list of background CAN
|
||||
# messages, print which bits in the interesting file have never appeared
|
||||
# in the background files.
|
||||
|
||||
# Expects the CSV file to be in the format from can_logger.py
|
||||
# Bus,MessageID,Message,MessageLength
|
||||
# 0,0x292,0x040000001068,6
|
||||
|
||||
# The old can_logger.py format is also supported:
|
||||
# Bus,MessageID,Message
|
||||
# 0,344,c000c00000000000
|
||||
|
||||
|
||||
import binascii
|
||||
import csv
|
||||
import sys
|
||||
from panda import Panda
|
||||
|
||||
class Message():
|
||||
"""Details about a specific message ID."""
|
||||
def __init__(self, message_id):
|
||||
self.message_id = message_id
|
||||
self.data = {} # keyed by hex string encoded message data
|
||||
self.ones = [0] * 8 # bit set if 1 is seen
|
||||
self.zeros = [0] * 8 # bit set if 0 has been seen
|
||||
|
||||
def printBitDiff(self, other):
|
||||
"""Prints bits that are set or cleared compared to other background."""
|
||||
for i in xrange(len(self.ones)):
|
||||
new_ones = ((~other.ones[i]) & 0xff) & self.ones[i]
|
||||
if new_ones:
|
||||
print 'id %s new one at byte %d bitmask %d' % (
|
||||
self.message_id, i, new_ones)
|
||||
new_zeros = ((~other.zeros[i]) & 0xff) & self.zeros[i]
|
||||
if new_zeros:
|
||||
print 'id %s new zero at byte %d bitmask %d' % (
|
||||
self.message_id, i, new_zeros)
|
||||
|
||||
|
||||
class Info():
|
||||
"""A collection of Messages."""
|
||||
|
||||
def __init__(self):
|
||||
self.messages = {} # keyed by MessageID
|
||||
|
||||
def load(self, filename):
|
||||
"""Given a CSV file, adds information about message IDs and their values."""
|
||||
with open(filename, 'rb') as input:
|
||||
reader = csv.reader(input)
|
||||
next(reader, None) # skip the CSV header
|
||||
for row in reader:
|
||||
if row[1].startswith('0x'):
|
||||
message_id = row[1][2:] # remove leading '0x'
|
||||
else:
|
||||
message_id = hex(int(row[1]))[2:] # old message IDs are in decimal
|
||||
if row[1].startswith('0x'):
|
||||
data = row[2][2:] # remove leading '0x'
|
||||
else:
|
||||
data = row[2]
|
||||
if message_id not in self.messages:
|
||||
self.messages[message_id] = Message(message_id)
|
||||
message = self.messages[message_id]
|
||||
if data not in self.messages[message_id].data:
|
||||
message.data[data] = True
|
||||
bytes = bytearray.fromhex(data)
|
||||
for i in xrange(len(bytes)):
|
||||
message.ones[i] = message.ones[i] | int(bytes[i])
|
||||
# Inverts the data and masks it to a byte to get the zeros as ones.
|
||||
message.zeros[i] = message.zeros[i] | ( (~int(bytes[i])) & 0xff)
|
||||
|
||||
def PrintUnique(interesting_file, background_files):
|
||||
background = Info()
|
||||
for background_file in background_files:
|
||||
background.load(background_file)
|
||||
interesting = Info()
|
||||
interesting.load(interesting_file)
|
||||
for message_id in interesting.messages:
|
||||
if message_id not in background.messages:
|
||||
print 'New message_id: %s' % message_id
|
||||
else:
|
||||
interesting.messages[message_id].printBitDiff(
|
||||
background.messages[message_id])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 3:
|
||||
print 'Usage:\n%s interesting.csv background*.csv' % sys.argv[0]
|
||||
sys.exit(0)
|
||||
PrintUnique(sys.argv[1], sys.argv[2:])
|
||||
@@ -0,0 +1,77 @@
|
||||
DEBUG = False
|
||||
|
||||
def msg(x):
|
||||
if DEBUG:
|
||||
print "S:",x.encode("hex")
|
||||
if len(x) <= 7:
|
||||
ret = chr(len(x)) + x
|
||||
else:
|
||||
assert False
|
||||
return ret.ljust(8, "\x00")
|
||||
|
||||
def isotp_send(panda, x, addr, bus=0):
|
||||
if len(x) <= 7:
|
||||
panda.can_send(addr, msg(x), bus)
|
||||
else:
|
||||
ss = chr(0x10 + (len(x)>>8)) + chr(len(x)&0xFF) + x[0:6]
|
||||
x = x[6:]
|
||||
idx = 1
|
||||
sends = []
|
||||
while len(x) > 0:
|
||||
sends.append(((chr(0x20 + (idx&0xF)) + x[0:7]).ljust(8, "\x00")))
|
||||
x = x[7:]
|
||||
idx += 1
|
||||
|
||||
# actually send
|
||||
panda.can_send(addr, ss, bus)
|
||||
rr = recv(panda, 1, addr+8, bus)[0]
|
||||
panda.can_send_many([(addr, None, s, 0) for s in sends])
|
||||
|
||||
kmsgs = []
|
||||
def recv(panda, cnt, addr, nbus):
|
||||
global kmsgs
|
||||
ret = []
|
||||
|
||||
while len(ret) < cnt:
|
||||
kmsgs += panda.can_recv()
|
||||
nmsgs = []
|
||||
for ids, ts, dat, bus in kmsgs:
|
||||
if ids == addr and bus == nbus and len(ret) < cnt:
|
||||
ret.append(dat)
|
||||
else:
|
||||
pass
|
||||
kmsgs = nmsgs
|
||||
return map(str, ret)
|
||||
|
||||
def isotp_recv(panda, addr, bus=0):
|
||||
msg = recv(panda, 1, addr, bus)[0]
|
||||
|
||||
if ord(msg[0])&0xf0 == 0x10:
|
||||
# first
|
||||
tlen = ((ord(msg[0]) & 0xf) << 8) | ord(msg[1])
|
||||
dat = msg[2:]
|
||||
|
||||
# 0 block size?
|
||||
CONTINUE = "\x30" + "\x00"*7
|
||||
|
||||
panda.can_send(addr-8, CONTINUE, bus)
|
||||
|
||||
idx = 1
|
||||
for mm in recv(panda, (tlen-len(dat) + 7)/8, addr, bus):
|
||||
assert ord(mm[0]) == (0x20 | idx)
|
||||
dat += mm[1:]
|
||||
idx += 1
|
||||
elif ord(msg[0])&0xf0 == 0x00:
|
||||
# single
|
||||
tlen = ord(msg[0]) & 0xf
|
||||
dat = msg[1:]
|
||||
else:
|
||||
assert False
|
||||
|
||||
dat = dat[0:tlen]
|
||||
|
||||
if DEBUG:
|
||||
print "R:",dat.encode("hex")
|
||||
|
||||
return dat
|
||||
|
||||
Executable
+59
@@ -0,0 +1,59 @@
|
||||
#!/usr/bin/env python
|
||||
import time
|
||||
import struct
|
||||
from panda import Panda
|
||||
from hexdump import hexdump
|
||||
from isotp import isotp_send, isotp_recv
|
||||
|
||||
# 0x7e0 = Toyota
|
||||
# 0x18DB33F1 for Honda?
|
||||
|
||||
def get_current_data_for_pid(pid):
|
||||
# 01 xx = Show current data
|
||||
isotp_send(panda, "\x01"+chr(pid), 0x7e0)
|
||||
return isotp_recv(panda, 0x7e8)
|
||||
|
||||
def get_supported_pids():
|
||||
ret = []
|
||||
pid = 0
|
||||
while 1:
|
||||
supported = struct.unpack(">I", get_current_data_for_pid(pid)[2:])[0]
|
||||
for i in range(1+pid, 0x21+pid):
|
||||
if supported & 0x80000000:
|
||||
ret.append(i)
|
||||
supported <<= 1
|
||||
pid += 0x20
|
||||
if pid not in ret:
|
||||
break
|
||||
return ret
|
||||
|
||||
if __name__ == "__main__":
|
||||
panda = Panda()
|
||||
panda.set_safety_mode(Panda.SAFETY_ELM327)
|
||||
panda.can_clear(0)
|
||||
|
||||
# 09 02 = Get VIN
|
||||
isotp_send(panda, "\x09\x02", 0x7e0)
|
||||
ret = isotp_recv(panda, 0x7e8)
|
||||
hexdump(ret)
|
||||
print "VIN: %s" % ret[2:]
|
||||
|
||||
# 03 = get DTCS
|
||||
isotp_send(panda, "\x03", 0x7e0)
|
||||
dtcs = isotp_recv(panda, 0x7e8)
|
||||
print "DTCs:", dtcs[2:].encode("hex")
|
||||
|
||||
supported_pids = get_supported_pids()
|
||||
print "Supported PIDs:",supported_pids
|
||||
|
||||
while 1:
|
||||
speed = struct.unpack(">B", get_current_data_for_pid(13)[2:])[0] # kph
|
||||
rpm = struct.unpack(">H", get_current_data_for_pid(12)[2:])[0]/4.0 # revs
|
||||
throttle = struct.unpack(">B", get_current_data_for_pid(17)[2:])[0]/255.0 * 100 # percent
|
||||
temp = struct.unpack(">B", get_current_data_for_pid(5)[2:])[0] - 40 # degrees C
|
||||
load = struct.unpack(">B", get_current_data_for_pid(4)[2:])[0]/255.0 * 100 # percent
|
||||
print "%d KPH, %d RPM, %.1f%% Throttle, %d deg C, %.1f%% load" % (speed, rpm, throttle, temp, load)
|
||||
time.sleep(0.2)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
#!/usr/bin/env python
|
||||
import sys
|
||||
import binascii
|
||||
from panda import Panda
|
||||
|
||||
def tesla_tester():
|
||||
|
||||
try:
|
||||
print("Trying to connect to Panda over USB...")
|
||||
p = Panda()
|
||||
|
||||
except AssertionError:
|
||||
print("USB connection failed. Trying WiFi...")
|
||||
|
||||
try:
|
||||
p = Panda("WIFI")
|
||||
except:
|
||||
print("WiFi connection timed out. Please make sure your Panda is connected and try again.")
|
||||
sys.exit(0)
|
||||
|
||||
body_bus_speed = 125 # Tesla Body busses (B, BF) are 125kbps, rest are 500kbps
|
||||
body_bus_num = 1 # My TDC to OBD adapter has PT on bus0 BDY on bus1 and CH on bus2
|
||||
p.set_can_speed_kbps(body_bus_num, body_bus_speed)
|
||||
|
||||
# Now set the panda from its default of SAFETY_NOOUTPUT (read only) to SAFETY_ALLOUTPUT
|
||||
# Careful, as this will let us send any CAN messages we want (which could be very bad!)
|
||||
print("Setting Panda to output mode...")
|
||||
p.set_safety_mode(Panda.SAFETY_ALLOUTPUT)
|
||||
|
||||
# BDY 0x248 is the MCU_commands message, which includes folding mirrors, opening the trunk, frunk, setting the cars lock state and more. For our test, we will edit the 3rd byte, which is MCU_lockRequest. 0x01 will lock, 0x02 will unlock:
|
||||
print("Unlocking Tesla...")
|
||||
p.can_send(0x248, "\x00\x00\x02\x00\x00\x00\x00\x00", bus_num)
|
||||
|
||||
#Or, we can set the first byte, MCU_frontHoodCommand + MCU_liftgateSwitch, to 0x01 to pop the frunk, or 0x04 to open/close the trunk (0x05 should open both)
|
||||
print("Opening Frunk...")
|
||||
p.can_send(0x248, "\x01\x00\x00\x00\x00\x00\x00\x00", bus_num)
|
||||
|
||||
#Back to safety...
|
||||
print("Disabling output on Panda...")
|
||||
p.set_safety_mode(Panda.SAFETY_NOOUTPUT)
|
||||
|
||||
print("Reading VIN from 0x568. This is painfully slow and can take up to 3 minutes (1 minute per message; 3 messages needed for full VIN)...")
|
||||
|
||||
cnt = 0
|
||||
vin = {}
|
||||
while True:
|
||||
#Read the VIN
|
||||
can_recv = p.can_recv()
|
||||
for address, _, dat, src in can_recv:
|
||||
if src == body_bus_num:
|
||||
if address == 1384: #0x568 is VIN
|
||||
vin_index = int(binascii.hexlify(dat)[:2]) #first byte is the index, 00, 01, 02
|
||||
vin_string = binascii.hexlify(dat)[2:] #rest of the string is the actual VIN data
|
||||
vin[vin_index] = vin_string.decode("hex")
|
||||
print("Got VIN index " + str(vin_index) + " data " + vin[vin_index])
|
||||
cnt += 1
|
||||
#if we have all 3 parts of the VIN, print it and break out of our while loop
|
||||
if cnt == 3:
|
||||
print("VIN: " + vin[0] + vin[1] + vin[2][:3])
|
||||
break
|
||||
|
||||
if __name__ == "__main__":
|
||||
tesla_tester()
|
||||
Reference in New Issue
Block a user