Build a V2X Security Simulator

Simulate V2X vehicle networks with ECDSA signatures and attack detection.

Introduction

30 Second Summary

Self-driving cars and connected vehicles talk to each other constantly, sharing their speed, position, and direction hundreds of times per second. But what happens when an attacker injects fake messages into that conversation?

In this project, you will build a V2X security simulator that uses real cryptographic signatures to authenticate vehicle messages, simulates attacks against the network, and detects those attacks in real time.

What You'll Build

A Python FastAPI web application that simulates a V2X (Vehicle-to-Everything) communication network with ECDSA message signing, attack simulation modules, and a rule-based detection engine.

By the end of this project, you'll have:

  • A working V2X security simulation backend that signs and verifies vehicle messages using real ECDSA cryptography, the same algorithm specified in IEEE 1609.2.
  • A vehicle simulation engine with pseudonym certificate rotation for privacy-preserving authentication.
  • A deployed web app on Render with an interactive dashboard showing vehicles, attacks, and detection results accessible via a public URL.
  • Secret Mission: Add a machine-learning anomaly detector using Isolation Forest that learns normal V2X traffic patterns and flags attacks the rule-based system misses.

Are there any prerequisites?

You should have basic Python familiarity (variables, functions, lists) and a GitHub account. No prior cryptography or networking experience is needed.

Build the V2X Crypto Engine

Your environment is set up and ready to go. Now it's time to build the cryptographic foundation that secures every message in your V2X network.

Real V2X vehicles use ECDSA digital signatures to prove that each broadcast is authentic. In this step, you'll implement the same algorithm (on the same elliptic curve) specified by IEEE 1609.2, plus a certificate rotation system that protects driver privacy.

In this step, get ready to:
  • Implement ECDSA key generation, message signing, and verification.
  • Build a pseudonym certificate manager that rotates keys for privacy.
  • Verify the crypto engine works end-to-end with a test script.
Implement ECDSA key generation and message signing

Every vehicle in a V2X network needs a way to prove its messages are genuine. The SECP256R1 elliptic curve (also called P-256) is the curve mandated by IEEE 1609.2 for signing Basic Safety Messages.

  • In VS Code, open app/crypto_engine.py from the Explorer sidebar.
  • Replace any existing content with the following code:
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes
from cryptography.exceptions import InvalidSignature


class V2XCryptoEngine:
    """Implements ECDSA signing/verification for V2X messages (IEEE 1609.2 style)."""

    @staticmethod
    def generate_vehicle_keypair():
        """Generate an ECDSA keypair using SECP256R1 (P-256), the curve used in V2X."""
        private_key = ec.generate_private_key(ec.SECP256R1())
        public_key = private_key.public_key()
        return private_key, public_key

    @staticmethod
    def sign_message(private_key, message_bytes: bytes) -> bytes:
        """Sign a message using ECDSA with SHA-256."""
        signature = private_key.sign(message_bytes, ec.ECDSA(hashes.SHA256()))
        return signature

    @staticmethod
    def verify_message(public_key, message_bytes: bytes, signature: bytes) -> bool:
        """Verify an ECDSA signature. Returns True if valid, False otherwise."""
        try:
            public_key.verify(signature, message_bytes, ec.ECDSA(hashes.SHA256()))
            return True
        except InvalidSignature:
            return False

What does this code do?

This class provides three static methods that form the core of V2X message security:

  • generate_vehicle_keypair() creates a private/public key pair on the SECP256R1 curve. The private key signs messages. The public key lets others verify them.
  • sign_message() takes a private key and raw message bytes, then produces a unique digital signature using ECDSA with SHA-256.
  • verify_message() checks whether a signature is valid for a given message and public key. If the signature was tampered with or signed by a different key, it returns False.

You won't see a visible change yet. The next section adds the certificate rotation layer that uses this engine.

Implement pseudonym certificate rotation

In the real V2X PKI architecture, vehicles don't use a single identity forever. They hold a batch of pseudonym certificates and rotate between them to prevent tracking.

Your certificate manager simulates this by generating a pool of keypairs and cycling through them.

  • In app/crypto_engine.py, add the following class below the V2XCryptoEngine class:
class PseudonymCertManager:
    """Manages a pool of pseudonym certificates for privacy-preserving V2X communication."""

    def __init__(self, pool_size: int = 5):
        self.pool_size = pool_size
        self.keypairs = []
        self.current_index = 0
        self._generate_pool()

    def _generate_pool(self):
        """Generate a batch of keypairs (simulating pseudonym certificate provisioning)."""
        for _ in range(self.pool_size):
            private_key, public_key = V2XCryptoEngine.generate_vehicle_keypair()
            self.keypairs.append((private_key, public_key))

    def rotate_certificate(self):
        """Rotate to the next pseudonym certificate in the pool."""
        self.current_index = (self.current_index + 1) % self.pool_size

    def get_current_signing_key(self):
        """Get the current private key for signing."""
        return self.keypairs[self.current_index][0]

    def get_current_public_key(self):
        """Get the current public key for verification."""
        return self.keypairs[self.current_index][1]

    def get_all_public_keys(self):
        """Get all public keys in the pool (for verification by other vehicles)."""
        return [kp[1] for kp in self.keypairs]

What does this code do?

The certificate manager simulates how real vehicles handle pseudonym certificates:

  • _generate_pool() pre-generates 5 keypairs at startup. In reality, an Authorisation Authority (AA) would provision these to the vehicle.
  • rotate_certificate() switches to the next keypair in the pool. This is how vehicles prevent location tracking by changing their cryptographic identity.
  • get_current_signing_key() and get_current_public_key() expose whichever keypair is currently active.
  • Save app/crypto_engine.py.
  • Verify the crypto engine works end-to-end by running this test command in your terminal:
python -c "
from app.crypto_engine import V2XCryptoEngine, PseudonymCertManager

priv, pub = V2XCryptoEngine.generate_vehicle_keypair()
sig = V2XCryptoEngine.sign_message(priv, b'Hello V2X')
print('Signature valid:', V2XCryptoEngine.verify_message(pub, b'Hello V2X', sig))

mgr = PseudonymCertManager()
sig2 = V2XCryptoEngine.sign_message(mgr.get_current_signing_key(), b'BSM')
print('Before rotation:', V2XCryptoEngine.verify_message(mgr.get_current_public_key(), b'BSM', sig2))
mgr.rotate_certificate()
sig3 = V2XCryptoEngine.sign_message(mgr.get_current_signing_key(), b'BSM')
print('After rotation:', V2XCryptoEngine.verify_message(mgr.get_current_public_key(), b'BSM', sig3))
"

You should see three lines, all showing True:

Signature valid: True
Before rotation: True
After rotation: True

What does this test prove?

The first line confirms that a freshly generated keypair can sign and verify a message. The second line confirms the certificate manager works with its initial key. The third line proves that after rotation, the new key also signs and verifies correctly.

Seeing a ModuleNotFoundError?

  • Make sure you are running the command from the project root folder (the one containing the `app/` directory), not from inside `app/` itself.
  • Confirm your virtual environment is active. You should see `(venv)` at the start of your terminal prompt.
  • Check that you installed dependencies earlier with `pip install -r requirements.txt`.

Help me debug my crypto engine import error.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here's the complete app/crypto_engine.py file:

from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes
from cryptography.exceptions import InvalidSignature


class V2XCryptoEngine:
    """Implements ECDSA signing/verification for V2X messages (IEEE 1609.2 style)."""

    @staticmethod
    def generate_vehicle_keypair():
        """Generate an ECDSA keypair using SECP256R1 (P-256), the curve used in V2X."""
        private_key = ec.generate_private_key(ec.SECP256R1())
        public_key = private_key.public_key()
        return private_key, public_key

    @staticmethod
    def sign_message(private_key, message_bytes: bytes) -> bytes:
        """Sign a message using ECDSA with SHA-256."""
        signature = private_key.sign(message_bytes, ec.ECDSA(hashes.SHA256()))
        return signature

    @staticmethod
    def verify_message(public_key, message_bytes: bytes, signature: bytes) -> bool:
        """Verify an ECDSA signature. Returns True if valid, False otherwise."""
        try:
            public_key.verify(signature, message_bytes, ec.ECDSA(hashes.SHA256()))
            return True
        except InvalidSignature:
            return False


class PseudonymCertManager:
    """Manages a pool of pseudonym certificates for privacy-preserving V2X communication."""

    def __init__(self, pool_size: int = 5):
        self.pool_size = pool_size
        self.keypairs = []
        self.current_index = 0
        self._generate_pool()

    def _generate_pool(self):
        """Generate a batch of keypairs (simulating pseudonym certificate provisioning)."""
        for _ in range(self.pool_size):
            private_key, public_key = V2XCryptoEngine.generate_vehicle_keypair()
            self.keypairs.append((private_key, public_key))

    def rotate_certificate(self):
        """Rotate to the next pseudonym certificate in the pool."""
        self.current_index = (self.current_index + 1) % self.pool_size

    def get_current_signing_key(self):
        """Get the current private key for signing."""
        return self.keypairs[self.current_index][0]

    def get_current_public_key(self):
        """Get the current public key for verification."""
        return self.keypairs[self.current_index][1]

    def get_all_public_keys(self):
        """Get all public keys in the pool (for verification by other vehicles)."""
        return [kp[1] for kp in self.keypairs]

Your crypto engine is locked and loaded. Next up, you'll build the vehicle simulation that uses this engine to broadcast signed messages across the network.

Build the Vehicle Simulation

Your crypto engine can sign and verify messages. Now you need something to actually generate those messages.

In this step, you will create virtual vehicles that broadcast signed position messages (called BSMs) and wire them up to a FastAPI web server so you can interact with the simulation through your browser.

In this step, get ready to:
  • Create a Vehicle class that broadcasts cryptographically signed BSMs.
  • Build a V2XNetwork simulation loop that moves vehicles and rotates pseudonym certificates.
  • Wire up FastAPI endpoints to control the simulation from your browser.
Create the vehicle model and message broadcasting

A real vehicle on a V2X network continuously broadcasts its position, speed, and heading in a signed message called a Basic Safety Message (BSM). Your simulation needs to replicate this behaviour.

  • Open app/simulation.py in VS Code and replace its contents with the following imports and Vehicle class:
import json
import time
import random
from app.crypto_engine import V2XCryptoEngine, PseudonymCertManager


class Vehicle:
    """Simulates a vehicle broadcasting signed BSM messages."""

    def __init__(self, vehicle_id: str, x: float, y: float, speed: float, heading: float):
        self.vehicle_id = vehicle_id
        self.x = x
        self.y = y
        self.speed = speed
        self.heading = heading
        # Each vehicle holds its own pool of pseudonym certificates
        self.cert_manager = PseudonymCertManager(pool_size=5)

    def broadcast_bsm(self) -> dict:
        """Create and sign a Basic Safety Message (BSM)."""
        # Package vehicle state into the BSM payload
        bsm_data = {
            "vehicle_id": self.vehicle_id,
            "x": round(self.x, 2),
            "y": round(self.y, 2),
            "speed": round(self.speed, 2),
            "heading": round(self.heading, 2),
            "timestamp": time.time(),
        }
        # Serialize deterministically so signature is reproducible
        message_bytes = json.dumps(bsm_data, sort_keys=True).encode()
        # Sign with the vehicle's current pseudonym key
        signature = V2XCryptoEngine.sign_message(
            self.cert_manager.get_current_signing_key(), message_bytes
        )
        return {
            "bsm": bsm_data,
            "message_bytes": message_bytes,
            "signature": signature,
            "public_key": self.cert_manager.get_current_public_key(),
        }

    def to_dict(self) -> dict:
        """Return vehicle state as a dictionary."""
        return {
            "vehicle_id": self.vehicle_id,
            "x": round(self.x, 2),
            "y": round(self.y, 2),
            "speed": round(self.speed, 2),
            "heading": round(self.heading, 2),
        }

What does this code do?

The Vehicle class models a single connected vehicle. Each vehicle has a position, speed, heading, and its own PseudonymCertManager (from the crypto engine you built in the previous step).

The broadcast_bsm() method packages the vehicle's current state into a JSON payload, serializes it to bytes, then signs it with ECDSA using the vehicle's current pseudonym key. This mirrors how real vehicles sign BSMs under IEEE 1609.2.

  • Below the Vehicle class, add the V2XNetwork class that manages multiple vehicles:

class V2XNetwork:
    """Manages the V2X simulation network."""

    def __init__(self):
        self.vehicles = []
        self.messages = []
        self.step_count = 0
        self._init_vehicles()

    def _init_vehicles(self):
        """Initialize the network with 5 simulated vehicles."""
        for i in range(5):
            vehicle = Vehicle(
                vehicle_id=f"VEH-{i+1:03d}",
                x=random.uniform(0, 500),
                y=random.uniform(0, 500),
                speed=random.uniform(30, 80),
                heading=random.uniform(0, 360),
            )
            self.vehicles.append(vehicle)

    def get_status(self) -> dict:
        """Get current network status."""
        return {
            "step_count": self.step_count,
            "vehicle_count": len(self.vehicles),
            "vehicles": [v.to_dict() for v in self.vehicles],
            "total_messages": len(self.messages),
        }

    def inject_message(self, message: dict):
        """Inject a message into the network (used by attack modules later)."""
        self.messages.append(message)
        if len(self.messages) > 100:
            self.messages = self.messages[-100:]

What does this code do?

The V2XNetwork is the central coordinator. On initialization, it spawns 5 vehicles (VEH-001 through VEH-005) at random positions with random speeds.

The inject_message() method exists so the attack modules (which you will build later) can push fake messages into the network. The message buffer is capped at 100 to prevent unbounded memory growth.

  • Save app/simulation.py.

You will not see a visible change yet. The next substep adds the movement and simulation loop that brings these vehicles to life.

Implement the simulation loop

Vehicles need to move. They also need to periodically rotate their pseudonym certificates for privacy, just like real V2X systems do.

  • In app/simulation.py, add the update_position method to the Vehicle class, between the __init__ method and the broadcast_bsm method:
    def update_position(self):
        """Move the vehicle based on current speed and heading."""
        import math
        rad = math.radians(self.heading)
        # Advance position based on speed and direction
        self.x += self.speed * math.cos(rad) * 0.1
        self.y += self.speed * math.sin(rad) * 0.1
        # Add slight random variation to simulate real driving
        self.heading += random.uniform(-5, 5)
        self.speed = max(0, self.speed + random.uniform(-2, 2))
        self.speed = min(self.speed, 120)

What does this code do?

Each time the simulation advances, vehicles move in the direction of their heading. A small random variation simulates real-world driving behaviour (slight speed and direction changes). The speed is capped at 120 km/h to keep values realistic.

  • Now add the step() method to the V2XNetwork class, between _init_vehicles and get_status:
    def step(self) -> list:
        """Advance the simulation by one step."""
        self.step_count += 1
        step_messages = []
        for vehicle in self.vehicles:
            # Move vehicle to new position
            vehicle.update_position()
            # Broadcast a signed BSM
            msg = vehicle.broadcast_bsm()
            # Verify the signature (proves message authenticity)
            verified = V2XCryptoEngine.verify_message(
                msg["public_key"], msg["message_bytes"], msg["signature"]
            )
            step_messages.append({
                "bsm": msg["bsm"],
                "verified": verified,
                "source": "legitimate",
                "step": self.step_count,
            })
            # Rotate pseudonym certificates every 5 steps for privacy
            if self.step_count % 5 == 0:
                vehicle.cert_manager.rotate_certificate()
        self.messages.extend(step_messages)
        if len(self.messages) > 100:
            self.messages = self.messages[-100:]
        return step_messages

What does this code do?

The step() method is the heart of the simulation. Each call advances every vehicle, broadcasts a signed BSM, and verifies the signature (proving the message is authentic).

Every 5 steps, vehicles rotate their pseudonym certificate. In real V2X, this prevents long-term tracking of individual vehicles while still allowing message authentication.

  • Save app/simulation.py.
  • Verify the simulation works by running this quick test in your terminal:
python -c "from app.simulation import V2XNetwork; n = V2XNetwork(); msgs = n.step(); print(f'Step 1: {len(msgs)} messages, all verified: {all(m[\"verified\"] for m in msgs)}')"

You should see output confirming 5 messages were generated and all verified as True.

Seeing an ImportError?

Make sure you are running the command from the v2x-security-sim project root (the folder containing the app/ directory). Also check that your virtual environment is activated.

If you see a syntax error, double-check that update_position and step are indented at the same level as the other methods in their respective classes.

help me debug this ImportError in my V2X simulation

✔️ Awesome, I've got everything!

Great. Double check you have saved app/simulation.py before continuing.

ⓧ I'd like to double check the full code

import json
import time
import random
from app.crypto_engine import V2XCryptoEngine, PseudonymCertManager


class Vehicle:
    def __init__(self, vehicle_id: str, x: float, y: float, speed: float, heading: float):
        self.vehicle_id = vehicle_id
        self.x = x
        self.y = y
        self.speed = speed
        self.heading = heading
        self.cert_manager = PseudonymCertManager(pool_size=5)

    def update_position(self):
        import math
        rad = math.radians(self.heading)
        self.x += self.speed * math.cos(rad) * 0.1
        self.y += self.speed * math.sin(rad) * 0.1
        self.heading += random.uniform(-5, 5)
        self.speed = max(0, self.speed + random.uniform(-2, 2))
        self.speed = min(self.speed, 120)

    def broadcast_bsm(self) -> dict:
        bsm_data = {
            "vehicle_id": self.vehicle_id,
            "x": round(self.x, 2),
            "y": round(self.y, 2),
            "speed": round(self.speed, 2),
            "heading": round(self.heading, 2),
            "timestamp": time.time(),
        }
        message_bytes = json.dumps(bsm_data, sort_keys=True).encode()
        signature = V2XCryptoEngine.sign_message(
            self.cert_manager.get_current_signing_key(), message_bytes
        )
        return {
            "bsm": bsm_data,
            "message_bytes": message_bytes,
            "signature": signature,
            "public_key": self.cert_manager.get_current_public_key(),
        }

    def to_dict(self) -> dict:
        return {
            "vehicle_id": self.vehicle_id,
            "x": round(self.x, 2),
            "y": round(self.y, 2),
            "speed": round(self.speed, 2),
            "heading": round(self.heading, 2),
        }


class V2XNetwork:
    def __init__(self):
        self.vehicles = []
        self.messages = []
        self.step_count = 0
        self._init_vehicles()

    def _init_vehicles(self):
        for i in range(5):
            vehicle = Vehicle(
                vehicle_id=f"VEH-{i+1:03d}",
                x=random.uniform(0, 500),
                y=random.uniform(0, 500),
                speed=random.uniform(30, 80),
                heading=random.uniform(0, 360),
            )
            self.vehicles.append(vehicle)

    def step(self) -> list:
        self.step_count += 1
        step_messages = []
        for vehicle in self.vehicles:
            vehicle.update_position()
            msg = vehicle.broadcast_bsm()
            verified = V2XCryptoEngine.verify_message(
                msg["public_key"], msg["message_bytes"], msg["signature"]
            )
            step_messages.append({
                "bsm": msg["bsm"],
                "verified": verified,
                "source": "legitimate",
                "step": self.step_count,
            })
            if self.step_count % 5 == 0:
                vehicle.cert_manager.rotate_certificate()
        self.messages.extend(step_messages)
        if len(self.messages) > 100:
            self.messages = self.messages[-100:]
        return step_messages

    def get_status(self) -> dict:
        return {
            "step_count": self.step_count,
            "vehicle_count": len(self.vehicles),
            "vehicles": [v.to_dict() for v in self.vehicles],
            "total_messages": len(self.messages),
        }

    def inject_message(self, message: dict):
        self.messages.append(message)
        if len(self.messages) > 100:
            self.messages = self.messages[-100:]
Wire up FastAPI endpoints

The simulation logic works, but you need a way to interact with it. FastAPI gives you API endpoints that you can call from a browser or a dashboard.

  • Before wiring up the main app, you need placeholder classes in app/attacks.py and app/detector.py so the imports work. Open app/attacks.py and add the following:
class SpoofingAttack:
    """Placeholder - full implementation in next step."""
    pass


class ReplayAttack:
    """Placeholder - full implementation in next step."""

    def capture(self, message):
        """Capture a message for later replay."""
        pass
  • Save app/attacks.py.
  • Open app/detector.py and add this placeholder:
class RuleBasedDetector:
    """Placeholder - full implementation in a later step."""

    def __init__(self):
        self.alerts = []

    def analyze(self, message):
        """Analyze a message. Returns empty list for now."""
        return []

    def get_recent_alerts(self, count=20):
        """Return recent alerts."""
        return []
  • Save app/detector.py.
  • Now open app/main.py and replace its contents with the full FastAPI application:
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from app.simulation import V2XNetwork
from app.attacks import SpoofingAttack, ReplayAttack
from app.detector import RuleBasedDetector

# Create the FastAPI app
app = FastAPI(title="V2X Security Simulator")
templates = Jinja2Templates(directory="app/templates")

# Initialize simulation components
network = V2XNetwork()
detector = RuleBasedDetector()
spoof_attack = SpoofingAttack()
replay_attack = ReplayAttack()


@app.get("/", response_class=HTMLResponse)
async def dashboard(request: Request):
    """Serve the main dashboard."""
    return templates.TemplateResponse(request, "index.html")


@app.get("/status")
async def get_status():
    """Get current network status."""
    return network.get_status()


@app.post("/step")
async def step_simulation():
    """Advance simulation by one step."""
    messages = network.step()
    step_alerts = []
    for msg in messages:
        # Run each message through the detector
        alerts = detector.analyze(msg)
        step_alerts.extend(alerts)
        # Capture verified messages for potential replay attacks later
        if msg["verified"]:
            replay_attack.capture(msg)
    return {"messages": [{k: v for k, v in m.items() if k != "message_bytes"} for m in messages], "alerts": step_alerts}


@app.get("/messages")
async def get_messages():
    """Get recent messages."""
    safe_messages = []
    for m in network.messages[-20:]:
        safe_messages.append({k: v for k, v in m.items() if k not in ("message_bytes", "signature", "public_key")})
    return {"messages": safe_messages}


@app.get("/alerts")
async def get_alerts():
    """Get recent detection alerts."""
    return {"alerts": detector.get_recent_alerts()}

What does this code do?

  • GET / serves the HTML dashboard (you will build this in a later step).
  • GET /status returns the current network state (vehicle positions, message count).
  • POST /step advances the simulation one tick, broadcasts messages, runs detection, and returns results.
  • GET /messages returns the last 20 messages (with non-serializable fields stripped out).
  • GET /alerts returns recent detection alerts (empty for now until you build the detector).

The POST /step endpoint filters out message_bytes from the response because raw bytes are not JSON-serializable.

  • Save app/main.py.
  • Start the FastAPI development server by running:
fastapi dev app/main.py

What does this command do?

The fastapi dev command starts a development server with auto-reload enabled. It detects the app object in your file and serves it at http://127.0.0.1:8000 by default. The interactive API docs are automatically available at http://127.0.0.1:8000/docs.

  • Open your browser and navigate to http://127.0.0.1:8000/docs.
  • Click on the POST /step endpoint, then click Try it out, then click Execute.

You should see a JSON response containing 5 messages, each with a BSM payload and "verified": true.

Server won't start?

If you see an import error mentioning app.attacks or app.detector, make sure you added the placeholder classes to those files and saved them.

If the dashboard endpoint (GET /) returns a template error, that is expected. The index.html template is still empty. The API endpoints (/status, /step, /messages) work independently.

help me debug my FastAPI server startup error

✔️ Awesome, I've got everything!

Great. Double check you have saved app/main.py before continuing.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from app.simulation import V2XNetwork
from app.attacks import SpoofingAttack, ReplayAttack
from app.detector import RuleBasedDetector

app = FastAPI(title="V2X Security Simulator")
templates = Jinja2Templates(directory="app/templates")

network = V2XNetwork()
detector = RuleBasedDetector()
spoof_attack = SpoofingAttack()
replay_attack = ReplayAttack()


@app.get("/", response_class=HTMLResponse)
async def dashboard(request: Request):
    return templates.TemplateResponse(request, "index.html")


@app.get("/status")
async def get_status():
    return network.get_status()


@app.post("/step")
async def step_simulation():
    messages = network.step()
    step_alerts = []
    for msg in messages:
        alerts = detector.analyze(msg)
        step_alerts.extend(alerts)
        if msg["verified"]:
            replay_attack.capture(msg)
    return {"messages": [{k: v for k, v in m.items() if k != "message_bytes"} for m in messages], "alerts": step_alerts}


@app.get("/messages")
async def get_messages():
    safe_messages = []
    for m in network.messages[-20:]:
        safe_messages.append({k: v for k, v in m.items() if k not in ("message_bytes", "signature", "public_key")})
    return {"messages": safe_messages}


@app.get("/alerts")
async def get_alerts():
    return {"alerts": detector.get_recent_alerts()}

Your simulation is alive. Vehicles are moving, signing messages, and verifying each other's signatures through a real API. Next up, you will build the attack modules that try to break this trust.

Implement Attack Simulation

Your simulation network is live and broadcasting signed BSM messages from five vehicles. Now it is time to think like an attacker.

In real V2X networks, two of the most common threats are spoofing (impersonating another vehicle with a fake message) and replay attacks (re-broadcasting a captured legitimate message at a later time). In this step, you will implement both attack types and wire them into your FastAPI endpoints so you can trigger them on demand.

In this step, get ready to:
  • Build a spoofing attack class that forges fake BSMs signed with an attacker-controlled key.
  • Build a replay attack class that captures and re-broadcasts legitimate messages.
  • Add API endpoints to trigger both attacks and verify they are detectable.
Implement the spoofing attack

A spoofing attack happens when an attacker injects a message that claims to come from a legitimate vehicle. The attacker signs the message with their own private key, so when the system checks the signature against the real vehicle's known public key, verification fails.

  • In VS Code, open app/attacks.py and replace any placeholder content with the following code:
import json
import time
import random
from app.crypto_engine import V2XCryptoEngine


class SpoofingAttack:
    """Simulates a spoofing attack: attacker signs a fake BSM with their own key."""

    def __init__(self):
        # Generate the attacker's own keypair (different from any legitimate vehicle)
        self.attacker_private_key, self.attacker_public_key = (
            V2XCryptoEngine.generate_vehicle_keypair()
        )

    def execute(self, target_vehicle_id: str) -> dict:
        """Create a spoofed BSM pretending to be from target_vehicle_id."""
        # Build a fake BSM claiming to be from the target vehicle
        fake_bsm = {
            "vehicle_id": target_vehicle_id,
            "x": round(random.uniform(0, 500), 2),
            "y": round(random.uniform(0, 500), 2),
            "speed": round(random.uniform(0, 150), 2),
            "heading": round(random.uniform(0, 360), 2),
            "timestamp": time.time(),
        }
        # Sign with the attacker's key (won't match the target's known public key)
        message_bytes = json.dumps(fake_bsm, sort_keys=True).encode()
        signature = V2XCryptoEngine.sign_message(self.attacker_private_key, message_bytes)

        return {
            "bsm": fake_bsm,
            "message_bytes": message_bytes,
            "signature": signature,
            "public_key": self.attacker_public_key,
            "source": "spoofed",
            "verified": False,
        }

What does this code do?

The SpoofingAttack class generates its own ECDSA keypair when initialized. This keypair belongs to the attacker, not to any legitimate vehicle on the network.

  • execute() builds a fake BSM using the target vehicle's ID but random position/speed data.
  • The fake message is signed with the attacker's private key. When the system later tries to verify it against the target vehicle's real public key, the signature will not match.
  • verified is hard-coded to False because the message is inherently unverifiable against the legitimate vehicle's credentials.
  • You won't see a change yet. The next chunk adds the replay attack, then you will test both together.
Implement the replay attack

A replay attack captures a legitimate, properly signed message and re-broadcasts it later. Because the original signature is still valid, the system cannot catch it through signature verification alone. Instead, it must check whether the timestamp is stale.

  • In app/attacks.py, add the ReplayAttack class below the SpoofingAttack class:
class ReplayAttack:
    """Simulates a replay attack: re-broadcasts a previously captured legitimate message."""

    def __init__(self):
        # Store captured legitimate messages for later replay
        self.captured_messages = []

    def capture(self, message: dict):
        """Capture a legitimate message for later replay."""
        self.captured_messages.append(message)

    def execute(self) -> dict:
        """Replay a captured message with its original (valid) signature."""
        if not self.captured_messages:
            return None
        # Pick a random captured message to replay
        original = random.choice(self.captured_messages)
        replayed_bsm = original["bsm"].copy()
        return {
            "bsm": replayed_bsm,
            "source": "replayed",
            "original_timestamp": replayed_bsm["timestamp"],
            "replay_time": time.time(),
            "verified": True,
        }

What does this code do?

The ReplayAttack class works in two phases.

  • capture() stores legitimate messages as they flow through the network. Your existing /step endpoint already calls this for every verified message.
  • execute() picks a random captured message and re-broadcasts it. The original timestamp stays unchanged, so the message appears old.
  • verified is True because the signature IS valid. The detection engine must catch this through timestamp freshness checks instead.
  • Save app/attacks.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

import json
import time
import random
from app.crypto_engine import V2XCryptoEngine


class SpoofingAttack:
    def __init__(self):
        self.attacker_private_key, self.attacker_public_key = (
            V2XCryptoEngine.generate_vehicle_keypair()
        )

    def execute(self, target_vehicle_id: str) -> dict:
        fake_bsm = {
            "vehicle_id": target_vehicle_id,
            "x": round(random.uniform(0, 500), 2),
            "y": round(random.uniform(0, 500), 2),
            "speed": round(random.uniform(0, 150), 2),
            "heading": round(random.uniform(0, 360), 2),
            "timestamp": time.time(),
        }
        message_bytes = json.dumps(fake_bsm, sort_keys=True).encode()
        signature = V2XCryptoEngine.sign_message(self.attacker_private_key, message_bytes)
        return {
            "bsm": fake_bsm,
            "message_bytes": message_bytes,
            "signature": signature,
            "public_key": self.attacker_public_key,
            "source": "spoofed",
            "verified": False,
        }


class ReplayAttack:
    def __init__(self):
        self.captured_messages = []

    def capture(self, message: dict):
        self.captured_messages.append(message)

    def execute(self) -> dict:
        if not self.captured_messages:
            return None
        original = random.choice(self.captured_messages)
        replayed_bsm = original["bsm"].copy()
        return {
            "bsm": replayed_bsm,
            "source": "replayed",
            "original_timestamp": replayed_bsm["timestamp"],
            "replay_time": time.time(),
            "verified": True,
        }

Now you need to expose these attacks through your API so they can be triggered from the dashboard.

  • Switch back to app/main.py.
  • Find the step_simulation function. Add the spoofing attack endpoint below it by pasting this code:
@app.post("/attack/spoof")
async def trigger_spoof():
    """Trigger a spoofing attack."""
    # Target the first vehicle in the network
    target_id = network.vehicles[0].vehicle_id
    spoofed = spoof_attack.execute(target_id)
    # Strip non-serializable fields before returning
    result = {k: v for k, v in spoofed.items() if k not in ("message_bytes", "signature", "public_key")}
    alerts = detector.analyze(result)
    network.inject_message(result)
    return {"attack": "spoofing", "result": result, "alerts": alerts}

What does this endpoint do?

  • It targets the first vehicle (VEH-001) and creates a spoofed BSM pretending to be from that vehicle.
  • The spoofed message is run through the detector (which you will build in the next step) and then injected into the network's message pool.
  • Non-serializable fields like raw bytes and key objects are stripped from the response.
  • Add the replay attack endpoint directly below trigger_spoof:
@app.post("/attack/replay")
async def trigger_replay():
    """Trigger a replay attack."""
    replayed = replay_attack.execute()
    if replayed is None:
        return {"attack": "replay", "error": "No messages captured yet. Run a few simulation steps first."}
    alerts = detector.analyze(replayed)
    network.inject_message(replayed)
    return {"attack": "replay", "result": replayed, "alerts": alerts}

What does this endpoint do?

  • It calls replay_attack.execute() to re-broadcast a previously captured legitimate message.
  • If no messages have been captured yet (the simulation has not stepped), it returns a helpful error message instead of crashing.
  • The replayed message carries the original (now stale) timestamp, which the detection engine will flag.
  • Save app/main.py.
  • In your terminal, restart the dev server by running:
fastapi dev app/main.py
  • Open http://127.0.0.1:8000/docs in your browser. You should see the two new endpoints POST /attack/spoof and POST /attack/replay listed in the interactive docs.

Server not starting?

  • Check that app/attacks.py has both SpoofingAttack and ReplayAttack classes. The imports in main.py expect both to exist.
  • Make sure the new endpoints are placed outside any existing function. They should be at the same indentation level as the other @app decorators.
  • If you see an IndentationError, confirm you have a blank line between the end of step_simulation and the new @app.post decorator.

Help me debug my FastAPI server startup error for the V2X attack endpoints.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/main.py before moving on.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from app.simulation import V2XNetwork
from app.attacks import SpoofingAttack, ReplayAttack
from app.detector import RuleBasedDetector

app = FastAPI(title="V2X Security Simulator")
templates = Jinja2Templates(directory="app/templates")

network = V2XNetwork()
detector = RuleBasedDetector()
spoof_attack = SpoofingAttack()
replay_attack = ReplayAttack()


@app.get("/", response_class=HTMLResponse)
async def dashboard(request: Request):
    return templates.TemplateResponse(request, "index.html")


@app.get("/status")
async def get_status():
    return network.get_status()


@app.post("/step")
async def step_simulation():
    messages = network.step()
    step_alerts = []
    for msg in messages:
        alerts = detector.analyze(msg)
        step_alerts.extend(alerts)
        if msg["verified"]:
            replay_attack.capture(msg)
    return {"messages": [{k: v for k, v in m.items() if k != "message_bytes"} for m in messages], "alerts": step_alerts}


@app.post("/attack/spoof")
async def trigger_spoof():
    target_id = network.vehicles[0].vehicle_id
    spoofed = spoof_attack.execute(target_id)
    result = {k: v for k, v in spoofed.items() if k not in ("message_bytes", "signature", "public_key")}
    alerts = detector.analyze(result)
    network.inject_message(result)
    return {"attack": "spoofing", "result": result, "alerts": alerts}


@app.post("/attack/replay")
async def trigger_replay():
    replayed = replay_attack.execute()
    if replayed is None:
        return {"attack": "replay", "error": "No messages captured yet. Run a few simulation steps first."}
    alerts = detector.analyze(replayed)
    network.inject_message(replayed)
    return {"attack": "replay", "result": replayed, "alerts": alerts}


@app.get("/messages")
async def get_messages():
    safe_messages = []
    for m in network.messages[-20:]:
        safe_messages.append({k: v for k, v in m.items() if k not in ("message_bytes", "signature", "public_key")})
    return {"messages": safe_messages}


@app.get("/alerts")
async def get_alerts():
    return {"alerts": detector.get_recent_alerts()}
Verify attacks are detectable

Both attack modules are wired up. Time to confirm they produce the expected signals that a detection engine can catch.

  • In the /docs UI, click on POST /step, then click Try it out, then click Execute. Run it 3 times to build up some captured messages.
  • Click on POST /attack/spoof, then Try it out, then Execute.
  • In the response body, confirm that "source": "spoofed" and "verified": false appear. This proves the spoofed message's signature does not match the target vehicle's key.
  • Click on POST /attack/replay, then Try it out, then Execute.
  • In the response body, confirm that "source": "replayed" appears and that original_timestamp is noticeably older than replay_time. This proves the message is stale.

Why does this matter?

These two signals form the basis of detection. A spoofed message fails signature verification. A replayed message has an outdated timestamp.

In the next step, you will build a detection engine that checks for exactly these conditions and raises alerts automatically.

Replay endpoint returning an error?

If you see "No messages captured yet. Run a few simulation steps first." in the response, run POST /step a few more times before trying the replay again. The replay attack needs captured messages to work with.

Help me troubleshoot my V2X replay attack endpoint.

Both attack vectors are live and producing detectable signals. Next up, you will build the detection engine that automatically catches these attacks and raises alerts in real time.

Build the Detection Engine and Dashboard

Your V2X network can now simulate attacks, but without a detection system, those attacks go unnoticed. Real V2X infrastructure uses layered defenses that analyze every incoming message for signs of tampering.

In this step, you will build a rule-based detection engine that catches spoofing and replay attacks automatically. You will also create an interactive dashboard so you can watch the entire system in action from your browser.

In this step, get ready to:
  • Implement a rule-based detection engine with three security checks.
  • Verify the detector catches both spoofing and replay attacks through the API.
  • Build an HTML dashboard for visual monitoring of the V2X network.
Implement rule-based detection

The detection engine inspects every message that enters the V2X network and checks it against a set of security rules. Each rule targets a specific attack pattern.

  • Open app/detector.py in VS Code and replace any placeholder content with the following code:
import time


class RuleBasedDetector:
    """Rule-based V2X anomaly detection engine."""

    MAX_PLAUSIBLE_SPEED = 200.0
    MAX_MESSAGE_AGE_SECONDS = 3.0

    def __init__(self):
        self.alerts = []

    def analyze(self, message: dict) -> list:
        """Analyze a message and return any detection alerts."""
        new_alerts = []

        # Rule 1: Signature verification failure (catches spoofing)
        if message.get("verified") is False:
            alert = {
                "type": "SIGNATURE_FAILURE",
                "severity": "HIGH",
                "description": f"Invalid signature for vehicle {message['bsm']['vehicle_id']}. Possible spoofing attack.",
                "vehicle_id": message["bsm"]["vehicle_id"],
                "timestamp": time.time(),
            }
            new_alerts.append(alert)

        # Rule 2: Timestamp freshness check (catches replay attacks)
        bsm = message.get("bsm", {})
        msg_timestamp = bsm.get("timestamp", 0)
        age = time.time() - msg_timestamp
        if age > self.MAX_MESSAGE_AGE_SECONDS:
            alert = {
                "type": "REPLAY_DETECTED",
                "severity": "MEDIUM",
                "description": f"Message from {bsm.get('vehicle_id', 'unknown')} is {age:.1f}s old. Possible replay attack.",
                "vehicle_id": bsm.get("vehicle_id", "unknown"),
                "timestamp": time.time(),
            }
            new_alerts.append(alert)

        # Rule 3: Speed plausibility check (catches impossible values)
        speed = bsm.get("speed", 0)
        if speed > self.MAX_PLAUSIBLE_SPEED:
            alert = {
                "type": "IMPLAUSIBLE_SPEED",
                "severity": "MEDIUM",
                "description": f"Vehicle {bsm.get('vehicle_id', 'unknown')} reports speed {speed} km/h (max plausible: {self.MAX_PLAUSIBLE_SPEED}).",
                "vehicle_id": bsm.get("vehicle_id", "unknown"),
                "timestamp": time.time(),
            }
            new_alerts.append(alert)

        # Store alerts and cap the list at 50
        self.alerts.extend(new_alerts)
        if len(self.alerts) > 50:
            self.alerts = self.alerts[-50:]

        return new_alerts

    def get_recent_alerts(self, count: int = 20) -> list:
        """Return the most recent alerts."""
        return self.alerts[-count:]

What does this code do?

The RuleBasedDetector class runs three checks on every message that enters the network:

  • SIGNATURE_FAILURE (HIGH severity): If a message's verified field is False, the signature does not match the expected public key. This catches spoofing attacks.
  • REPLAY_DETECTED (MEDIUM severity): If a message's timestamp is older than 3 seconds, it is likely a replayed copy of an earlier legitimate message.
  • IMPLAUSIBLE_SPEED (MEDIUM severity): If a message claims a speed above 200 km/h, the value is physically implausible and suggests data manipulation.
  • Save app/detector.py.

Why rule-based detection first?

Real V2X systems use rule-based checks as the first line of defense. Signature verification and timestamp freshness are non-negotiable. They are fast, deterministic, and catch the most common attack patterns.

ML-based detection adds a second layer for catching subtle anomalies that rules miss. You will explore this in the Secret Mission.

Seeing an import error?

Make sure the file is saved as app/detector.py (inside the app folder, not in the project root). The class name must be exactly RuleBasedDetector because main.py imports it by that name.

Why is my detector import failing?

Verify the detector catches attacks

Your app/main.py already imports RuleBasedDetector and passes every message through it. Now that the detector has real logic, the entire pipeline is functional.

  • Start the development server by running this command:
fastapi dev app/main.py
  • Open your browser and go to http://127.0.0.1:8000/docs to access the interactive API documentation.
  • Click POST /step, then click Try it out, then Execute. Run this a few times to generate traffic.
  • Click POST /attack/spoof and execute it. You should see a SIGNATURE_FAILURE alert in the response.
  • Click POST /attack/replay and execute it. You should see a REPLAY_DETECTED alert in the response.
  • Click GET /alerts and execute it. You should see all the alerts you just triggered listed together.

Not seeing alerts after an attack?

Make sure you ran at least one POST /step before trying POST /attack/replay. The replay attack needs captured messages to work with.

My V2X detector isn't generating alerts when I trigger attacks. Help me debug.

  • Stop the server by pressing Ctrl+C in your terminal. You will restart it after building the dashboard.
Build the HTML dashboard

The /docs interface is great for testing, but a visual dashboard makes the simulation much easier to operate and understand. You will build a single-page interface with controls, vehicle status, messages, and alert panels.

  • Open app/templates/index.html in VS Code and replace any placeholder content with the following code:
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>V2X Security Simulator</title>
    <style>
        /* Base styles for the dark theme dashboard */
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: system-ui, sans-serif; background: #1a1a2e; color: #eee; padding: 20px; }
        h1 { text-align: center; margin-bottom: 20px; color: #00d4ff; }
        .grid { display: grid; grid-template-columns: 1fr 1fr; gap: 20px; max-width: 1200px; margin: 0 auto; }
        .panel { background: #16213e; border-radius: 8px; padding: 16px; border: 1px solid #0f3460; }
        .panel h2 { color: #00d4ff; margin-bottom: 12px; font-size: 1.1em; }
        /* Button styles: blue for controls, red for attacks */
        button { background: #0f3460; color: #00d4ff; border: 1px solid #00d4ff; padding: 8px 16px; border-radius: 4px; cursor: pointer; margin: 4px; font-size: 0.9em; }
        button:hover { background: #00d4ff; color: #1a1a2e; }
        button.attack { border-color: #ff4757; color: #ff4757; }
        button.attack:hover { background: #ff4757; color: #fff; }
        /* Scrollable panels for data lists */
        .vehicle-list, .message-list, .alert-list { max-height: 300px; overflow-y: auto; font-size: 0.85em; }
        .vehicle-item { padding: 6px; border-bottom: 1px solid #0f3460; }
        .alert-item { padding: 6px; border-left: 3px solid #ff4757; margin-bottom: 4px; background: #1a1a2e; }
        .alert-item.normal { border-left-color: #2ed573; }
        /* Color-coded messages by source type */
        .message-item { padding: 4px; border-bottom: 1px solid #0f3460; }
        .message-item.spoofed { color: #ff4757; }
        .message-item.replayed { color: #ffa502; }
        .message-item.legitimate { color: #2ed573; }
        .status-bar { text-align: center; margin-bottom: 16px; color: #aaa; font-size: 0.9em; }
    </style>
</head>
<body>
    <h1>V2X Security Simulator</h1>
    <div class="status-bar" id="status-bar">Step: 0 | Vehicles: 0 | Messages: 0</div>
    <div class="grid">
        <!-- Controls panel with simulation and attack buttons -->
        <div class="panel">
            <h2>Controls</h2>
            <button onclick="stepSim()">Step Simulation</button>
            <button onclick="autoRun()">Auto Run (10 steps)</button>
            <br><br>
            <h2>Attacks</h2>
            <button class="attack" onclick="triggerSpoof()">Spoofing Attack</button>
            <button class="attack" onclick="triggerReplay()">Replay Attack</button>
        </div>
        <!-- Vehicle status panel -->
        <div class="panel">
            <h2>Vehicles</h2>
            <div class="vehicle-list" id="vehicle-list">No data yet. Click "Step Simulation".</div>
        </div>
        <!-- Messages panel showing all network traffic -->
        <div class="panel">
            <h2>Messages</h2>
            <div class="message-list" id="message-list">No messages yet.</div>
        </div>
        <!-- Detection alerts panel -->
        <div class="panel">
            <h2>Detection Alerts</h2>
            <div class="alert-list" id="alert-list">No alerts.</div>
        </div>
    </div>
    <script>
        // Advance the simulation by one step and display results
        async function stepSim() {
            const res = await fetch('/step', {method: 'POST'});
            const data = await res.json();
            updateMessages(data.messages);
            if (data.alerts.length > 0) updateAlerts(data.alerts);
            refreshStatus();
        }

        // Run 10 simulation steps automatically with a short delay between each
        async function autoRun() {
            for (let i = 0; i < 10; i++) {
                await stepSim();
                await new Promise(r => setTimeout(r, 300));
            }
        }

        // Trigger a spoofing attack and display any resulting alerts
        async function triggerSpoof() {
            const res = await fetch('/attack/spoof', {method: 'POST'});
            const data = await res.json();
            if (data.alerts) updateAlerts(data.alerts);
            refreshStatus();
        }

        // Trigger a replay attack and display any resulting alerts
        async function triggerReplay() {
            const res = await fetch('/attack/replay', {method: 'POST'});
            const data = await res.json();
            if (data.alerts) updateAlerts(data.alerts);
            refreshStatus();
        }

        // Refresh the status bar and vehicle list from the server
        async function refreshStatus() {
            const res = await fetch('/status');
            const data = await res.json();
            document.getElementById('status-bar').textContent =
                `Step: ${data.step_count} | Vehicles: ${data.vehicle_count} | Messages: ${data.total_messages}`;
            const vList = document.getElementById('vehicle-list');
            vList.innerHTML = data.vehicles.map(v =>
                `<div class="vehicle-item">${v.vehicle_id} | Pos: (${v.x}, ${v.y}) | Speed: ${v.speed} km/h</div>`
            ).join('');
        }

        // Prepend new messages to the messages panel with color coding
        function updateMessages(messages) {
            const mList = document.getElementById('message-list');
            const html = messages.map(m =>
                `<div class="message-item ${m.source}">[${m.source.toUpperCase()}] ${m.bsm.vehicle_id} | Speed: ${m.bsm.speed} | Verified: ${m.verified}</div>`
            ).join('');
            mList.innerHTML = html + mList.innerHTML;
        }

        // Prepend new alerts to the alerts panel
        function updateAlerts(alerts) {
            const aList = document.getElementById('alert-list');
            const html = alerts.map(a =>
                `<div class="alert-item">[${a.severity}] ${a.type}: ${a.description}</div>`
            ).join('');
            aList.innerHTML = html + aList.innerHTML;
        }
    </script>
</body>
</html>

What does this dashboard do?

The dashboard is a single HTML page with four panels arranged in a 2x2 grid:

  • Controls: Blue buttons advance the simulation. Red buttons trigger attacks.
  • Vehicles: Shows each vehicle's current position and speed, updated after every step.
  • Messages: Color-coded traffic feed. Green for legitimate, red for spoofed, orange for replayed.
  • Detection Alerts: Shows alerts from the rule-based detector with severity and description.
  • Save app/templates/index.html.
  • Restart the development server by running:
fastapi dev app/main.py
  • Open your browser and go to http://127.0.0.1:8000/ to see the dashboard.
  • Click Step Simulation a few times. You should see green messages appear in the Messages panel and vehicle positions update in the Vehicles panel.
  • Click Spoofing Attack. You should see a red SIGNATURE_FAILURE alert appear in the Detection Alerts panel.
  • Click Replay Attack. You should see a REPLAY_DETECTED alert appear.

Dashboard showing a blank page?

Check that your file is saved at app/templates/index.html (note the templates folder inside app). FastAPI looks for templates in the directory specified by Jinja2Templates(directory="app/templates") in your main.py.

My V2X dashboard is blank or not loading. Help me figure out why.

✔️ Awesome, I've got everything!

Great. Double check you have saved both app/detector.py and app/templates/index.html.

ⓧ I'd like to double check the full code

Here is the complete app/detector.py:

import time


class RuleBasedDetector:
    MAX_PLAUSIBLE_SPEED = 200.0
    MAX_MESSAGE_AGE_SECONDS = 3.0

    def __init__(self):
        self.alerts = []

    def analyze(self, message: dict) -> list:
        new_alerts = []
        if message.get("verified") is False:
            alert = {
                "type": "SIGNATURE_FAILURE",
                "severity": "HIGH",
                "description": f"Invalid signature for vehicle {message['bsm']['vehicle_id']}. Possible spoofing attack.",
                "vehicle_id": message["bsm"]["vehicle_id"],
                "timestamp": time.time(),
            }
            new_alerts.append(alert)
        bsm = message.get("bsm", {})
        msg_timestamp = bsm.get("timestamp", 0)
        age = time.time() - msg_timestamp
        if age > self.MAX_MESSAGE_AGE_SECONDS:
            alert = {
                "type": "REPLAY_DETECTED",
                "severity": "MEDIUM",
                "description": f"Message from {bsm.get('vehicle_id', 'unknown')} is {age:.1f}s old. Possible replay attack.",
                "vehicle_id": bsm.get("vehicle_id", "unknown"),
                "timestamp": time.time(),
            }
            new_alerts.append(alert)
        speed = bsm.get("speed", 0)
        if speed > self.MAX_PLAUSIBLE_SPEED:
            alert = {
                "type": "IMPLAUSIBLE_SPEED",
                "severity": "MEDIUM",
                "description": f"Vehicle {bsm.get('vehicle_id', 'unknown')} reports speed {speed} km/h (max plausible: {self.MAX_PLAUSIBLE_SPEED}).",
                "vehicle_id": bsm.get("vehicle_id", "unknown"),
                "timestamp": time.time(),
            }
            new_alerts.append(alert)
        self.alerts.extend(new_alerts)
        if len(self.alerts) > 50:
            self.alerts = self.alerts[-50:]
        return new_alerts

    def get_recent_alerts(self, count: int = 20) -> list:
        return self.alerts[-count:]

Here is the complete app/templates/index.html:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>V2X Security Simulator</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: system-ui, sans-serif; background: #1a1a2e; color: #eee; padding: 20px; }
        h1 { text-align: center; margin-bottom: 20px; color: #00d4ff; }
        .grid { display: grid; grid-template-columns: 1fr 1fr; gap: 20px; max-width: 1200px; margin: 0 auto; }
        .panel { background: #16213e; border-radius: 8px; padding: 16px; border: 1px solid #0f3460; }
        .panel h2 { color: #00d4ff; margin-bottom: 12px; font-size: 1.1em; }
        button { background: #0f3460; color: #00d4ff; border: 1px solid #00d4ff; padding: 8px 16px; border-radius: 4px; cursor: pointer; margin: 4px; font-size: 0.9em; }
        button:hover { background: #00d4ff; color: #1a1a2e; }
        button.attack { border-color: #ff4757; color: #ff4757; }
        button.attack:hover { background: #ff4757; color: #fff; }
        .vehicle-list, .message-list, .alert-list { max-height: 300px; overflow-y: auto; font-size: 0.85em; }
        .vehicle-item { padding: 6px; border-bottom: 1px solid #0f3460; }
        .alert-item { padding: 6px; border-left: 3px solid #ff4757; margin-bottom: 4px; background: #1a1a2e; }
        .alert-item.normal { border-left-color: #2ed573; }
        .message-item { padding: 4px; border-bottom: 1px solid #0f3460; }
        .message-item.spoofed { color: #ff4757; }
        .message-item.replayed { color: #ffa502; }
        .message-item.legitimate { color: #2ed573; }
        .status-bar { text-align: center; margin-bottom: 16px; color: #aaa; font-size: 0.9em; }
    </style>
</head>
<body>
    <h1>V2X Security Simulator</h1>
    <div class="status-bar" id="status-bar">Step: 0 | Vehicles: 0 | Messages: 0</div>
    <div class="grid">
        <div class="panel">
            <h2>Controls</h2>
            <button onclick="stepSim()">Step Simulation</button>
            <button onclick="autoRun()">Auto Run (10 steps)</button>
            <br><br>
            <h2>Attacks</h2>
            <button class="attack" onclick="triggerSpoof()">Spoofing Attack</button>
            <button class="attack" onclick="triggerReplay()">Replay Attack</button>
        </div>
        <div class="panel">
            <h2>Vehicles</h2>
            <div class="vehicle-list" id="vehicle-list">No data yet. Click "Step Simulation".</div>
        </div>
        <div class="panel">
            <h2>Messages</h2>
            <div class="message-list" id="message-list">No messages yet.</div>
        </div>
        <div class="panel">
            <h2>Detection Alerts</h2>
            <div class="alert-list" id="alert-list">No alerts.</div>
        </div>
    </div>
    <script>
        async function stepSim() {
            const res = await fetch('/step', {method: 'POST'});
            const data = await res.json();
            updateMessages(data.messages);
            if (data.alerts.length > 0) updateAlerts(data.alerts);
            refreshStatus();
        }
        async function autoRun() {
            for (let i = 0; i < 10; i++) {
                await stepSim();
                await new Promise(r => setTimeout(r, 300));
            }
        }
        async function triggerSpoof() {
            const res = await fetch('/attack/spoof', {method: 'POST'});
            const data = await res.json();
            if (data.alerts) updateAlerts(data.alerts);
            refreshStatus();
        }
        async function triggerReplay() {
            const res = await fetch('/attack/replay', {method: 'POST'});
            const data = await res.json();
            if (data.alerts) updateAlerts(data.alerts);
            refreshStatus();
        }
        async function refreshStatus() {
            const res = await fetch('/status');
            const data = await res.json();
            document.getElementById('status-bar').textContent =
                `Step: ${data.step_count} | Vehicles: ${data.vehicle_count} | Messages: ${data.total_messages}`;
            const vList = document.getElementById('vehicle-list');
            vList.innerHTML = data.vehicles.map(v =>
                `<div class="vehicle-item">${v.vehicle_id} | Pos: (${v.x}, ${v.y}) | Speed: ${v.speed} km/h</div>`
            ).join('');
        }
        function updateMessages(messages) {
            const mList = document.getElementById('message-list');
            const html = messages.map(m =>
                `<div class="message-item ${m.source}">[${m.source.toUpperCase()}] ${m.bsm.vehicle_id} | Speed: ${m.bsm.speed} | Verified: ${m.verified}</div>`
            ).join('');
            mList.innerHTML = html + mList.innerHTML;
        }
        function updateAlerts(alerts) {
            const aList = document.getElementById('alert-list');
            const html = alerts.map(a =>
                `<div class="alert-item">[${a.severity}] ${a.type}: ${a.description}</div>`
            ).join('');
            aList.innerHTML = html + aList.innerHTML;
        }
    </script>
</body>
</html>

Your detection engine is live and your dashboard gives you full visibility into the V2X network. Next up, you will deploy the entire application to the internet so anyone can access your simulator through a public URL.

Deploy to Render

Your V2X security simulator works locally. Now it is time to make it accessible to anyone with a browser.

In this step, you will deploy your app to Render, a cloud hosting platform with a free tier that requires no credit card. Once deployed, your simulator will have a public URL you can share.

In this step, get ready to:
  • Prepare your project for deployment and push to GitHub.
  • Create a free Render Web Service connected to your GitHub repo.
  • Verify the live deployment works end-to-end.
Prepare for deployment

Render needs to know how to install your dependencies and start your app. Your requirements.txt already lists everything needed. You just need to confirm it is up to date and push your latest code to GitHub.

  • Switch back to VS Code and open requirements.txt.
  • Confirm it contains these two lines:
fastapi[standard]==0.139.0
cryptography==49.0.0

Why these pinned versions matter for deployment

Pinning exact versions ensures Render installs the same packages you tested locally. Without pins, a newer release could introduce breaking changes on your deployed app.

  • Commit and push all your changes to GitHub by running these commands in the terminal:
git add .
git commit -m "Ready for Render deployment"
git push origin main
  • Verify: open your GitHub repo in the browser. You should see all your project files including app/main.py, app/templates/index.html, and requirements.txt with your latest commit message.

Push rejected or authentication error?

Make sure your GitHub remote is set correctly. Run git remote -v to confirm the remote URL points to your v2x-security-sim repo.

If you see a credentials error, you may need to authenticate with a personal access token or SSH key depending on your GitHub setup.

Help me fix my git push authentication error.

Deploy on Render

Render connects directly to your GitHub repository. Every time you push new code, it can automatically rebuild and redeploy your app.

  • Open your browser and navigate to render.com.
  • Click Get Started to create a free account. No credit card is required.
  • Sign up using your GitHub account. This makes connecting your repo easier in the next steps.

Why sign up with GitHub?

Signing up with GitHub automatically links your repositories to Render. This means you can select your repo directly during service creation without extra authorization steps.

  • In the Render Dashboard, click + New in the top-right corner.
  • Select Web Service from the dropdown menu.
  • Under the source options, select Git Provider and connect your GitHub account if prompted.
  • Find your v2x-security-sim repository in the list and click Connect.

The service creation form appears. Fill in the following fields:

  • Set Name to v2x-security-sim.
  • Set Language to Python.
  • Set Build Command to:
pip install -r requirements.txt
  • Set Start Command to:
fastapi run app/main.py --port 10000
  • Scroll down to the instance type selector and choose Free.
  • Click Create Web Service.

Why port 10000?

Render's default port for web services is 10000. Your start command must bind to this port so Render can route incoming traffic to your app.

Render now builds your app. You can watch the progress in the Events log on your service page. The first build typically takes 1-2 minutes.

  • Wait for the build to complete. You will see a green Live badge appear at the top of your service page when it is ready.

Build failed?

Check the build logs in the Render Dashboard. The most common issue is a typo in the build or start command.

Make sure your Build Command is exactly pip install -r requirements.txt and your Start Command is exactly fastapi run app/main.py --port 10000.

Also confirm that requirements.txt was pushed to GitHub. If it is missing from the repo, the build will fail.

Help me debug my Render build failure for my FastAPI Python app.

Verify the live deployment

Your app is now live on the internet. Time to confirm everything works end-to-end on the public URL.

  • Find your public URL at the top of your Render service page. It looks something like https://v2x-security-sim.onrender.com.
  • Click the URL to open your dashboard in a new tab.
  • Click the Step Simulation button a few times. You should see vehicles appear in the Vehicles panel and green messages in the Messages panel.
  • Click the Spoofing Attack button. You should see a red alert appear in the Detection Alerts panel with a SIGNATURE_FAILURE type.
  • Click the Replay Attack button. You should see a REPLAY_DETECTED alert appear.

What about the app sleeping?

On the free tier, your app spins down after 15 minutes of no inbound traffic. When someone visits the URL after that, it takes about 1 minute to wake back up. This is normal and expected for a demo app.

Since the app stores simulation state in memory, the state resets each time the app wakes up. This is fine for a simulation demo.

Dashboard loads but buttons do nothing?

Open your browser's developer console (press F12 or Cmd+Option+I on macOS) and check for network errors. The most common issue is the app still starting up after sleeping.

Wait about 60 seconds and try again. If the issue persists, check the Render Dashboard logs for runtime errors.

Help me debug my V2X simulator dashboard not responding on Render.

Your V2X security simulator is live. Anyone with the link can now see real-time vehicle message authentication, attack simulation, and threat detection in action.

Secret mission

Add ML-Based Anomaly Detection

Your rule-based detector catches attacks with obvious signatures. But what about an attacker who steals a legitimate vehicle's private key? The signature verifies correctly, but the movement patterns are wrong. Can you build an ML model that learns normal V2X traffic and flags the anomalies your rules miss?

Clean Up Your Resources

Clean Up Your Resources

This project runs on Render's free tier, so there are no ongoing costs. Your app sleeps after 15 minutes of inactivity and wakes up on the next request.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Render free web service (your V2X Security Simulator deployment).
  • GitHub repository (v2x-security-sim).
  • Local project folder (v2x-security-sim) with Python virtual environment.

Keep everything running

No action needed. Choose this if you are still actively building or want to keep your simulator accessible.

  • Your Render free tier costs nothing. You get 750 free instance hours per workspace per calendar month.
  • Your app will sleep when unused and wake on the next request (takes about 1 minute).
  • Your GitHub repository remains free and accessible.
  • Your local v2x-security-sim folder stays on your Desktop.

Pause - I'll come back to this later

Suspend your Render service to free up your instance hours. Your code and configuration are preserved.

  • Open the Render Dashboard at dashboard.render.com.
  • Click the checkbox next to your V2X Security Simulator service.
  • Click Suspend in the action menu that appears.
  • When you are ready to resume, select the service again and click Resume in the action menu.

Why suspend on free tier?

Free tier services auto-sleep after 15 minutes of inactivity, so suspending is optional. However, suspending prevents the service from waking up on accidental requests and consuming your 750 free instance hours.

Delete - I don't want to use this again

Remove all project resources and start fresh. This is permanent.

Delete your Render web service:

  • Open the Render Dashboard at dashboard.render.com.
  • Click on your V2X Security Simulator service to open its details.
  • Click Settings in the left pane.
  • Scroll to the bottom and click Delete Web Service.
  • Type the service name to confirm, then click Delete.

Delete your local project folder:

  • Remove the project directory by running these commands in your terminal:
cd ~/Desktop
rm -rf v2x-security-sim

Windows users

If you are on Windows, use this command instead:

cd ~\Desktop
Remove-Item -Recurse -Force v2x-security-sim

Delete your GitHub repository (optional):

  • Go to your v2x-security-sim repository on GitHub.
  • Click Settings in the top navigation.
  • Scroll to the bottom of the page to the Danger Zone section.
  • Click Delete this repository and follow the confirmation prompts.

Nice Work!

Nice Work!

Nice work! You just built and deployed a full V2X security simulator that uses real-world cryptography to authenticate vehicle messages, simulates cyberattacks against the network, and detects those attacks in real-time.

What you learned:

  • Built a V2X security simulation with real ECDSA cryptography (SECP256R1 + SHA-256), the same algorithm mandated by IEEE 1609.2 for vehicle communications, including pseudonym certificate rotation for privacy-preserving authentication.
  • Created spoofing and replay attack modules and a rule-based detection engine that catches signature failures, stale timestamps, and implausible speeds.
  • Deployed a live FastAPI web application with an interactive dashboard to Render's free tier, accessible via a public URL.
  • Secret Mission: Added an Isolation Forest ML anomaly detector that learns normal V2X traffic patterns and catches attacks the rule-based system misses.

Ready to quiz yourself?