How to Use remoteCAN

1. Overview Architecture

Several challenges in the CAN traffic analysis and vehicle security categories connect to live, containerized truck network simulators (such as the RAMN 4-ECU vehicle model simulated in Renode).

These simulators emulate real commercial vehicle ECUs exchanging standard SAE J1939 telematics and powertrain traffic. Rather than requiring physical vehicle hardware or specialized CAN adapters, simulation endpoints expose a remote SLCAN serial interface over TCP.

2. In-Browser CAN Analyzer (SloppyCAN) Recommended for Beginners

The easiest way to inspect, trace, and interact with vehicle CAN traffic in real time without local software installation is via the web-based SloppyCAN protocol analyzer:

Launch SloppyCAN Web Analyzer →

Direct URL: https://nmfta-repo.github.io/sloppycan/

How to Connect:

1. Open SloppyCAN in your browser (Chrome, Edge, or Firefox).
2. In the connection bar / popover, set the Adapter to SLCAN (socket tunnel).
3. Enter the simulator connection URL provided in the challenge (e.g. socket://sim.fleetcyberchallenge.com:33939 or sim.fleetcyberchallenge.com:33939).
4. Set the baud rate to 500k (standard SAE J1939 commercial vehicle rate).
5. Click Connect. The interface will start streaming live J1939 CAN IDs, frame counters, and data bytes.

Key Features:

• CAN ID Monitor: Real-time frequency, period, and delta tracking per Arbitration ID.
• J1939 Decoding: Automatic decoding of Parameter Group Numbers (PGNs), Source Addresses (SA), and Priorities.
• Transmit & Fuzzing: Send manual request frames (e.g. Request PGN 0x18EA00F9) or run periodic diagnostic probes.

3. Interacting with Python (python-can & Scapy) Scripting

For automated solvers, protocol reverse engineering, and multi-step exploitation, Python provides the most versatile toolkit. The standard python-can library natively supports TCP-based SLCAN channels via the socket:// URL scheme.

3.1 Connecting & Listening to J1939 Traffic

Establish a connection to the remote simulation target and listen for standard J1939 broadcast traffic:

import can

# Initialize SLCAN bus connected to remote TCP simulator
bus = can.Bus(
    interface="slcan",
    channel="socket://sim.fleetcyberchallenge.com:33939",
    bitrate=500000
)

# Receive and inspect incoming frames
print("Listening for J1939 traffic...")
for _ in range(10):
    msg = bus.recv(timeout=2.0)
    if msg is not None:
        print(f"CAN ID: 0x{msg.arbitration_id:08X} (Ext: {msg.is_extended_id}) | Data: {msg.data.hex()}")

bus.shutdown()

3.2 Requesting Active J1939 Addresses (Address Claim Enumeration)

To discover all active ECUs / Controller Applications (CAs) on the network, send a global Request for Address Claimed frame (PGN 59904 / 0xEA00 requesting PGN 60928 / 0xEE00). In SAE J1939, requested PGNs are transmitted in 3-byte little-endian format (0x00, 0xEE, 0x00):

import time
import can

bus = can.Bus(
    interface="slcan",
    channel="socket://sim.fleetcyberchallenge.com:33939",
    bitrate=500000
)

# Broadcast Request for Address Claimed (PGN 59904 / 0xEA00 -> PGN 60928 / 0xEE00)
# Arbitration ID: Priority 6 (0x18), PGN 0xEA00, Global DA 0xFF, Tool SA 0xF1 -> 0x18EAFFF1
req_msg = can.Message(
    arbitration_id=0x18EAFFF1,
    data=[0x00, 0xEE, 0x00],  # PGN 60928 in 3-byte Little-Endian format
    is_extended_id=True
)
bus.send(req_msg)
print("[+] Broadcast Address Claim request sent. Collecting active addresses...")

# Collect responding Source Addresses over 5 seconds
claimed_addresses = set()
deadline = time.time() + 5.0

while time.time() < deadline:
    msg = bus.recv(timeout=0.5)
    if msg is None:
        continue
    # PGN 60928 (0xEE00) Address Claim response format: 0x18EEFF<SA>
    pf = (msg.arbitration_id >> 16) & 0xFF
    sa = msg.arbitration_id & 0xFF
    if pf == 0xEE:
        if sa not in claimed_addresses:
            claimed_addresses.add(sa)
            print(f"  -> Discovered active ECU at Source Address: 0x{sa:02X} (decimal {sa})")

print(f"[+] Total active addresses claimed: {[hex(a) for a in sorted(claimed_addresses)]}")
bus.shutdown()

3.3 Requesting & Decoding J1939 NAMEs (64-bit Device Identity)

Every J1939 ECU transmits a unique 64-bit (8-byte) NAME during Address Claiming (defined in SAE J1939-81). You can request an individual ECU's NAME by sending a unicast request to its specific Destination Address (DA). The 8-byte payload contains identity, manufacturer, and function bitfields:

import struct
import time
import can

def decode_j1939_name(name_bytes: bytes) -> dict:
    """Decodes 8-byte (64-bit) SAE J1939 NAME bitfields per SAE J1939-81."""
    val = struct.unpack("<Q", name_bytes)[0]
    return {
        "identity_number": val & 0x1FFFFF,                     # Bits 0-20 (21 bits)
        "manufacturer_code": (val >> 21) & 0x7FF,              # Bits 21-31 (11 bits)
        "ecu_instance": (val >> 32) & 0x07,                    # Bits 32-34 (3 bits)
        "function_instance": (val >> 35) & 0x1F,               # Bits 35-39 (5 bits)
        "function": (val >> 40) & 0xFF,                        # Bits 40-47 (8 bits)
        "vehicle_system": (val >> 49) & 0x7F,                  # Bits 49-55 (7 bits)
        "vehicle_system_instance": (val >> 56) & 0x0F,         # Bits 56-59 (4 bits)
        "industry_group": (val >> 60) & 0x07,                  # Bits 60-62 (3 bits)
        "arbitrary_address_capable": bool((val >> 63) & 0x01),  # Bit 63 (1 bit)
    }

bus = can.Bus(
    interface="slcan",
    channel="socket://sim.fleetcyberchallenge.com:33939",
    bitrate=500000
)

# Send targeted unicast request to a specific ECU (e.g. target DA 0xCA)
target_da = 0xCA
unicast_req = can.Message(
    arbitration_id=0x18EA00F1 | (target_da << 8),  # PGN 59904 with PDU Specific = target_da
    data=[0x00, 0xEE, 0x00],
    is_extended_id=True
)
bus.send(unicast_req)

deadline = time.time() + 5.0
while time.time() < deadline:
    msg = bus.recv(timeout=0.5)
    if msg and ((msg.arbitration_id >> 16) & 0xFF) == 0xEE and (msg.arbitration_id & 0xFF) == target_da:
        raw_name = bytes(msg.data)
        print(f"[+] Received 64-bit NAME from SA 0x{target_da:02X}: {raw_name.hex()}")
        parsed = decode_j1939_name(raw_name)
        for field, value in parsed.items():
            print(f"    {field}: {value}")
        # Note: If identity fields embed ASCII text, extract printable characters:
        ascii_chars = "".join(chr(b) for b in raw_name if 32 <= b <= 126)
        print(f"    ASCII representation: {ascii_chars}")
        break

bus.shutdown()

3.4 Multi-Request Seed-Key Security Access (KWP2000 / UDS)

Protected diagnostics on truck ECUs often require a multi-request handshake (ISO 14229 UDS / ISO 14230 KWP2000): starting an extended session (0x10), requesting a security seed (0x27 0x01), calculating the authorization key, submitting the key (0x27 0x02), and reading protected identifiers (0x22) using ISO 15765-2 (ISO-TP) multi-frame transfer:

import time
import can

bus = can.Bus(
    interface="slcan",
    channel="socket://sim.fleetcyberchallenge.com:33939",
    bitrate=500000
)

# Diagnostic addressing (e.g. Proprietary Request to ECU target 0xE6 from Tool 0xF1)
REQ_ID = 0x18EFE6F1
RESP_ID = 0x18EFF1E6

def drain(bus):
    """Flushes unread frames so responses are received immediately."""
    while bus.recv(timeout=0.02) is not None:
        pass

def send_and_recv(bus, data_bytes, timeout=4.0):
    drain(bus)
    bus.send(can.Message(arbitration_id=REQ_ID, is_extended_id=True, data=data_bytes))
    deadline = time.time() + timeout
    while time.time() < deadline:
        msg = bus.recv(timeout=0.2)
        if msg and msg.arbitration_id == RESP_ID:
            return bytes(msg.data)
    return None

# Step 1: Start Extended Diagnostic Session (Service 0x10 0x85 or 0x10 0x03)
print("[1] Entering Diagnostic Session (0x10 0x85)...")
resp = send_and_recv(bus, [0x02, 0x10, 0x85, 0, 0, 0, 0, 0])
if not resp or resp[1] != 0x50:
    raise RuntimeError(f"Session failed: {resp.hex() if resp else 'timeout'}")
print(f"    Diagnostic session active: {resp.hex()}")

# Step 2: Request Security Access Seed (Service 0x27 0x01)
print("[2] Requesting Security Access Seed (0x27 0x01)...")
resp = send_and_recv(bus, [0x02, 0x27, 0x01, 0, 0, 0, 0, 0])
if not resp or resp[1] != 0x67 or resp[2] != 0x01:
    raise RuntimeError(f"Seed request rejected: {resp.hex() if resp else 'timeout'}")

seed = int.from_bytes(resp[3:7], "big")
print(f"    Received Seed: 0x{seed:08X}")

# Step 3: Calculate Key from Seed
# (In practice: consult reverse-engineered algorithms, lookup tables, or affine relations)
def compute_key(seed_val: int) -> int:
    return (0x1337 * seed_val + 0xC0DE) & 0xFFFFFFFF  # Example calculation

key = compute_key(seed)
print(f"[3] Calculated Key: 0x{key:08X}")

# Step 4: Submit Security Key (Service 0x27 0x02)
key_bytes = key.to_bytes(4, "big")
print("[4] Submitting Key (0x27 0x02)...")
resp = send_and_recv(bus, [0x06, 0x27, 0x02, *key_bytes, 0])
if not resp or resp[1] != 0x67 or resp[2] != 0x02:
    raise RuntimeError(f"Security unlock failed: {resp.hex() if resp else 'timeout'}")
print(f"    Security Access Granted! Positive response: {resp.hex()}")

# Step 5: Read Protected Data Identifier (0x22 0x00 0x01) with ISO-TP Reassembly
print("[5] Querying Protected Data Identifier (0x22 0x00 0x01)...")
drain(bus)
bus.send(can.Message(arbitration_id=REQ_ID, is_extended_id=True, data=[0x03, 0x22, 0x00, 0x01, 0, 0, 0, 0]))

payload = bytearray()
total_len = 0
deadline = time.time() + 5.0

while time.time() < deadline:
    msg = bus.recv(timeout=0.3)
    if not msg or msg.arbitration_id != RESP_ID:
        continue
    data = bytes(msg.data)
    frame_type = (data[0] >> 4) & 0x0F
    if frame_type == 0x0:  # Single Frame (SF)
        sf_len = data[0] & 0x0F
        payload = data[1:1 + sf_len]
        break
    elif frame_type == 0x1:  # First Frame (FF)
        total_len = ((data[0] & 0x0F) << 8) | data[1]
        payload.extend(data[2:])
        # Send ISO-TP Flow Control (0x30 = CTS, BlockSize=0, STmin=0)
        bus.send(can.Message(arbitration_id=REQ_ID, is_extended_id=True, data=[0x30, 0x00, 0x00, 0, 0, 0, 0, 0]))
    elif frame_type == 0x2:  # Consecutive Frame (CF)
        payload.extend(data[1:])
        if len(payload) >= total_len:
            break

print(f"    Decoded Payload ({len(payload)} bytes): {bytes(payload)}")
bus.shutdown()

3.5 Preparing pyserial for python-can & Scapy (slcan socket:// Optimization)

Both python-can (using the slcan interface) and Scapy (using CANSocket) rely on pyserial under the hood to communicate with serial devices and remote socket:// streams. However, standard pyserial's socket handler (serial.urlhandler.protocol_socket) contains a known limitation where its in_waiting property returns 1 instead of querying the actual number of buffered bytes waiting in the socket.

Because serial CAN readers poll in_waiting to batch incoming bytes efficiently, this bug can throttle throughput down to ~8 frames/second, causing receive loops to lag severely behind background J1939 traffic and miss rapid response frames.

To ensure real-time responsiveness (20–50ms response times) across all scripts, tools, and Jupyter notebooks on Windows, Linux, and macOS, follow either of the recommended options below:

Option A: Install a Fixed pyserial Fork in requirements.txt (Recommended)

If managing dependencies via a virtual environment or container, install the fixed pyserial fork directly using git:

# requirements.txt for python-can and Scapy with fixed pyserial socket support
python-can>=4.3.0
scapy>=2.5.0

# Install pyserial with socket in_waiting fix directly from git repository fork:
pyserial @ git+https://github.com/BenGardiner/pyserial.git@fix-protocol-socket-in-waiting

Option B: Cross-Platform Runtime Monkey-Patch & Velocity Drain

If you cannot alter your installed packages or are working inside an interactive notebook, place this cross-platform patch and read-velocity drain helper at the top of your script or notebook cell before creating any can.Bus or Scapy socket instances:

import sys
import time
import serial.urlhandler.protocol_socket
from scapy.all import sniff

# Cross-platform patch for pyserial in_waiting over socket:// URLs
if sys.platform == "win32":
    import ctypes
    _FIONREAD = 0x4004667F  # Winsock FIONREAD ioctl
    _ioctlsocket = ctypes.windll.ws2_32.ioctlsocket
    _ioctlsocket.argtypes = [ctypes.c_uint64, ctypes.c_long, ctypes.POINTER(ctypes.c_ulong)]
    _ioctlsocket.restype = ctypes.c_int

    def _patched_in_waiting(self):
        if not self.is_open:
            raise serial.PortNotOpenError
        argp = ctypes.c_ulong(0)
        res = _ioctlsocket(self._socket.fileno(), _FIONREAD, ctypes.byref(argp))
        return argp.value if res == 0 else 0
else:
    import fcntl
    import termios
    import struct

    def _patched_in_waiting(self):
        if not self.is_open:
            raise serial.PortNotOpenError
        buf = struct.pack("I", 0)
        out = fcntl.ioctl(self._socket.fileno(), termios.FIONREAD, buf)
        return struct.unpack("I", out)[0]

# Apply patch to protocol_socket Serial class used by python-can and Scapy
serial.urlhandler.protocol_socket.Serial.in_waiting = property(_patched_in_waiting)

def drain_can_socket(sock, batch_size=400, live_traffic_threshold=0.05):
    """
    Drains a CAN socket buffer using read velocity to clear stale backlog.
    At 500 kbps, 400 frames take ~100ms on the wire.
    If pulling 400 frames takes >= 50ms, we have caught up to live wire traffic.
    """
    while True:
        start_time = time.monotonic()
        chunk = sniff(opened_socket=sock, count=batch_size, timeout=live_traffic_threshold * 1.5)
        elapsed = time.monotonic() - start_time
        if not chunk or len(chunk) < batch_size or elapsed >= live_traffic_threshold:
            break

4. Linux SocketCAN Integration CLI Tools

On Linux, you can attach the remote TCP socket directly to a local SocketCAN interface (such as can0) using socat and slcand:

# Step 1: Bridge the remote TCP port to a local virtual serial PTY
socat pty,link=/tmp/remotecan,raw,echo=0 tcp:sim.fleetcyberchallenge.com:33939 &

# Step 2: Attach slcand to the PTY (S6 = 500 kbps)
sudo slcand -o -c -s6 /tmp/remotecan can0
sudo ip link set up can0

# Step 3: Use standard can-utils
candump can0
cansniffer can0
cansend can0 18EA00F9#00FE00

5. Commercial Vehicle Protocol Quick Reference Reference