Common attack vectors:

1. Default credentials
   ├── Admin/admin, root/root, or no password at all
   └── Millions of devices use identical factory defaults

2. Unencrypted communication
   ├── MQTT on port 1883 (plaintext)
   └── Sniffable on any shared network

3. No update mechanism
   ├── Vulnerabilities discovered after deployment
   └── No way to patch without physical access

4. Weak authentication
   ├── Pre-shared keys hardcoded in firmware
   └── No certificate rotation

5. Exposed management interfaces
   └── Telnet, FTP, HTTP admin pages accessible from the internet

# Python — verify TLS is working correctly
import ssl
import socket

def check_tls(host, port=443):
    """Verify TLS certificate of a remote server."""
    context = ssl.create_default_context()
    # context.check_hostname = False  # NEVER do this in production!
    # context.verify_mode = ssl.CERT_NONE  # NEVER do this!

    with socket.create_connection((host, port)) as sock:
        with context.wrap_socket(sock, server_hostname=host) as ssock:
            cert = ssock.getpeercert()
            print(f"Subject:  {dict(x[0] for x in cert['subject'])}")
            print(f"Issuer:   {dict(x[0] for x in cert['issuer'])}")
            print(f"Expires:  {cert['notAfter']}")
            print(f"Protocol: {ssock.version()}")
            return True

check_tls("broker.hivemq.com", 8883)

# MQTT over TLS on Raspberry Pi / Python
import paho.mqtt.client as mqtt
import ssl

client = mqtt.Client()

# Option A: Server certificate verification only (simpler)
client.tls_set(
    ca_certs   = "ca.crt",          # Download from your broker
    tls_version = ssl.PROTOCOL_TLS_CLIENT
)

# Option B: Mutual TLS (mTLS) — device has its own certificate
# The broker verifies the device is who it claims to be
client.tls_set(
    ca_certs   = "ca.crt",
    certfile   = "device-cert.crt",  # Unique per device
    keyfile    = "device-key.pem",   # Private key, NEVER share
    tls_version = ssl.PROTOCOL_TLS_CLIENT
)

client.connect("your-broker.com", 8883)

# BAD — credentials visible in source code and git history
WIFI_PASSWORD = "MyPassword123"
MQTT_API_KEY  = "abc123secretkey"

# GOOD — load from environment variables or a config file excluded from git
import os

WIFI_PASSWORD = os.environ.get("WIFI_PASSWORD")
MQTT_API_KEY  = os.environ.get("MQTT_API_KEY")

if not WIFI_PASSWORD:
    raise EnvironmentError("WIFI_PASSWORD not set")

# MicroPython ESP32 — credentials in a config.py (excluded from git)
# config.py (add to .gitignore)
WIFI_SSID     = "MyNetwork"
WIFI_PASSWORD = "MyPassword"
MQTT_BROKER   = "mybroker.example.com"
MQTT_USER     = "device-01"
MQTT_PASSWORD = "unique-per-device-token"
DEVICE_ID     = "esp32-a4f3b2"

# main.py — import safely
try:
    from config import WIFI_SSID, WIFI_PASSWORD, MQTT_BROKER
except ImportError:
    print("ERROR: config.py missing. Copy config.example.py and fill in values.")
    raise

# .gitignore
# config.py      ← never commit credentials
# *.key          ← never commit private keys
# *.pem          ← never commit certificates

# paho-mqtt client with authentication
client.username_pw_set(
    username="device-sensor-01",
    password="Str0ng!UniquePerDevice#Token"
)
# ✓ Each device should have a unique password
# ✓ Rotate passwords periodically
# ✗ Password transmitted over the wire (mitigated by TLS)

# Generate CA key and certificate (run once)
openssl genrsa -out ca.key 4096
openssl req -new -x509 -key ca.key -out ca.crt -days 3650 \
    -subj "/CN=IoT-CA/O=BigXStar/C=IN"

# Generate device certificate (run for each device)
DEVICE_ID="esp32-001"
openssl genrsa -out ${DEVICE_ID}.key 2048
openssl req -new -key ${DEVICE_ID}.key -out ${DEVICE_ID}.csr \
    -subj "/CN=${DEVICE_ID}/O=BigXStar/C=IN"
openssl x509 -req -in ${DEVICE_ID}.csr \
    -CA ca.crt -CAkey ca.key -CAcreateserial \
    -out ${DEVICE_ID}.crt -days 365

# Deploy to device: ca.crt, esp32-001.crt, esp32-001.key
# Never deploy ca.key to any device!

# MQTT subscriber — always validate incoming data
import json

def parse_sensor_payload(payload_bytes):
    """Safely parse and validate incoming sensor data."""
    try:
        data = json.loads(payload_bytes.decode('utf-8'))
    except (json.JSONDecodeError, UnicodeDecodeError):
        print("Invalid JSON payload — discarded")
        return None

    # Validate required fields
    required = ['device_id', 'temperature', 'humidity']
    if not all(k in data for k in required):
        print(f"Missing fields: {required}")
        return None

    # Validate types and ranges
    try:
        device_id   = str(data['device_id'])[:64]  # Max 64 chars
        temperature = float(data['temperature'])
        humidity    = float(data['humidity'])
    except (ValueError, TypeError):
        print("Invalid field types")
        return None

    # Sanity bounds — reject physically impossible values
    if not (-40 <= temperature <= 125):
        print(f"Implausible temperature: {temperature}")
        return None

    if not (0 <= humidity <= 100):
        print(f"Implausible humidity: {humidity}")
        return None

    # Sanitize device_id (alphanumeric + dashes only)
    import re
    if not re.match(r'^[a-zA-Z0-9\-_]{1,64}$', device_id):
        print(f"Invalid device_id: {device_id}")
        return None

    return {
        'device_id':   device_id,
        'temperature': round(temperature, 2),
        'humidity':    round(humidity, 2)
    }

Recommended network architecture:

  Internet
     │
   Router/Firewall
     │
  ┌──┼──────────────┐
  │  │              │
Main VLAN      IoT VLAN
(192.168.1.x)  (10.0.2.x)
  │              │
  PCs, phones  Sensors, cameras,
  NAS, etc.    smart devices

Firewall rules:
  IoT VLAN → Internet:     ALLOW (port 8883 MQTT, 443 HTTPS only)
  IoT VLAN → Main VLAN:    DENY ALL
  Main VLAN → IoT VLAN:    DENY ALL (or limited management access)
  Internet → IoT VLAN:     DENY ALL (no inbound)

# Security audit checklist as code
SECURITY_CHECKLIST = {
    "no_default_credentials":    True,   # Changed from factory defaults
    "tls_enabled":               True,   # All communication encrypted
    "certificate_auth":          True,   # mTLS where possible
    "credentials_not_hardcoded": True,   # Using env vars or secure storage
    "input_validation":          True,   # All incoming data validated
    "ota_updates_signed":        True,   # Firmware updates verified by signature
    "secure_boot":               True,   # Device only runs signed firmware
    "network_segmented":         True,   # IoT VLAN separate from main network
    "minimal_services":          True,   # No Telnet, FTP, unnecessary ports
    "logging_enabled":           True,   # Security events logged
    "physical_access_controlled": True,  # JTAG/UART debug disabled in production
}

passed = sum(SECURITY_CHECKLIST.values())
total  = len(SECURITY_CHECKLIST)
print(f"Security score: {passed}/{total} ({passed/total*100:.0f}%)")

failed = [k for k, v in SECURITY_CHECKLIST.items() if not v]
if failed:
    print(f"Action required: {', '.join(failed)}")