Hands-On: Ethernet Traffic Capture

#!/bin/bash
# Capture automotive Ethernet traffic (requires Linux with Ethernet interface)
# Replace eth0 with your interface (e.g., enp3s0 or the TAP interface)

IFACE="eth0"
CAPTURE_FILE="vehicle_trace.pcap"

# Basic capture: all frames on interface
tcpdump -i $IFACE -w $CAPTURE_FILE &
TCPDUMP_PID=$!
echo "Capturing on $IFACE for 30 seconds..."
sleep 30
kill $TCPDUMP_PID

echo "Frames captured: $(tcpdump -r $CAPTURE_FILE | wc -l)"

# Filter: only SOME/IP traffic (UDP port 30490 for SOME/IP-SD; app ports vary)
tcpdump -r $CAPTURE_FILE -w someip_only.pcap 'udp port 30490 or udp portrange 30000-30499'

# Filter: only DoIP traffic (TCP/UDP port 13400)
tcpdump -r $CAPTURE_FILE -w doip_only.pcap 'tcp port 13400 or udp port 13400'

# Filter: only gPTP (EtherType 0x88F7)
tcpdump -r $CAPTURE_FILE -w gptp.pcap 'ether proto 0x88F7'

# Open in Wireshark for analysis
wireshark vehicle_trace.pcap &

#!/usr/bin/env python3
# Manually decode a raw SOME/IP message from bytes
# (Wireshark does this automatically, but understanding the format is essential)

import struct

# Example: raw SOME/IP Request frame (captured from pcap)
# Service 0x0123, Method 0x0ABC, length 12, client 0x0001, session 0x0001
raw_someip = bytes([
    0x01, 0x23,    # Service ID
    0x0A, 0xBC,    # Method ID
    0x00, 0x00, 0x00, 0x0C,   # Length (12 = 8 header + 4 payload)
    0x00, 0x01,    # Client ID
    0x00, 0x01,    # Session ID
    0x01,          # Protocol Version (1)
    0x01,          # Interface Version (1)
    0x00,          # Message Type: Request (0x00)
    0x00,          # Return Code: E_OK (0x00)
    # Payload: uint32 value = 0x0000AABB
    0x00, 0x00, 0xAA, 0xBB
])

def decode_someip_header(data: bytes) -> dict:
    if len(data) < 16:
        raise ValueError("Too short for SOME/IP header")
    svc_id, meth_id, length, client_id, session_id, proto_ver, iface_ver, msg_type, ret_code =         struct.unpack(">HHIHHBBBB", data[:16])
    return {
        "service_id":   f"0x{svc_id:04X}",
        "method_id":    f"0x{meth_id:04X}",
        "length":       length,
        "client_id":    f"0x{client_id:04X}",
        "session_id":   f"0x{session_id:04X}",
        "proto_ver":    proto_ver,
        "iface_ver":    iface_ver,
        "msg_type":     {0x00:"Request",0x01:"RequestNoReturn",0x02:"Notification",
                         0x80:"Response",0x81:"Error"}.get(msg_type, f"0x{msg_type:02X}"),
        "return_code":  {0x00:"E_OK",0x01:"E_NOT_OK",0x09:"E_UNKNOWN_SERVICE"}.get(ret_code, f"0x{ret_code:02X}"),
        "payload":      data[16:16+(length-8)].hex()
    }

header = decode_someip_header(raw_someip)
for k, v in header.items():
    print(f"  {k:15s}: {v}")

#!/usr/bin/env python3
# Parse pcap and report VLAN tag PCP values per protocol
from scapy.all import rdpcap, Ether, Dot1Q, IP, UDP, TCP

packets = rdpcap("vehicle_trace.pcap")

pcp_stats = {}  # dict of {(protocol, PCP): count}

for pkt in packets:
    if pkt.haslayer(Dot1Q):
        pcp = pkt[Dot1Q].prio
        if pkt.haslayer(UDP):
            dport = pkt[UDP].dport
            if dport == 13400:
                proto = "DoIP"
            elif dport == 30490:
                proto = "SOME/IP-SD"
            elif 30000 <= dport <= 30499:
                proto = "SOME/IP"
            else:
                proto = "UDP-other"
        elif pkt.haslayer(TCP):
            proto = "TCP"
        else:
            proto = "Non-IP"
        key = (proto, pcp)
        pcp_stats[key] = pcp_stats.get(key, 0) + 1

print("Protocol         | PCP | Count")
print("-" * 35)
for (proto, pcp), count in sorted(pcp_stats.items()):
    pcp_desc = ["BE","Bkg","Spare","Excl","Ctrl","Video","Voice","Net"][pcp]
    print(f"{proto:16s} | {pcp}   ({pcp_desc}) | {count}")

# Expected: DoIP = PCP 3 (diagnostics), camera SOME/IP = PCP 6 (video), SD = PCP 5