Firmware Over-the-Air (FOTA) — IoT Device Update Management Guide

import hashlib
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec, utils

def sign_firmware(firmware_path: str, private_key_path: str) -> bytes:
    with open(firmware_path, 'rb') as f:
        firmware = f.read()

    firmware_hash = hashlib.sha256(firmware).digest()

    with open(private_key_path, 'rb') as f:
        private_key = ec.load_der_private_key(f.read(), password=None)

    signature = private_key.sign(
        firmware_hash,
        ec.ECDSA(hashes.SHA256())
    )

    return signature

def verify_firmware(firmware_path: str, signature: bytes,
                    public_key_path: str) -> bool:
    with open(firmware_path, 'rb') as f:
        firmware = f.read()

    firmware_hash = hashlib.sha256(firmware).digest()

    with open(public_key_path, 'rb') as f:
        public_key = ec.load_der_public_key(f.read())

    try:
        public_key.verify(signature, firmware_hash, ec.ECDSA(hashes.SHA256()))
        return True
    except Exception:
        return False

#include <WiFi.h>
#include <HTTPClient.h>
#include <Update.h>

const char* WIFI_SSID = "YourNetwork";
const char* WIFI_PASS = "YourPassword";
const char* FIRMWARE_URL = "https://ota.dodatech.com/firmware/esp32-v2.1.bin";

void setup() {
  Serial.begin(115200);
  WiFi.begin(WIFI_SSID, WIFI_PASS);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
  }
  Serial.println("WiFi connected");

  HTTPClient http;
  http.begin(FIRMWARE_URL);
  int httpCode = http.GET();

  if (httpCode == HTTP_CODE_OK) {
    int contentLength = http.getSize();
    bool canUpdate = Update.begin(contentLength);

    if (canUpdate) {
      WiFiClient* stream = http.getStreamPtr();
      size_t written = Update.writeStream(*stream);

      if (written == contentLength && Update.end()) {
        Serial.println("Update complete, restarting...");
        ESP.restart();
      } else {
        Serial.printf("Update failed: %s\n", Update.errorString());
      }
    }
  }
  http.end();
}

void loop() {
  // Normal device operation
}

from http.server import HTTPServer, BaseHTTPRequestHandler
import os, json

FIRMWARE_DIR = "/srv/fota"
DEVICE_REGISTRY = {}

class FOTAHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/api/v1/manifest":
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            manifest = {
                "latest_version": "2.1.0",
                "release_date": "2026-06-24",
                "files": [
                    {
                        "name": "esp32-firmware-v2.1.0.bin",
                        "size": 524288,
                        "sha256": "a1b2c3...",
                        "url": "/firmware/esp32-v2.1.0.bin]
                    }
                ]
            }
            self.wfile.write(json.dumps(manifest).encode())

        elif self.path.startswith("/firmware/"):
            filename = self.path.split("/")[-1]
            filepath = os.path.join(FIRMWARE_DIR, filename)
            if os.path.exists(filepath):
                self.send_response(200)
                self.send_header("Content-Type", "application/octet-stream")
                self.send_header("Content-Length", str(os.path.getsize(filepath)))
                self.end_headers()
                with open(filepath, "rb") as f:
                    self.wfile.write(f.read())
            else:
                self.send_response(404)
                self.end_headers()

server = HTTPServer(("0.0.0.0", 8080), FOTAHandler)
server.serve_forever()

import subprocess, os

def generate_delta(old_firmware: str, new_firmware: str, delta_path: str):
    subprocess.run(["bsdiff", old_firmware, new_firmware, delta_path], check=True)

def apply_delta(old_firmware: str, delta_path: str, output_path: str):
    subprocess.run(["bspatch", old_firmware, output_path, delta_path], check=True)

# Delta is typically 10-30% of full binary size
old_size = os.path.getsize("firmware-v2.0.bin")
new_size = os.path.getsize("firmware-v2.1.bin")
delta_size = os.path.getsize("delta-v2.0-to-v2.1.bin")
print(f"Old: {old_size} bytes, New: {new_size} bytes, Delta: {delta_size} bytes")
print(f"Delta is {delta_size / new_size * 100:.1f}% of full binary")

import hashlib, os, json

class FOTAUpdate:
    def __init__(self, device_id: str, flash_size: int = 1048576):
        self.device_id = device_id
        self.flash = bytearray(flash_size)
        self.active_slot = 0
        self.slot_size = flash_size // 2
        self.firmware_versions = ["", ""]

    def write_firmware(self, slot: int, data: bytes, version: str):
        start = slot * self.slot_size
        end = start + len(data)
        if end > len(self.flash):
            raise ValueError("Firmware too large for slot")
        self.flash[start:end] = data
        self.firmware_versions[slot] = version
        print(f"Wrote v{version} to Slot {'A' if slot == 0 else 'B'}")

    def verify_and_boot(self, slot: int, expected_hash: str) -> bool:
        start = slot * self.slot_size
        data = bytes(self.flash[start:start + self.slot_size]).rstrip(b'\x00')
        actual_hash = hashlib.sha256(data).hexdigest()
        if actual_hash == expected_hash:
            self.active_slot = slot
            print(f"Booted from Slot {'A' if slot == 0 else 'B'}: v{self.firmware_versions[slot]}")
            return True
        print(f"Hash mismatch on Slot {'A' if slot == 0 else 'B'}")
        return False

    def rollout_update(self, new_firmware: bytes, version: str, firmware_hash: str):
        inactive_slot = 1 if self.active_slot == 0 else 0
        self.write_firmware(inactive_slot, new_firmware, version)
        if self.verify_and_boot(inactive_slot, firmware_hash):
            print(f"Update successful: v{self.firmware_versions[self.active_slot]}")
        else:
            self.verify_and_boot(self.active_slot, firmware_hash)
            print("Rollback to previous firmware")

device = FOTAUpdate("sensor-01")
device.write_firmware(0, b"old_firmware_data_" * 1000, "1.0.0")
device.verify_and_boot(0, hashlib.sha256(b"old_firmware_data_" * 1000).hexdigest())

new_fw = b"new_firmware_data_" * 1100
new_hash = hashlib.sha256(new_fw).hexdigest()
device.rollout_update(new_fw, "2.0.0", new_hash)