IoT Security — Hardening and Best Practices

IoT security fails not because the technology doesn't exist, but because it's skipped under time and cost pressure. This topic builds the security mindset and gives you concrete, implementable controls.

OWASP IoT Top 10

The Open Web Application Security Project maintains a specific top 10 for IoT:

#	Vulnerability	Real Example
I1	Weak/default passwords	Mirai botnet: "admin/admin" on 600K cameras
I2	Insecure network services	Telnet open on port 23
I3	Insecure ecosystem interfaces	Unencrypted MQTT on port 1883
I4	Lack of secure update mechanism	Firmware cannot be updated without physical access
I5	Use of insecure or outdated components	LibSSL 2014, unpatched kernel
I6	Insufficient privacy protection	Sensor data sold/leaked without consent
I7	Insecure data transfer and storage	Passwords in plaintext logs
I8	Lack of device management	No way to revoke compromised device
I9	Insecure default settings	Debug mode on, all ports open
I10	Lack of physical hardening	JTAG/UART exposed, SD card removable

Security Audit Script

python

# security_audit.py — automated IoT device security check
import socket
import ssl
import json
import re

class IoTSecurityAudit:
    def __init__(self, device_ip, mqtt_port=1883, https_port=443):
        self.ip         = device_ip
        self.mqtt_port  = mqtt_port
        self.https_port = https_port
        self.findings   = []
        self.score      = 100  # Start at 100, deduct for issues

    def check_open_ports(self):
        """Scan common dangerous ports."""
        dangerous_ports = {
            21: "FTP",
            23: "Telnet",
            80: "HTTP (unencrypted)",
            5900: "VNC",
            8080: "HTTP Alternate",
        }
        open_dangerous = []
        for port, name in dangerous_ports.items():
            try:
                sock = socket.create_connection((self.ip, port), timeout=2)
                sock.close()
                open_dangerous.append(f"{port}/{name}")
            except (socket.timeout, ConnectionRefusedError):
                pass

        if open_dangerous:
            self.findings.append({
                "severity": "HIGH",
                "issue":    f"Dangerous ports open: {', '.join(open_dangerous)}",
                "fix":      "Disable unused services (Telnet, FTP, VNC)"
            })
            self.score -= 20

    def check_mqtt_auth(self):
        """Test if MQTT broker requires authentication."""
        try:
            sock = socket.create_connection((self.ip, self.mqtt_port), timeout=3)
            # Send MQTT CONNECT packet without credentials (anonymous)
            connect_packet = bytes([
                0x10, 0x0E,                          # Fixed header: CONNECT
                0x00, 0x04, 0x4D, 0x51, 0x54, 0x54,  # Protocol name: MQTT
                0x04,                                  # Protocol level 3.1.1
                0x00,                                  # Connect flags: no credentials
                0x00, 0x3C,                            # Keepalive: 60s
                0x00, 0x02, 0x41, 0x42                 # Client ID: "AB"
            ])
            sock.send(connect_packet)
            response = sock.recv(4)
            sock.close()

            # CONNACK return code 0x00 = accepted (no auth required!)
            if len(response) >= 4 and response[3] == 0x00:
                self.findings.append({
                    "severity": "CRITICAL",
                    "issue":    f"MQTT broker at port {self.mqtt_port} allows anonymous access",
                    "fix":      "Enable MQTT authentication and disable anonymous access"
                })
                self.score -= 30
        except:
            pass  # Port closed or connection refused = better than open anonymous

    def check_tls(self):
        """Verify TLS is configured and certificates are valid."""
        try:
            context = ssl.create_default_context()
            with socket.create_connection((self.ip, self.https_port), timeout=3) as sock:
                with context.wrap_socket(sock, server_hostname=self.ip) as ssock:
                    cert = ssock.getpeercert()
                    # Check certificate expiry
                    import datetime
                    not_after = datetime.datetime.strptime(
                        cert['notAfter'], '%b %d %H:%M:%S %Y %Z'
                    )
                    days_left = (not_after - datetime.datetime.utcnow()).days
                    if days_left < 30:
                        self.findings.append({
                            "severity": "HIGH",
                            "issue":    f"TLS certificate expires in {days_left} days",
                            "fix":      "Renew certificate before expiry"
                        })
                        self.score -= 15
        except ssl.SSLCertVerificationError:
            self.findings.append({
                "severity": "CRITICAL",
                "issue":    "TLS certificate invalid or self-signed without trust",
                "fix":      "Use a certificate from a trusted CA"
            })
            self.score -= 25
        except ConnectionRefusedError:
            self.findings.append({
                "severity": "HIGH",
                "issue":    f"No HTTPS on port {self.https_port}",
                "fix":      "Enable TLS for all web interfaces"
            })
            self.score -= 20

    def run(self):
        print(f"Auditing {self.ip}...")
        self.check_open_ports()
        self.check_mqtt_auth()
        self.check_tls()
        self._report()

    def _report(self):
        self.score = max(0, self.score)
        grade = (
            "A" if self.score >= 90 else
            "B" if self.score >= 80 else
            "C" if self.score >= 70 else
            "D" if self.score >= 60 else "F"
        )
        print(f"\n{'─'*50}")
        print(f"Security Score: {self.score}/100 (Grade: {grade})")
        print(f"{'─'*50}")

        if not self.findings:
            print("No issues found!")
        else:
            for f in sorted(self.findings, key=lambda x: x['severity']):
                print(f"\n[{f['severity']}] {f['issue']}")
                print(f"  Fix: {f['fix']}")
        return self.findings

Firmware Hardening Checklist

python

# hardening_config.py — settings that MUST be changed before production
HARDENING = {
    # Authentication
    "default_credentials_changed": True,   # Changed from factory default
    "unique_per_device_credentials": True, # Not the same as other devices
    "strong_password_policy": True,        # >= 16 chars, random

    # Communication
    "tls_all_connections": True,           # No plaintext MQTT or HTTP
    "certificate_pinning": False,          # Optional but recommended
    "mutual_tls": True,                    # mTLS where supported

    # Firmware
    "secure_boot_enabled": True,           # Only runs signed firmware
    "code_signing_verified": True,         # All OTA updates verified
    "debug_interfaces_disabled": True,     # JTAG, UART disabled in production
    "rollback_protection": True,           # Cannot downgrade to vulnerable version

    # Network
    "minimal_attack_surface": True,        # Only necessary ports open
    "network_segmented": True,             # IoT VLAN isolated
    "firewall_rules_configured": True,     # Whitelist-only outbound

    # Data
    "credentials_not_hardcoded": True,     # No passwords in source code
    "sensitive_data_encrypted": True,      # Keys, certs encrypted at rest
    "logs_no_credentials": True,           # Logs don't contain passwords/tokens

    # Lifecycle
    "ota_update_mechanism": True,          # Can be updated remotely
    "device_can_be_revoked": True,         # Can decommission compromised device
    "vulnerability_disclosure": True,      # Contact for security issues
}

passed = sum(HARDENING.values())
total  = len(HARDENING)
print(f"Hardening: {passed}/{total} controls implemented")

missing = [k for k, v in HARDENING.items() if not v]
if missing:
    print("\nAction required:")
    for m in missing:
        print(f"  ✗ {m.replace('_', ' ').title()}")

Certificate Lifecycle Management

python

# cert_manager.py — track and automate certificate renewals
from datetime import datetime, timedelta
import subprocess
import os

class CertificateManager:
    def __init__(self, ca_key="ca.key", ca_cert="ca.crt"):
        self.ca_key  = ca_key
        self.ca_cert = ca_cert
        self.issued  = {}  # device_id → expiry date

    def issue_device_cert(self, device_id, validity_days=365):
        """Issue a new certificate for a device."""
        key_file  = f"certs/{device_id}.key"
        cert_file = f"certs/{device_id}.crt"
        csr_file  = f"certs/{device_id}.csr"

        os.makedirs("certs", exist_ok=True)

        # Generate private key
        subprocess.run([
            "openssl", "genrsa", "-out", key_file, "2048"
        ], check=True, capture_output=True)

        # Generate CSR
        subprocess.run([
            "openssl", "req", "-new",
            "-key", key_file,
            "-out", csr_file,
            "-subj", f"/CN={device_id}/O=BigXStar/C=IN"
        ], check=True, capture_output=True)

        # Sign with CA
        subprocess.run([
            "openssl", "x509", "-req",
            "-in", csr_file,
            "-CA", self.ca_cert,
            "-CAkey", self.ca_key,
            "-CAcreateserial",
            "-out", cert_file,
            "-days", str(validity_days)
        ], check=True, capture_output=True)

        os.remove(csr_file)

        expiry = datetime.now() + timedelta(days=validity_days)
        self.issued[device_id] = expiry

        print(f"Issued cert for {device_id}, expires {expiry.date()}")
        return key_file, cert_file

    def check_expiring(self, warn_days=30):
        """Return devices whose certificates expire within warn_days."""
        deadline = datetime.now() + timedelta(days=warn_days)
        expiring = {
            dev: exp for dev, exp in self.issued.items()
            if exp <= deadline
        }
        if expiring:
            print(f"WARNING: {len(expiring)} certs expiring soon:")
            for dev, exp in expiring.items():
                days_left = (exp - datetime.now()).days
                print(f"  {dev}: {days_left} days left ({exp.date()})")
        return expiring

    def revoke(self, device_id):
        """Revoke a compromised device's certificate."""
        # In production: update CRL (Certificate Revocation List) or use OCSP
        cert_file = f"certs/{device_id}.crt"
        subprocess.run([
            "openssl", "ca",
            "-revoke", cert_file,
            "-config", "openssl.cnf"
        ])
        del self.issued[device_id]
        print(f"Revoked certificate for {device_id}")

cm = CertificateManager()
key, cert = cm.issue_device_cert("esp32-node-001", validity_days=365)
key, cert = cm.issue_device_cert("esp32-node-002", validity_days=365)
cm.check_expiring(warn_days=90)

Security Incident Response

python

# incident_response.py — detect and respond to anomalous device behavior
import paho.mqtt.client as mqtt
import json
import time

ANOMALY_THRESHOLDS = {
    "reconnects_per_hour": 10,  # Unusual churn = compromised device?
    "publish_rate_per_min": 60,  # Unusual data volume = data exfiltration?
    "unknown_topics": True,     # Publishing to unexpected topics
}

known_topics = {
    "sensors/#",
    "devices/+/status",
    "devices/+/telemetry"
}

device_stats = {}

def track_device(device_id, topic):
    now = time.time()
    if device_id not in device_stats:
        device_stats[device_id] = {
            "first_seen": now,
            "last_seen":  now,
            "message_count": 0,
            "topics": set(),
            "suspicious": False
        }

    stats = device_stats[device_id]
    stats["last_seen"]      = now
    stats["message_count"] += 1
    stats["topics"].add(topic)

    # Check rate
    duration_min = (now - stats["first_seen"]) / 60 or 0.001
    rate = stats["message_count"] / duration_min
    if rate > ANOMALY_THRESHOLDS["publish_rate_per_min"]:
        if not stats["suspicious"]:
            print(f"ALERT: {device_id} publishing at {rate:.0f} msg/min — possible exfiltration!")
            stats["suspicious"] = True
            quarantine_device(device_id)

def quarantine_device(device_id):
    """Disconnect and blacklist a suspicious device."""
    print(f"Quarantining device: {device_id}")
    # In production:
    # 1. Publish to Mosquitto's $SYS topic to forcibly disconnect
    # 2. Update firewall rules to block device MAC/IP
    # 3. Revoke device certificate
    # 4. Page on-call security engineer
    # 5. Preserve logs for forensics

def on_message(client, userdata, msg):
    try:
        topic = msg.topic
        data  = json.loads(msg.payload)
        device_id = data.get("device_id", "unknown")
        track_device(device_id, topic)
    except:
        pass

client = mqtt.Client("security-monitor")
client.on_connect = lambda c, u, f, r: c.subscribe("#", qos=1)
client.on_message = on_message
client.connect("localhost", 1883, 60)
client.loop_forever()

Summary

OWASP IoT Top 10 gives you the most common vulnerability categories — use it as a checklist before deployment
Security auditing scripts can automatically detect open ports, anonymous MQTT, and invalid TLS certificates
The hardening checklist covers authentication, communication, firmware, network, data, and lifecycle
Certificate lifecycle management is critical — expired certificates silently break devices in the field
Revocation is the nuclear option for compromised devices — plan for it from day one
Anomaly detection on device behavior patterns (publish rate, unusual topics) can detect compromised devices

Next, we'll build OTA (over-the-air) firmware update pipelines — keeping your devices secure and up-to-date without physical access.