Stopping Ghost Commands: How to Detect Replay Attacks in Life-Critical IoT Devices Using Python

Tuhin Paul
2h
61
0
0

Article

Table of Contents

Introduction
What Is a Replay Attack in IoT?
Real-World Scenario: Smart Insulin Pumps in Healthcare
How to Detect Replay Attacks in Command Streams
Complete Implementation with Test Cases
Best Practices and Performance Tips
Conclusion

Introduction

In the rapidly expanding world of the Internet of Things (IoT), security is no longer optional—it’s existential. Among the most insidious threats is the replay attack, where an adversary intercepts a legitimate command and retransmits it later to trigger unintended actions. Unlike brute-force or spoofing attacks, replay attacks exploit the validity of real messages, making them especially dangerous in safety-critical systems.

This article explores how to detect replay attacks in IoT command streams using practical, real-time techniques—illustrated through a life-or-death scenario from modern healthcare.

What Is a Replay Attack in IoT?

A replay attack occurs when an attacker captures a valid data transmission (e.g., “dispense 5 units of insulin”) and maliciously repeats it at a later time. Since the command is authentic and properly formatted, naive systems accept it as legitimate—potentially causing overdose, device malfunction, or system compromise.

Key characteristics:

No decryption or key cracking needed
Works even on encrypted channels if messages aren’t time-bound
Particularly devastating in medical, industrial, and automotive IoT

Real-World Scenario: Smart Insulin Pumps in Healthcare

Imagine a diabetic patient using a Bluetooth-connected insulin pump. Their doctor sends a one-time command via a secure app: “Administer 3 units now.”

An eavesdropper nearby logs this encrypted command. Hours later—when the patient’s blood sugar is already low—they replay the same command. The pump, trusting the valid signature, delivers another dose. The result? Severe hypoglycemia, hospitalization, or worse.

This isn’t hypothetical. In 2011, researchers demonstrated exactly this vulnerability in commercial insulin pumps. Today, with millions of connected medical devices, replay protection is non-negotiable.

How to Detect Replay Attacks in Command Streams

The core defense: ensure every command is fresh and unique. Two proven techniques:

Timestamps with Tolerance Windows
Each command includes a timestamp. The device rejects commands outside a narrow time window (e.g., ±5 seconds).
Monotonic Nonces (Sequence Numbers)
Each command carries an incrementing counter. The device tracks the last accepted nonce and rejects duplicates or regressions.

We combine both for robustness:

import time
from collections import defaultdict

class ReplayDetector:
    def __init__(self, time_window_sec=5, max_devices=1000):
        self.time_window = time_window_sec
        self.last_nonce = defaultdict(int)  # device_id -> last nonce
        self.seen_commands = set()          # Optional: for short-term dedup

    def is_replay(self, device_id: str, nonce: int, timestamp: float, command_hash: str) -> bool:
        current_time = time.time()
        
        # 1. Check timestamp freshness
        if abs(current_time - timestamp) > self.time_window:
            return True  # Too old or from the future

        # 2. Check nonce monotonicity
        if nonce <= self.last_nonce[device_id]:
            return True  # Replayed or out-of-order

        # 3. (Optional) Check exact duplicate via hash
        if command_hash in self.seen_commands:
            return True

        # Update state
        self.last_nonce[device_id] = nonce
        self.seen_commands.add(command_hash)
        
        # Clean old hashes periodically (simplified)
        if len(self.seen_commands) > 10000:
            self.seen_commands.clear()  # In production, use TTL cache

        return False

This class runs on the IoT gateway or device firmware and validates every incoming command before execution.

Complete Implementation with Test Cases

import time
from collections import defaultdict
import hashlib
import sys # Used for interactive mode exit

# --- Replay Detector Class ---

class ReplayDetector:
    """
    Detects potential replay attacks based on timestamp freshness,
    nonce monotonicity, and command hash deduplication.
    """
    def __init__(self, time_window_sec=5, max_devices=1000):
        # Time window for considering a command fresh (in seconds)
        self.time_window = time_window_sec
        # Stores the highest seen nonce for each device_id
        self.last_nonce = defaultdict(int)  # device_id -> last nonce
        # Stores hashes of recent commands for short-term exact deduplication
        self.seen_commands = set()          # command_hash -> None (set is faster)
        # max_devices is currently unused but can be for scaling/cleanup logic

    def is_replay(self, device_id: str, nonce: int, timestamp: float, command_hash: str) -> bool:
        """
        Checks if the given command is a potential replay.

        Returns:
            True if it is a replay, False otherwise.
        """
        current_time = time.time()
        
        # 1. Check timestamp freshness
        # The command must not be too old OR significantly in the future.
        if abs(current_time - timestamp) > self.time_window:
            print(f"   [REPLAY] Fails Timestamp Check. Age: {abs(current_time - timestamp):.2f}s (Max: {self.time_window}s)")
            return True  # Too old or from the future

        # 2. Check nonce monotonicity
        # Nonce must be strictly greater than the last one seen for this device.
        if nonce <= self.last_nonce[device_id]:
            print(f"   [REPLAY] Fails Nonce Check. Nonce: {nonce} <= Last Nonce: {self.last_nonce[device_id]}")
            return True  # Replayed or out-of-order

        # 3. (Optional) Check exact duplicate via hash
        # Checks if the exact command hash has been seen recently, which could
        # catch a unique command being immediately replayed with a new nonce.
        if command_hash in self.seen_commands:
            print(f"   [REPLAY] Fails Hash Check. Command hash already seen.")
            return True

        # --- Update state (ONLY if NOT a replay) ---
        
        self.last_nonce[device_id] = nonce
        self.seen_commands.add(command_hash)
        
        # Clean old hashes periodically (simplified for a demonstration)
        # In a real system, you'd use a TTL (Time-To-Live) cache like Redis
        # or an LRU (Least Recently Used) cache for better performance.
        if len(self.seen_commands) > 10000:
            # Simple cleanup - clear everything.
            self.seen_commands.clear()
            # Note: This is a weak cleanup. A real system would need a time-based eviction.

        return False

# --- Helper Function ---

def hash_command(cmd: str) -> str:
    """Generates a SHA-256 hash for a given command string."""
    return hashlib.sha256(cmd.encode()).hexdigest()

# --- Interactive Demonstration ---

def interactive_demo():
    """Runs a simple interactive demonstration of the ReplayDetector."""
    
    # Instantiate the detector with a 3-second window
    detector = ReplayDetector(time_window_sec=3) 
    DEVICE_ID = "Pump-001"
    
    print("--- Replay Detector Interactive Demo ---")
    print(f"Device ID: {DEVICE_ID} | Time Window: {detector.time_window}s")
    print("Type commands (e.g., INSULIN:3U), 'wait', or 'exit'.")
    print("-" * 40)
    
    nonce = 0
    
    while True:
        try:
            # Prompt for a command
            user_input = input(f"\n[{DEVICE_ID} Nonce {nonce+1}] Enter Command: ").strip()

            if user_input.lower() in ('exit', 'quit'):
                print("Exiting demo.")
                sys.exit(0)
            
            if user_input.lower() == 'wait':
                print(f"-> Waiting for {detector.time_window + 1} seconds to simulate old timestamp...")
                time.sleep(detector.time_window + 1)
                continue

            command = user_input
            current_time = time.time()
            command_hash = hash_command(command)
            
            # Nonce is incremented for the NEW command (if valid)
            test_nonce = nonce + 1 
            
            print(f"-> Sending: '{command}' | Nonce: {test_nonce} | Timestamp: {current_time:.2f}")

            # Run the detection
            is_replay = detector.is_replay(DEVICE_ID, test_nonce, current_time, command_hash)
            
            if is_replay:
                print(" **COMMAND REJECTED**: DETECTED AS REPLAY 🚨")
            else:
                print(" **COMMAND ACCEPTED**: New, Valid Command.")
                # If accepted, we officially advance the nonce
                nonce = test_nonce
                
        except KeyboardInterrupt:
            print("\nExiting demo.")
            sys.exit(0)
        except Exception as e:
            print(f"\nAn error occurred: {e}")
            sys.exit(1)

if __name__ == "__main__":
    interactive_demo()

Best Practices and Performance Tips

Use hardware-backed secure clocks to prevent timestamp manipulation.
Reset nonces securely during device pairing—never reuse across sessions.
Limit seen_commands memory with a time-to-live (TTL) cache (e.g., cachetools.TTLCache).
Log and alert on replay attempts—this is often the first sign of active tampering.
Combine with mutual TLS so only authenticated devices can send commands.

For resource-constrained devices, skip the hash dedup and rely on nonces + timestamps—they’re lightweight and sufficient in most cases.

Conclusion

Replay attacks turn legitimate commands into weapons. In domains like healthcare IoT, the stakes are human lives. By embedding simple, stateful validation—timestamps, nonces, and freshness checks—we can neutralize this threat without heavy cryptography. The insulin pump example reminds us: security in IoT isn’t about preventing “hacking” in the abstract. It’s about ensuring that when a device acts, it acts only when and how it should. With the right design, replay attacks become not just detectable—but impossible to execute successfully.