tags: DoIP, Python, TCP Framing, Socket Programming, 파서, 네트워크, DevBJ
지난 [DoIP 마스터 가이드 5부작]을 통해 DoIP의 전체적인 숲을 둘러보셨다면, 이제 진짜 소매를 걷어붙이고 코드를 만져볼 시간입니다.
실무에서 DoIP 스택을 상용 솔루션(Vector 등) 없이 직접 파이썬(Python)이나 C/C++로 바닥부터 구현해야 할 때, 개발자들을 가장 먼저 좌절하게 만드는 녀석이 바로 'TCP 스트림 프레이밍(Stream Framing)'입니다. (지난 시즌 1편에서 뼈 아프게 강조했던 그 녀석이죠!)
오늘은 파이썬(Python) 소켓 프로그래밍을 활용하여, 물밀듯이 밀려오는 TCP 스트림 속에서 완벽하게 DoIP 패킷 한 덩어리를 썰어내는 파서(Parser) 구현법을 실전 코드로 깊게 파헤쳐 보겠습니다.
📌 [DoIP 마스터 가이드 시즌 2: 실전 심화편 목차][IT/자동차] - [DoIP 마스터 가이드 시즌2 - 1편] 파이썬(Python)으로 직접 짜보는 DoIP 파서와 TCP 프레이밍 구현
[IT/자동차] - [DoIP 마스터 가이드 시즌2 - 2편] Wireshark 실전! 패킷 덤프(PCAP)로 뜯어보는 DoIP 흐름과 디버깅
[IT/자동차] - [DoIP 마스터 가이드 시즌2 - 3편] UDS 딥다이브! 악명 높은 보안 접속(0x27)과 루틴 제어(0x31) 해부
[IT/자동차] - [DoIP 마스터 가이드 시즌2 - 4편] 실전 엣지 케이스! "왜 이 상황에서만 통신이 죽을까? (完)
1. 실패하는 파서의 전형적인 모습 (초보자의 덫)
먼저, 99%의 입문자가 작성하고 장렬하게 에러를 뿜는 코드를 보겠습니다.
import socket
# [위험한 코드] 절대 이렇게 짜면 안 됩니다!
def bad_doip_receiver(sock):
while True:
# 1. 4096 바이트를 무작정 읽는다.
data = sock.recv(4096)
if not data:
break
# 2. 받은 데이터가 무조건 DoIP 패킷 1개일 것이라 착각한다.
payload_type = int.from_bytes(data[2:4], byteorder='big')
payload_length = int.from_bytes(data[4:8], byteorder='big')
# 3. 만약 패킷이 쪼개져서 3바이트만 들어왔다면? -> Index Error 발생!
# 4. 패킷 2개가 뭉쳐 들어왔다면? -> 뒤에 있는 패킷은 버려짐!
print(f"Type: {payload_type}, Length: {payload_length}")UDP는 메시지 경계가 유지되지만, TCP는 그저 바이트의 연속적인 흐름(Stream)일 뿐입니다. sock.recv()가 한 번에 DoIP 패킷 한 개를 온전히 던져줄 것이란 기대는 완전히 버려야 합니다.
2. 완벽한 프레이밍(Framing)을 위한 3단계 전략
안전한 파서를 만들기 위해서는 '수신 버퍼(Receive Buffer)'를 두고 다음 3단계를 무한 반복하는 구조로 설계해야 합니다.
- 데이터 누적 (Accumulate): 소켓에서 읽은 데이터를 무조건
rx_buffer에 이어 붙인다. - 헤더 검사 (Check Header):
rx_buffer의 길이가 최소 8바이트(Generic Header 길이) 이상인지 확인하고, 8바이트가 넘으면Payload Length를 파싱하여 '기대하는 전체 패킷 길이'를 계산한다. - 데이터 절삭 (Extract & Slice):
rx_buffer에 '기대하는 전체 패킷 길이'만큼 데이터가 모였다면, 딱 그만큼만 잘라내어(Slice) 처리하고, 남은 데이터는 버퍼에 남겨둔다.
3. 실전 파이썬 코드 구현 (DevBJ 스타일)
자, 위 전략을 파이썬 코드로 우아하게 구현해 보겠습니다. 실무에 바로 가져다 쓰셔도 좋을 만큼 방어적으로 작성했습니다.
import socket
import struct
class DoipParser:
def __init__(self):
self.rx_buffer = bytearray()
self.DOIP_HEADER_SIZE = 8
def process_stream(self, new_data):
self.rx_buffer.extend(new_data)
packets = []
while True:
# 1. 헤더(8바이트)조차 다 안 모였으면 대기
if len(self.rx_buffer) < self.DOIP_HEADER_SIZE:
break
# 2. 8바이트가 모였다면 Payload Length 파싱
protocol_ver, inv_ver, payload_type, payload_length = struct.unpack(
'!BBHI', self.rx_buffer[:self.DOIP_HEADER_SIZE]
)
if protocol_ver ^ inv_ver != 0xFF:
print("[에러] DoIP 헤더 버전 오류! 버퍼 초기화 진행")
self.rx_buffer.clear()
break
total_packet_size = self.DOIP_HEADER_SIZE + payload_length
# 3. 전체 패킷 길이만큼 버퍼에 데이터가 찼는지 확인
if len(self.rx_buffer) < total_packet_size:
break
# 4. 완벽한 패킷 1개 썰어내기 (Slice)
complete_packet = self.rx_buffer[:total_packet_size]
# 5. 처리한 패킷은 버퍼에서 날려버림
self.rx_buffer = self.rx_buffer[total_packet_size:]
packets.append({
"type": hex(payload_type),
"length": payload_length,
"payload": complete_packet[self.DOIP_HEADER_SIZE:]
})
return packets💡 코드 핵심 포인트 리뷰
bytearray()사용: 파이썬에서 잦은 문자열(바이트) 결합과 슬라이싱을 할 때는 불변(immutable)인bytes객체보다 가변(mutable)인bytearray를 쓰는 것이 메모리 관리 측면에서 훨씬 유리합니다.struct.unpack('!BBHI', ...): 네트워크 바이트 오더(Big-endian,!)에 맞춰 1바이트(B), 2바이트(H), 4바이트(I)를 아주 깔끔하게 파싱하는 파이썬의 강력한 도구입니다.
📝 1편 마무리
이제 아무리 데이터가 잘게 쪼개져서 들어오거나, 무식하게 뭉쳐서 들어와도 절대 죽지 않는 튼튼한 DoIP 파서를 가지게 되셨습니다. 이 파서 클래스 하나만 잘 만들어두면, 그 위에 라우팅 액티베이션이든 UDS든 비즈니스 로직만 예쁘게 얹으시면 됩니다.
코드가 준비되었으니, 다음 시간은 실전 네트워크 분석입니다. 코드가 뭔가 이상하게 동작할 때, 우리의 눈을 밝혀줄 Wireshark 패킷 덤프(PCAP)를 헥사(Hex) 레벨에서 직접 뜯어보는 방법을 알아보겠습니다.
👉 [다음 글 보기: 시즌2 - 2편. Wireshark 패킷 덤프(PCAP)로 뜯어보는 DoIP 통신 흐름과 디버깅]
코드 구현하시다가 막히거나 궁금한 점이 있다면 언제든 댓글 남겨주세요! DevBJ였습니다.