背景
前回の記事でラズパイとOBD-Ⅱ通信（インターフェースはCAN）ができるようになりました。
具体的には、SID=0x22による読み出し、SID=0x2Eによる値の書き換えができました。
今回はNegative Response Code（通称NRC）の作り込みに挑戦します。
NRCは数多くありますが、セキュリティアクセスに興味があるため、NRC：0x33（securityAccessDenied）とNRC：0x35（Invalid key）を実装してみたいと思います。
前回の記事内容はこちらをご確認ください。
SID=0x2E（Write data by Identifier）に制限をかける
SID=0x2Eとはデータの書き換えコマンドですが、データをコマンドで誰でも簡単に書き換えられることはセキュリティ上よろしくありません。
そのため、データの改ざんから保護したいデータに対してはセキュリティをかけることがあります。
DID=0x2020という領域を用意し、書き込みの要求が来たらNRC：0x33を応答する仕組みを入れてみます。
まずはSID=0x22でDID=0x2020が読み出せることを確認します（初期値は0xFF）。
次に、SID=0x2EでDID=0x2020に書き込みしてみます（書き込む値は0x99）。
NRC：0x33（securityAccessDenied）を応答できました！
応答内容を解説します。
- 0x7F：NRCを応答する際に付与される固定値（これが出たらNRCだとすぐにわかります）
- 0x2E：ラズパイに対して要求されたコマンド（書き込み要求）
- 0x33：NRC（securityAccessDinied）
Pythonスクリプト（取りあえずNRC応答する版）
import tkinter as tk
from tkinter import ttk
import can
import threading
import time
class CANTransmitterApp:
def __init__(self, master):
self.master = master
self.master.title("CAN Transmitter")
self.memory = {}
# Interface Input
self.interface_label = ttk.Label(master, text="Interface:")
self.interface_label.grid(row=0, column=0, padx=10, pady=10, sticky=tk.W)
self.interface_entry = ttk.Entry(master)
self.interface_entry.grid(row=0, column=1, padx=10, pady=10)
self.interface_entry.insert(tk.END, "can0")
# CAN ID, CAN Data, and Period Inputs
self.can_id_labels = [ttk.Label(master, text=f"CAN ID {i+1}:") for i in range(3)]
self.data_labels = [ttk.Label(master, text=f"CAN Data {i+1} (comma-separated hex values):") for i in range(3)]
self.period_labels = [ttk.Label(master, text=f"Period {i+1} (seconds):") for i in range(3)]
self.can_id_entries = [ttk.Entry(master) for i in range(3)]
self.data_entries = [ttk.Entry(master) for i in range(3)]
self.period_entries = [ttk.Entry(master) for i in range(3)]
for i in range(3):
# CAN ID
self.can_id_labels[i].grid(row=1+i, column=0, padx=10, pady=10, sticky=tk.W)
self.can_id_entries[i].grid(row=1+i, column=1, padx=10, pady=10)
self.can_id_entries[i].insert(tk.END, f"{123 + i:03X}")
# CAN Data
self.data_labels[i].grid(row=1+i, column=2, padx=10, pady=10, sticky=tk.W)
self.data_entries[i].grid(row=1+i, column=3, padx=10, pady=10)
self.data_entries[i].insert(tk.END, "11,22,33,44,55,66,77,88")
# Period
self.period_labels[i].grid(row=1+i, column=4, padx=10, pady=10, sticky=tk.W)
self.period_entries[i].grid(row=1+i, column=5, padx=10, pady=10)
self.period_entries[i].insert(tk.END, "1.0")
# Control Buttons
self.start_button = ttk.Button(master, text="Start Sending", command=self.start_sending)
self.start_button.grid(row=4, column=0, columnspan=3, padx=10, pady=10)
self.stop_button = ttk.Button(master, text="Stop Sending", command=self.stop_sending)
self.stop_button.grid(row=4, column=3, columnspan=3, padx=10, pady=10)
# Status Label
self.status_label = ttk.Label(master, text="")
self.status_label.grid(row=5, columnspan=6, padx=10, pady=10)
# Log Text
self.log_text = tk.Text(master, height=10, width=80)
self.log_text.grid(row=6, columnspan=6, padx=10, pady=10)
self.is_sending = False
self.bus = None
self.send_timers = [None, None, None]
def start_sending(self):
if self.is_sending:
self.status_label.config(text="Already sending.")
return
interface = self.interface_entry.get().strip()
can_ids = [int(entry.get().strip(), 16) for entry in self.can_id_entries]
data_strs = [entry.get().strip() for entry in self.data_entries]
periods = [float(entry.get().strip()) for entry in self.period_entries]
try:
data_list = []
for data_str in data_strs:
data = [int(byte, 16) for byte in data_str.split(',')]
if len(data) != 8:
raise ValueError("Data length must be 8 bytes.")
data_list.append(data)
except ValueError:
self.status_label.config(text="Invalid CAN data format.")
return
try:
self.bus = can.interface.Bus(channel=interface, bustype='socketcan', bitrate=500000)
self.log_message("info", f"CAN interface {interface} connected.")
except Exception as e:
self.status_label.config(text=f"Error: {str(e)}")
return
self.is_sending = True
self.receive_thread = threading.Thread(target=self.receive_messages, daemon=True)
self.receive_thread.start()
for i in range(3):
self.send_messages(can_ids[i], data_list[i], periods[i], i)
def send_messages(self, can_id, data, period, index):
message = can.Message(arbitration_id=can_id, data=data)
if self.is_sending:
try:
self.bus.send(message)
self.log_message("send", message)
except can.CanError as e:
self.log_message("error", f"Failed to send CAN message - {str(e)}")
self.status_label.config(text=f"Error: Failed to send CAN message - {str(e)}")
self.send_timers[index] = threading.Timer(period, self.send_messages, args=(can_id, data, period, index))
self.send_timers[index].start()
def receive_messages(self):
while self.is_sending:
try:
message = self.bus.recv()
if message:
self.log_message("recv", message)
# Check for UDS command with CANID 0x740
if message.arbitration_id == 0x740:
response_data = self.create_ud_response(message.data)
response_message = can.Message(arbitration_id=0x748, data=response_data)
self.bus.send(response_message)
self.log_message("send", response_message)
except Exception as e:
self.log_message("error", f"Failed to receive CAN message - {str(e)}")
def create_ud_response(self, request_data):
# ISO 14229 compliant response example
request_data_list = list(request_data)
print("Received request data:", ", ".join(f"0x{byte:02X}" for byte in request_data_list))
# SID=0x22
if request_data_list[1] == 0x22:
# DID=0x10,0x10
if request_data_list[2] == 0x10 and request_data_list[3] == 0x10:
did = (request_data_list[2], request_data_list[3])
if did in self.memory:
# value is valid
response_value = self.memory[did]
else:
# value is not valid
response_value = 0xFF
# responce data
return [0x04] + [0x62] + [0x10, 0x10, response_value] + [0x00] * 3
# DID=0x20,0x20
if request_data_list[2] == 0x20 and request_data_list[3] == 0x20:
did = (request_data_list[2], request_data_list[3])
if did in self.memory:
# value is valid
response_value = self.memory[did]
else:
# value is not valid
response_value = 0xFF
# responce data
return [0x04] + [0x62] + [0x20, 0x20, response_value] + [0x00] * 3
# SID=0x2E
elif request_data_list[1] == 0x2E:
# DID=0x10,0x10
if request_data_list[2] == 0x10 and request_data_list[3] == 0x10:
# write memory
value_to_write = request_data_list[4] if len(request_data_list) > 4 else 0x00
did = (0x10, 0x10)
self.memory[did] = value_to_write
# responce data
return [0x03] + [0x6E] + [0x10, 0x10] + [0x00] * 4
# DID=0x20,0x20
if request_data_list[2] == 0x20 and request_data_list[3] == 0x20:
# write memory
#value_to_write = request_data_list[4] if len(request_data_list) > 4 else 0x00
#did = (0x20, 0x20)
#self.memory[did] = value_to_write
# responce data
#return [0x03] + [0x6E] + [0x10, 0x10] + [0x00] * 4
return [0x03, 0x7F, 0x2E, 0x33, 0x00, 0x00, 0x00, 0x00] #securityAccessDenied
return [0x7F, request_data_list[0], 0x31] # Negative response
def stop_sending(self):
if not self.is_sending:
self.status_label.config(text="Not currently sending.")
return
self.is_sending = False
for timer in self.send_timers:
if timer:
timer.cancel()
if self.bus:
self.bus.shutdown()
self.status_label.config(text="Stopped sending.")
def log_message(self, msg_type, message):
if msg_type == "send":
log_entry = f"Send : ID={hex(message.arbitration_id)} Data={','.join(f'0x{byte:02X}' for byte in message.data)}\n"
elif msg_type == "recv":
log_entry = f"Recv : ID={hex(message.arbitration_id)} Data={','.join(f'0x{byte:02X}' for byte in message.data)}\n"
elif msg_type == "error":
log_entry = f"Error : {message}\n"
else:
log_entry = f"{message}\n"
self.log_text.insert(tk.END, log_entry)
self.log_text.see(tk.END)
def main():
root = tk.Tk()
app = CANTransmitterApp(root)
root.mainloop()
if __name__ == "__main__":
main()
SID=0x27（Security Access）を実装
DID=0x2020への書き込み要求に対して制限をかけることができました。
しかし、このままではこの領域がずっと利用できない状態のため可用性がありません。
そのため、正規のユーザであることをラズパイが認証し、正規のユーザーであることを検証後、書き込み要求を受理できるようにします。
そもそも、SID 0x27（Security Access）とは、暗号技術を利用し、正規のユーザーを見分けるためのUDS仕様です。
まずは、先ほどと同様にSID 0x2E、DID=0x2020が拒否されることを確認します。
次に、SID=0x27、SubFunction=0x01でSeedをリクエストしてみます。
Seed：A8 3A 63 B3 C1をラズパイが応答してくれました。
応答内容を解説します。
- 67：応答SID
- 01：応答Subfuntion
- A8 3A 63 B3 C1：Seed
次にSID=0x27、SubFunction=0x02でKeyを送信してみます。
ラズパイ側で演算したkeyと一致しているため、解除に成功しました。
ラズパイは今、アクセスしているユーザーが正規のユーザーであると認識した状態です。
応答データそのものはフォーマットが変ですが、後で修正します。。。
次に再度SID=0x2Eによる書き込みを試行します。
DIDは0x20 20、書き込む値は0x99です。
SID=0x2Eによる書き込みに対し、ラズパイが正常応答をしてくれました！
DID：0x20 20の領域に0x99が書き込めているはずですので、SID=0x22で値を確認します。
DID=0x2020の領域に保存されている値が0xFF⇒0x99に更新されています！
応答内容を解説します。
- 62：応答SID
- 20 20：DID
- 99：保存されている値
Security Accessのアルゴリズム
一般的には、SeedやKeyに関し、十分に長い値を採用すべきなのですが、8Byte以上になるとFlow Controlなどを考慮しないといけなくなります。
今回の実験とは本質的に関係ないため、一旦は通信が8Byteに収まる長さにします。
ラズパイ側でSeedからKeyを演算するアルゴリズムはこちらです。
- 40bitの乱数（seed）を生成する
- 40bitの別の値（Prekey）とXORする
- XORした結果をSHA-256に入力する（前半128bitだけ利用する）
- 出力結果の0Byte目、5Byte目、9Byte目、13Byte目を別の値（key）に保持する
- 受信した鍵（Recvkey）とkeyを比較し、一致していればセキュリティを解除
Prekeyとは、いわゆる「事前共有鍵」と呼ばれるもので、認証をする側とされる側でそれぞれ同じ値を所持している必要があります。
よく使われる手法ですが、都度都度生成される乱数と事前共有鍵をXOR（排他的論理和）することで、認証に使われる値の基礎が出来上がります。
しかし、この値が仮に漏えいした場合、XORしただけなので事前共有鍵（Prekey）が算出できてしまいます。
そのため、認証に使われる値をハッシュ化します。
ハッシュ化とは、ハッシュ関数（SHAなど）と呼ばれる特殊な計算方法を用い、別の全く異なる値に変換する方法です。
このハッシュ化された値は元に戻すことができない（不可逆性）性質があります。
この「元に戻せない」性質がハッシュ化の旨味でして、仮にハッシュ化された値が漏えいしても、事前共有鍵は算出不可能です。
一方で、ハッシュ関数には「同じ値からは同じ値が算出される」性質があります。
そのため、事前共有鍵を守ったところで、ハッシュ化された値が漏えいすると、一番重要なKeyが算出されてしまいます。
ここまで来て、Seedの旨味が活きてきます。
Seedは認証の都度生成するので、認証に使われる値は毎回異なる値となります。
こういった一連の防御手法はチャレンジ&レスポンスと呼ばれております。
詳しく知りたい方は、こちらの書籍がとてもわかりやすく解説くださっています（私の愛読書です）。
文字だけだと混乱すると思いますので、以下にシーケンス的なものを用意しました。
Security Accessのシーケンス
パソコン側の処理はまだ実装していないので、本来こういうことをやりたかった、というシーケンス図です。
本来は送信側（パソコン）で演算してkeyを送信するものですが、ラズパイ側でkeyの値をコンソールに出力させ、その値を直接USB2CANのGUIから送信する形を取りました。
Key Bytes (Hex): の部分がkeyに該当します。
これはSID 0x27、SubFunction=0x01を受信したタイミングで生成するようにしています。
一番下に出力されている”1″という数字に関しては、セキュリティ解除状態であることを示すデバッグ用の値です。
プログラム起動時に0（セキュリティ状態）を設定するようにしています。
pythonスクリプト（セキュリティアクセス対応版）
完成したスクリプトはこちらです。
import tkinter as tk
from tkinter import ttk
import can
import threading
import time
import os
import hashlib
class CANTransmitterApp:
def __init__(self, master):
self.master = master
self.master.title("CAN Transmitter")
self.memory = {}
self.security = 0 #lock
self.key = {}
self.Recvkey = {}
# Interface Input
self.interface_label = ttk.Label(master, text="Interface:")
self.interface_label.grid(row=0, column=0, padx=10, pady=10, sticky=tk.W)
self.interface_entry = ttk.Entry(master)
self.interface_entry.grid(row=0, column=1, padx=10, pady=10)
self.interface_entry.insert(tk.END, "can0")
# CAN ID, CAN Data, and Period Inputs
self.can_id_labels = [ttk.Label(master, text=f"CAN ID {i+1}:") for i in range(3)]
self.data_labels = [ttk.Label(master, text=f"CAN Data {i+1} (comma-separated hex values):") for i in range(3)]
self.period_labels = [ttk.Label(master, text=f"Period {i+1} (seconds):") for i in range(3)]
self.can_id_entries = [ttk.Entry(master) for i in range(3)]
self.data_entries = [ttk.Entry(master) for i in range(3)]
self.period_entries = [ttk.Entry(master) for i in range(3)]
for i in range(3):
# CAN ID
self.can_id_labels[i].grid(row=1+i, column=0, padx=10, pady=10, sticky=tk.W)
self.can_id_entries[i].grid(row=1+i, column=1, padx=10, pady=10)
self.can_id_entries[i].insert(tk.END, f"{123 + i:03X}")
# CAN Data
self.data_labels[i].grid(row=1+i, column=2, padx=10, pady=10, sticky=tk.W)
self.data_entries[i].grid(row=1+i, column=3, padx=10, pady=10)
self.data_entries[i].insert(tk.END, "11,22,33,44,55,66,77,88")
# Period
self.period_labels[i].grid(row=1+i, column=4, padx=10, pady=10, sticky=tk.W)
self.period_entries[i].grid(row=1+i, column=5, padx=10, pady=10)
self.period_entries[i].insert(tk.END, "1.0")
# Control Buttons
self.start_button = ttk.Button(master, text="Start Sending", command=self.start_sending)
self.start_button.grid(row=4, column=0, columnspan=3, padx=10, pady=10)
self.stop_button = ttk.Button(master, text="Stop Sending", command=self.stop_sending)
self.stop_button.grid(row=4, column=3, columnspan=3, padx=10, pady=10)
# Status Label
self.status_label = ttk.Label(master, text="")
self.status_label.grid(row=5, columnspan=6, padx=10, pady=10)
# Log Text
self.log_text = tk.Text(master, height=10, width=80)
self.log_text.grid(row=6, columnspan=6, padx=10, pady=10)
self.is_sending = False
self.bus = None
self.send_timers = [None, None, None]
def start_sending(self):
if self.is_sending:
self.status_label.config(text="Already sending.")
return
interface = self.interface_entry.get().strip()
can_ids = [int(entry.get().strip(), 16) for entry in self.can_id_entries]
data_strs = [entry.get().strip() for entry in self.data_entries]
periods = [float(entry.get().strip()) for entry in self.period_entries]
try:
data_list = []
for data_str in data_strs:
data = [int(byte, 16) for byte in data_str.split(',')]
if len(data) != 8:
raise ValueError("Data length must be 8 bytes.")
data_list.append(data)
except ValueError:
self.status_label.config(text="Invalid CAN data format.")
return
try:
self.bus = can.interface.Bus(channel=interface, bustype='socketcan', bitrate=500000)
self.log_message("info", f"CAN interface {interface} connected.")
except Exception as e:
self.status_label.config(text=f"Error: {str(e)}")
return
self.is_sending = True
self.receive_thread = threading.Thread(target=self.receive_messages, daemon=True)
self.receive_thread.start()
for i in range(3):
self.send_messages(can_ids[i], data_list[i], periods[i], i)
def send_messages(self, can_id, data, period, index):
message = can.Message(arbitration_id=can_id, data=data)
if self.is_sending:
try:
self.bus.send(message)
self.log_message("send", message)
except can.CanError as e:
self.log_message("error", f"Failed to send CAN message - {str(e)}")
self.status_label.config(text=f"Error: Failed to send CAN message - {str(e)}")
self.send_timers[index] = threading.Timer(period, self.send_messages, args=(can_id, data, period, index))
self.send_timers[index].start()
def receive_messages(self):
while self.is_sending:
try:
message = self.bus.recv()
if message:
self.log_message("recv", message)
# Check for UDS command with CANID 0x740
if message.arbitration_id == 0x740:
response_data = self.create_ud_response(message.data)
response_message = can.Message(arbitration_id=0x748, data=response_data)
self.bus.send(response_message)
self.log_message("send", response_message)
except Exception as e:
self.log_message("error", f"Failed to receive CAN message - {str(e)}")
def create_ud_response(self, request_data):
# ISO 14229 compliant response example
request_data_list = list(request_data)
print("Received request data:", ", ".join(f"0x{byte:02X}" for byte in request_data_list))
# SID=0x22
if request_data_list[1] == 0x22:
# DID=0x10,0x10
if request_data_list[2] == 0x10 and request_data_list[3] == 0x10:
did = (request_data_list[2], request_data_list[3])
if did in self.memory:
# value is valid
response_value = self.memory[did]
else:
# value is not valid
response_value = 0xFF
# responce data
return [0x04] + [0x62] + [0x10, 0x10, response_value] + [0x00] * 3
# DID=0x20,0x20
if request_data_list[2] == 0x20 and request_data_list[3] == 0x20:
did = (request_data_list[2], request_data_list[3])
if did in self.memory:
# value is valid
response_value = self.memory[did]
else:
# value is not valid
response_value = 0xFF
# responce data
return [0x04] + [0x62] + [0x20, 0x20, response_value] + [0x00] * 3
# SID=0x2E
elif request_data_list[1] == 0x2E:
# DID=0x10,0x10
if request_data_list[2] == 0x10 and request_data_list[3] == 0x10:
# write memory
value_to_write = request_data_list[4] if len(request_data_list) > 4 else 0x00
did = (0x10, 0x10)
self.memory[did] = value_to_write
# responce data
return [0x03] + [0x6E] + [0x10, 0x10] + [0x00] * 4
# DID=0x20,0x20
if request_data_list[2] == 0x20 and request_data_list[3] == 0x20:
print(self.security)
# write memory
#value_to_write = request_data_list[4] if len(request_data_list) > 4 else 0x00
#did = (0x20, 0x20)
#self.memory[did] = value_to_write
# responce data
#return [0x03] + [0x6E] + [0x10, 0x10] + [0x00] * 4
if self.security == 0:#lock
return [0x03, 0x7F, 0x2E, 0x33, 0x00, 0x00, 0x00, 0x00] #securityAccessDenied
else:
# write memory
value_to_write = request_data_list[4] if len(request_data_list) > 4 else 0x00
did = (0x20, 0x20)
self.memory[did] = value_to_write
# responce data
return [0x03] + [0x6E] + [0x20, 0x20] + [0x00] * 4
# SID=0x27 0x01 Request seed
elif request_data_list[1] == 0x27 and request_data_list[2] == 0x01:
#randum number generate
random_seed = os.urandom(5)
#convert to int
random_seed_int = int.from_bytes(random_seed, 'big') & ((1 << 40) - 1)
#prekey
prekey = 0x1234567890
prekey &= ((1 << 40) - 1)
#XOR
xor_result = random_seed_int ^ prekey
#SHA-128
hash_result = hashlib.sha256(xor_result.to_bytes(5, 'big')).digest()[:16]
#generate key
self.key = {
'byte_0': hash_result[0],
'byte_5': hash_result[5],
'byte_9': hash_result[9],
'byte_13': hash_result[13],
}
#debug
print("Random Seed (40 bits):", random_seed_int)
print("Prekey (40 bits):", prekey)
print("XOR Result:", xor_result)
print("SHA-128 Output:", hash_result)
print("Key Bytes:", self.key)
key_bytes_hex = [f"0x{self.key[byte]:02X}" for byte in ['byte_0', 'byte_5', 'byte_9', 'byte_13']]
print("Key Bytes (Hex):", ", ".join(key_bytes_hex))
random_seed_list = list(random_seed)
# responce data
return [0x07] + [0x67] + [0x01] + random_seed_list
# SID=0x27 0x02 Send key
elif request_data_list[1] == 0x27 and request_data_list[2] == 0x02:
#Recvkey
self.Recvkey = {
'byte_0': request_data_list[3],
'byte_5': request_data_list[4],
'byte_9': request_data_list[5],
'byte_13': request_data_list[6],
}
#Verification
if (self.Recvkey['byte_0'] == self.key['byte_0'] and
self.Recvkey['byte_5'] == self.key['byte_5'] and
self.Recvkey['byte_9'] == self.key['byte_9'] and
self.Recvkey['byte_13'] == self.key['byte_13']):
self.security = 1 #Unlock
print(self.security)
return [0x02] + [0x67] + [0x02] + [0x00] * 5
else:
return [0x02] + [0x27] + [0x35] + [0x00] * 5
return [0x7F, request_data_list[0], 0x31] # Negative response
def stop_sending(self):
if not self.is_sending:
self.status_label.config(text="Not currently sending.")
return
self.is_sending = False
for timer in self.send_timers:
if timer:
timer.cancel()
if self.bus:
self.bus.shutdown()
self.status_label.config(text="Stopped sending.")
def log_message(self, msg_type, message):
if msg_type == "send":
log_entry = f"Send : ID={hex(message.arbitration_id)} Data={','.join(f'0x{byte:02X}' for byte in message.data)}\n"
elif msg_type == "recv":
log_entry = f"Recv : ID={hex(message.arbitration_id)} Data={','.join(f'0x{byte:02X}' for byte in message.data)}\n"
elif msg_type == "error":
log_entry = f"Error : {message}\n"
else:
log_entry = f"{message}\n"
self.log_text.insert(tk.END, log_entry)
self.log_text.see(tk.END)
def main():
root = tk.Tk()
app = CANTransmitterApp(root)
root.mainloop()
if __name__ == "__main__":
main()
。
不正なkeyを送信～NRC=0x35（Invalid key）～
先ほどは解除に成功した例ですが、不正なkeyを送信した際にNRC=35（Invalid key）を応答することを確認します。
NRC=35（Invalid key）を応答できています。
が、またもフォーマットが正しくないですね（後で修正します）。
今後の目標
今回はコンソール画面上のKeyをUSB2CANのGUI上で手動でコピるというダサい手段でしたので、次回はパソコン内で演算し、その値をCANフレームとして自動送信する仕組みを構築したいと思います。
24年10月追記
パソコン内でセキュリティアクセスのKeyを算出し、CANフレームとして自動送信する仕組みを実現できました！
ご興味ございましたら是非ご覧ください。
[…] Raspberry PI 4 Model B（ラズパイ）でCAN通信~セキュリティアクセス~ […]