REST APIs, Webhooks, and Device Integration

{
  "status": "ok",
  "data": {
    "device_id": "esp32-001",
    "temperature": 23.4,
    "humidity": 61.2,
    "timestamp": "2026-03-10T14:30:00Z"
  }
}

{
  "status": "error",
  "error": {
    "code": 404,
    "message": "Device esp32-999 not found"
  }
}

flask==3.1.0
flask-limiter==3.5.0

import sqlite3
import json
from datetime import datetime, timezone

DB_PATH = "devices.db"

def get_db():
    """Get a database connection with row factory enabled."""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    return conn

def init_db():
    """Create tables if they do not exist."""
    conn = get_db()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS devices (
            device_id   TEXT PRIMARY KEY,
            name        TEXT NOT NULL,
            type        TEXT NOT NULL DEFAULT 'sensor',
            location    TEXT,
            config      TEXT DEFAULT '{}',
            status      TEXT DEFAULT 'active',
            registered  TEXT NOT NULL,
            last_seen   TEXT
        );

        CREATE TABLE IF NOT EXISTS readings (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            device_id   TEXT NOT NULL,
            payload     TEXT NOT NULL,
            timestamp   TEXT NOT NULL,
            FOREIGN KEY (device_id) REFERENCES devices(device_id)
        );

        CREATE TABLE IF NOT EXISTS commands (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            device_id   TEXT NOT NULL,
            command     TEXT NOT NULL,
            params      TEXT DEFAULT '{}',
            status      TEXT DEFAULT 'pending',
            created     TEXT NOT NULL,
            executed    TEXT,
            FOREIGN KEY (device_id) REFERENCES devices(device_id)
        );

        CREATE INDEX IF NOT EXISTS idx_readings_device
            ON readings(device_id, timestamp);
        CREATE INDEX IF NOT EXISTS idx_commands_device
            ON commands(device_id, created);
    """)
    conn.close()

def device_exists(device_id):
    """Check if a device is registered."""
    conn = get_db()
    row = conn.execute(
        "SELECT 1 FROM devices WHERE device_id = ?", (device_id,)
    ).fetchone()
    conn.close()
    return row is not None

def register_device(device_id, name, device_type="sensor",
                    location=None, config=None):
    """Register a new device. Returns True on success."""
    conn = get_db()
    now = datetime.now(timezone.utc).isoformat()
    config_str = json.dumps(config or {})
    try:
        conn.execute(
            """INSERT INTO devices
               (device_id, name, type, location, config, registered)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (device_id, name, device_type, location, config_str, now)
        )
        conn.commit()
        return True
    except sqlite3.IntegrityError:
        return False
    finally:
        conn.close()

def get_all_devices():
    """Return all registered devices."""
    conn = get_db()
    rows = conn.execute(
        "SELECT * FROM devices ORDER BY registered DESC"
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]

def get_device(device_id):
    """Return a single device or None."""
    conn = get_db()
    row = conn.execute(
        "SELECT * FROM devices WHERE device_id = ?", (device_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None

def update_device_config(device_id, config):
    """Update a device's configuration. Returns True if found."""
    conn = get_db()
    now = datetime.now(timezone.utc).isoformat()
    cursor = conn.execute(
        "UPDATE devices SET config = ?, last_seen = ? WHERE device_id = ?",
        (json.dumps(config), now, device_id)
    )
    conn.commit()
    updated = cursor.rowcount > 0
    conn.close()
    return updated

def delete_device(device_id):
    """Remove a device and all its data. Returns True if found."""
    conn = get_db()
    cursor = conn.execute(
        "DELETE FROM devices WHERE device_id = ?", (device_id,)
    )
    conn.execute(
        "DELETE FROM readings WHERE device_id = ?", (device_id,)
    )
    conn.execute(
        "DELETE FROM commands WHERE device_id = ?", (device_id,)
    )
    conn.commit()
    deleted = cursor.rowcount > 0
    conn.close()
    return deleted

def insert_reading(device_id, payload):
    """Store a sensor reading."""
    conn = get_db()
    now = datetime.now(timezone.utc).isoformat()
    conn.execute(
        "INSERT INTO readings (device_id, payload, timestamp) VALUES (?, ?, ?)",
        (device_id, json.dumps(payload), now)
    )
    conn.execute(
        "UPDATE devices SET last_seen = ? WHERE device_id = ?",
        (now, device_id)
    )
    conn.commit()
    conn.close()

def get_latest_reading(device_id):
    """Return the most recent reading for a device."""
    conn = get_db()
    row = conn.execute(
        """SELECT * FROM readings
           WHERE device_id = ?
           ORDER BY timestamp DESC LIMIT 1""",
        (device_id,)
    ).fetchone()
    conn.close()
    if row:
        result = dict(row)
        result["payload"] = json.loads(result["payload"])
        return result
    return None

def get_readings_range(device_id, start, end, limit=1000):
    """Return readings within a time range."""
    conn = get_db()
    rows = conn.execute(
        """SELECT * FROM readings
           WHERE device_id = ? AND timestamp >= ? AND timestamp <= ?
           ORDER BY timestamp ASC LIMIT ?""",
        (device_id, start, end, limit)
    ).fetchall()
    conn.close()
    results = []
    for r in rows:
        d = dict(r)
        d["payload"] = json.loads(d["payload"])
        results.append(d)
    return results

def insert_command(device_id, command, params=None):
    """Queue a command for a device. Returns the command ID."""
    conn = get_db()
    now = datetime.now(timezone.utc).isoformat()
    cursor = conn.execute(
        """INSERT INTO commands (device_id, command, params, created)
           VALUES (?, ?, ?, ?)""",
        (device_id, command, json.dumps(params or {}), now)
    )
    conn.commit()
    cmd_id = cursor.lastrowid
    conn.close()
    return cmd_id

import os
import hashlib
import secrets
from functools import wraps
from flask import request, jsonify

# In production, store hashed keys in a database.
# For this lesson, we use a simple dictionary.
API_KEYS = {
    # key: {"name": "owner", "permissions": ["read", "write", "admin"]}
}

def generate_api_key(name, permissions=None):
    """Generate a new API key in swk_ format."""
    raw = secrets.token_hex(24)
    key = f"swk_{raw}"
    API_KEYS[key] = {
        "name": name,
        "permissions": permissions or ["read"],
    }
    return key

def require_auth(permission="read"):
    """Decorator that checks for a valid API key with the required permission."""
    def decorator(f):
        @wraps(f)
        def wrapped(*args, **kwargs):
            auth_header = request.headers.get("Authorization", "")

            if auth_header.startswith("Bearer "):
                key = auth_header[7:]
            else:
                key = request.headers.get("X-API-Key", "")

            if not key:
                return jsonify({
                    "status": "error",
                    "error": {
                        "code": 401,
                        "message": "Missing API key. Use Authorization: Bearer <key> "
                                   "or X-API-Key header."
                    }
                }), 401

            key_data = API_KEYS.get(key)
            if not key_data:
                return jsonify({
                    "status": "error",
                    "error": {
                        "code": 401,
                        "message": "Invalid API key."
                    }
                }), 401

            if permission not in key_data["permissions"]:
                return jsonify({
                    "status": "error",
                    "error": {
                        "code": 403,
                        "message": f"Key does not have '{permission}' permission."
                    }
                }), 403

            return f(*args, **kwargs)
        return wrapped
    return decorator

# Generate a default admin key for development
DEFAULT_KEY = generate_api_key("admin", ["read", "write", "admin"])

from flask import Flask, request, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

from models import (
    init_db, get_all_devices, get_device, register_device,
    update_device_config, delete_device, device_exists,
    insert_reading, get_latest_reading, get_readings_range,
    insert_command
)
from auth import require_auth, DEFAULT_KEY

app = Flask(__name__)

# Rate limiting: 100 requests per minute per IP
limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["100 per minute"],
    storage_uri="memory://",
)

# ================================================================
# Device Management Endpoints
# ================================================================

@app.route("/api/devices", methods=["GET"])
@require_auth("read")
def list_devices():
    """GET /api/devices - List all registered devices."""
    devices = get_all_devices()
    return jsonify({
        "status": "ok",
        "count": len(devices),
        "data": devices
    })

@app.route("/api/devices", methods=["POST"])
@require_auth("write")
def create_device():
    """POST /api/devices - Register a new device."""
    body = request.get_json(silent=True)
    if not body:
        return jsonify({
            "status": "error",
            "error": {"code": 400, "message": "Request body must be JSON."}
        }), 400

    device_id = body.get("device_id")
    name = body.get("name")

    if not device_id or not name:
        return jsonify({
            "status": "error",
            "error": {"code": 400,
                      "message": "Fields 'device_id' and 'name' are required."}
        }), 400

    success = register_device(
        device_id=device_id,
        name=name,
        device_type=body.get("type", "sensor"),
        location=body.get("location"),
        config=body.get("config"),
    )

    if not success:
        return jsonify({
            "status": "error",
            "error": {"code": 409,
                      "message": f"Device '{device_id}' already exists."}
        }), 409

    device = get_device(device_id)
    return jsonify({"status": "ok", "data": device}), 201

@app.route("/api/devices/<device_id>", methods=["GET"])
@require_auth("read")
def get_device_info(device_id):
    """GET /api/devices/{id} - Get device details."""
    device = get_device(device_id)
    if not device:
        return jsonify({
            "status": "error",
            "error": {"code": 404,
                      "message": f"Device '{device_id}' not found."}
        }), 404

    return jsonify({"status": "ok", "data": device})

@app.route("/api/devices/<device_id>", methods=["DELETE"])
@require_auth("admin")
def remove_device(device_id):
    """DELETE /api/devices/{id} - Decommission a device."""
    deleted = delete_device(device_id)
    if not deleted:
        return jsonify({
            "status": "error",
            "error": {"code": 404,
                      "message": f"Device '{device_id}' not found."}
        }), 404

    return jsonify({"status": "ok", "message": "Device removed."}), 200

# ================================================================
# Sensor Data Endpoints
# ================================================================

@app.route("/api/devices/<device_id>/data", methods=["GET"])
@require_auth("read")
def get_device_data(device_id):
    """
    GET /api/devices/{id}/data - Get sensor readings.
    Query params: start, end (ISO 8601 timestamps) for historical range.
    Without params, returns the latest reading.
    """
    if not device_exists(device_id):
        return jsonify({
            "status": "error",
            "error": {"code": 404,
                      "message": f"Device '{device_id}' not found."}
        }), 404

    start = request.args.get("start")
    end = request.args.get("end")

    if start and end:
        # Historical range query
        limit = request.args.get("limit", 1000, type=int)
        limit = min(limit, 10000)  # cap at 10k rows
        readings = get_readings_range(device_id, start, end, limit)
        return jsonify({
            "status": "ok",
            "count": len(readings),
            "data": readings
        })
    else:
        # Latest reading
        reading = get_latest_reading(device_id)
        if not reading:
            return jsonify({
                "status": "ok",
                "data": None,
                "message": "No readings yet."
            })
        return jsonify({"status": "ok", "data": reading})

@app.route("/api/devices/<device_id>/data", methods=["POST"])
@require_auth("write")
@limiter.limit("60 per minute")
def ingest_data(device_id):
    """POST /api/devices/{id}/data - Ingest a sensor reading via HTTP."""
    if not device_exists(device_id):
        return jsonify({
            "status": "error",
            "error": {"code": 404,
                      "message": f"Device '{device_id}' not found."}
        }), 404

    body = request.get_json(silent=True)
    if not body:
        return jsonify({
            "status": "error",
            "error": {"code": 400, "message": "Request body must be JSON."}
        }), 400

    insert_reading(device_id, body)
    return jsonify({"status": "ok", "message": "Reading stored."}), 201

# ================================================================
# Command Endpoints
# ================================================================

@app.route("/api/devices/<device_id>/command", methods=["POST"])
@require_auth("write")
def send_command(device_id):
    """POST /api/devices/{id}/command - Send a command to a device."""
    if not device_exists(device_id):
        return jsonify({
            "status": "error",
            "error": {"code": 404,
                      "message": f"Device '{device_id}' not found."}
        }), 404

    body = request.get_json(silent=True)
    if not body or "command" not in body:
        return jsonify({
            "status": "error",
            "error": {"code": 400,
                      "message": "Field 'command' is required."}
        }), 400

    cmd_id = insert_command(
        device_id=device_id,
        command=body["command"],
        params=body.get("params"),
    )

    return jsonify({
        "status": "ok",
        "data": {"command_id": cmd_id, "status": "pending"}
    }), 201

# ================================================================
# Configuration Endpoints
# ================================================================

@app.route("/api/devices/<device_id>/config", methods=["GET"])
@require_auth("read")
def get_config(device_id):
    """GET /api/devices/{id}/config - Get device configuration."""
    device = get_device(device_id)
    if not device:
        return jsonify({
            "status": "error",
            "error": {"code": 404,
                      "message": f"Device '{device_id}' not found."}
        }), 404

    import json
    config = json.loads(device.get("config", "{}"))
    return jsonify({"status": "ok", "data": config})

@app.route("/api/devices/<device_id>/config", methods=["PUT"])
@require_auth("write")
def update_config(device_id):
    """PUT /api/devices/{id}/config - Update device configuration."""
    body = request.get_json(silent=True)
    if not body:
        return jsonify({
            "status": "error",
            "error": {"code": 400, "message": "Request body must be JSON."}
        }), 400

    updated = update_device_config(device_id, body)
    if not updated:
        return jsonify({
            "status": "error",
            "error": {"code": 404,
                      "message": f"Device '{device_id}' not found."}
        }), 404

    return jsonify({"status": "ok", "message": "Configuration updated."})

# ================================================================
# Health Check (no auth required)
# ================================================================

@app.route("/api/health", methods=["GET"])
@limiter.exempt
def health():
    """GET /api/health - Server health check."""
    return jsonify({"status": "ok", "service": "iot-rest-api"})

# ================================================================
# Entry Point
# ================================================================

if __name__ == "__main__":
    init_db()
    print(f"\n  Default API key: {DEFAULT_KEY}")
    print(f"  Use: curl -H 'Authorization: Bearer {DEFAULT_KEY}' ...\n")
    app.run(host="0.0.0.0", port=5000, debug=True)

import hmac
import hashlib
import json
import logging
from datetime import datetime, timezone

from flask import Flask, request, jsonify

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("webhook")

# Shared secret (in production, load from environment variable)
WEBHOOK_SECRET = "your-webhook-secret-change-this"

def verify_signature(payload_bytes, signature_header):
    """
    Verify the HMAC-SHA256 signature of a webhook payload.

    Args:
        payload_bytes: The raw request body as bytes.
        signature_header: The value of the X-Webhook-Signature header.

    Returns:
        True if the signature is valid, False otherwise.
    """
    if not signature_header:
        logger.warning("Missing X-Webhook-Signature header")
        return False

    # The header format is "sha256=<hex digest>"
    if not signature_header.startswith("sha256="):
        logger.warning("Invalid signature format: %s", signature_header)
        return False

    received_sig = signature_header[7:]  # strip "sha256=" prefix

    # Compute expected signature
    expected_sig = hmac.new(
        key=WEBHOOK_SECRET.encode("utf-8"),
        msg=payload_bytes,
        digestmod=hashlib.sha256,
    ).hexdigest()

    # Use constant-time comparison to prevent timing attacks
    return hmac.compare_digest(received_sig, expected_sig)

def process_event(event):
    """
    Process a validated webhook event.
    Replace this with your actual business logic.
    """
    event_type = event.get("event", "unknown")

    if event_type == "device_offline":
        logger.info(
            "ALERT: Device %s went offline at %s",
            event.get("device_id"),
            event.get("timestamp"),
        )
        # Example: send a Slack notification, page an engineer, etc.

    elif event_type == "threshold_exceeded":
        logger.info(
            "ALERT: Device %s, %s = %s (threshold: %s, severity: %s)",
            event.get("device_id"),
            event.get("condition"),
            event.get("current_value"),
            event.get("threshold"),
            event.get("severity"),
        )
        # Example: trigger an automation rule

    elif event_type == "device_online":
        logger.info(
            "INFO: Device %s came back online at %s",
            event.get("device_id"),
            event.get("timestamp"),
        )

    else:
        logger.info("Received event type '%s': %s", event_type, event)

@app.route("/webhooks/iot", methods=["POST"])
def webhook_endpoint():
    """Receive and process webhook events."""

    # Step 1: Get the raw body (before JSON parsing, for signature verification)
    payload_bytes = request.get_data()

    # Step 2: Verify the HMAC signature
    signature = request.headers.get("X-Webhook-Signature", "")
    if not verify_signature(payload_bytes, signature):
        logger.warning(
            "Invalid signature from %s", request.remote_addr
        )
        return jsonify({"status": "error", "message": "Invalid signature."}), 401

    # Step 3: Parse the JSON payload
    try:
        event = json.loads(payload_bytes)
    except json.JSONDecodeError:
        return jsonify({"status": "error", "message": "Invalid JSON."}), 400

    # Step 4: Log receipt
    logger.info(
        "Received webhook: event=%s, device=%s",
        event.get("event"),
        event.get("device_id"),
    )

    # Step 5: Process the event
    process_event(event)

    # Step 6: Return 200 to acknowledge receipt
    return jsonify({"status": "ok", "received": True}), 200

if __name__ == "__main__":
    logger.info("Webhook receiver starting on port 8080")
    app.run(host="0.0.0.0", port=8080, debug=True)

import hmac
import hashlib
import json
import requests
from datetime import datetime, timezone

WEBHOOK_URL = "http://localhost:8080/webhooks/iot"
WEBHOOK_SECRET = "your-webhook-secret-change-this"

def send_test_event(event_data):
    """Sign and send a test webhook event."""
    payload = json.dumps(event_data).encode("utf-8")

    # Compute HMAC-SHA256 signature
    signature = hmac.new(
        key=WEBHOOK_SECRET.encode("utf-8"),
        msg=payload,
        digestmod=hashlib.sha256,
    ).hexdigest()

    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Signature": f"sha256={signature}",
    }

    response = requests.post(WEBHOOK_URL, data=payload, headers=headers)
    print(f"Status: {response.status_code}")
    print(f"Response: {response.json()}")
    return response

# Test 1: Device offline event
print("--- Test 1: Device offline ---")
send_test_event({
    "event": "device_offline",
    "alert_id": "alert-001",
    "device_id": "esp32-001",
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "message": "Device missed 3 consecutive heartbeats.",
})

# Test 2: Threshold exceeded event
print("\n--- Test 2: Threshold exceeded ---")
send_test_event({
    "event": "threshold_exceeded",
    "alert_id": "alert-002",
    "device_id": "esp32-001",
    "condition": "temperature",
    "current_value": 38.7,
    "threshold": 35.0,
    "severity": "critical",
    "timestamp": datetime.now(timezone.utc).isoformat(),
})

# Test 3: Invalid signature (should be rejected)
print("\n--- Test 3: Invalid signature ---")
payload = json.dumps({"event": "test"}).encode("utf-8")
bad_sig = "sha256=0000000000000000000000000000000000000000000000000000000000000000"
response = requests.post(
    WEBHOOK_URL,
    data=payload,
    headers={
        "Content-Type": "application/json",
        "X-Webhook-Signature": bad_sig,
    },
)
print(f"Status: {response.status_code} (expected 401)")
print(f"Response: {response.json()}")

{
  "event": "device_offline",
  "alert_id": "swk_alert_8f3a2b1c",
  "device_id": "esp32-001",
  "device_name": "Greenhouse Sensor",
  "timestamp": "2026-03-10T14:30:00Z",
  "message": "Device missed 3 consecutive heartbeats.",
  "severity": "warning",
  "metadata": {
    "last_seen": "2026-03-10T14:25:00Z",
    "expected_interval_s": 30
  }
}

{
  "event": "threshold_exceeded",
  "alert_id": "swk_alert_4d7e9f0a",
  "device_id": "esp32-001",
  "device_name": "Greenhouse Sensor",
  "condition": "temperature > 35.0",
  "current_value": 38.7,
  "threshold": 35.0,
  "severity": "critical",
  "timestamp": "2026-03-10T15:10:00Z",
  "metadata": {
    "unit": "celsius",
    "duration_above_s": 120
  }
}

#include "esp_http_client.h"
#include "esp_log.h"
#include "cJSON.h"

static const char *TAG = "http_fallback";

/* Your REST API server address */
#define API_URL     "http://192.168.1.100:5000/api/devices/esp32-001/data"
#define API_KEY     "swk_your_api_key_here"

/* For SiliconWit.io, use:
 * #define API_URL  "https://siliconwit.io/api/devices/ingest"
 * and add the root CA certificate for TLS.
 */

esp_err_t http_post_sensor_data(float temperature, float humidity)
{
    /* Build JSON payload */
    cJSON *root = cJSON_CreateObject();
    cJSON_AddStringToObject(root, "device_id", "esp32-001");
    cJSON_AddNumberToObject(root, "temperature", temperature);
    cJSON_AddNumberToObject(root, "humidity", humidity);
    char *payload = cJSON_PrintUnformatted(root);
    cJSON_Delete(root);

    if (!payload) {
        ESP_LOGE(TAG, "Failed to create JSON payload");
        return ESP_FAIL;
    }

    /* Configure HTTP client */
    esp_http_client_config_t config = {
        .url = API_URL,
        .method = HTTP_METHOD_POST,
        .timeout_ms = 10000,
    };
    esp_http_client_handle_t client = esp_http_client_init(&config);

    /* Set headers */
    esp_http_client_set_header(client, "Content-Type", "application/json");
    esp_http_client_set_header(client, "Authorization",
                               "Bearer " API_KEY);

    /* Set POST data */
    esp_http_client_set_post_field(client, payload, strlen(payload));

    /* Perform the request */
    esp_err_t err = esp_http_client_perform(client);

    if (err == ESP_OK) {
        int status = esp_http_client_get_status_code(client);
        ESP_LOGI(TAG, "HTTP POST status=%d", status);

        if (status == 201) {
            ESP_LOGI(TAG, "Data ingested successfully via HTTP");
        } else {
            ESP_LOGW(TAG, "Server returned status %d", status);
        }
    } else {
        ESP_LOGE(TAG, "HTTP POST failed: %s", esp_err_to_name(err));
    }

    esp_http_client_cleanup(client);
    free(payload);
    return err;
}

void sensor_publish_task(void *pvParameters)
{
    while (1) {
        float temp = read_temperature();
        float hum = read_humidity();

        /* Try MQTT first */
        int mqtt_rc = mqtt_publish_reading(temp, hum);

        if (mqtt_rc != 0) {
            ESP_LOGW(TAG, "MQTT publish failed (rc=%d), trying HTTP", mqtt_rc);
            http_post_sensor_data(temp, hum);
        }

        vTaskDelay(pdMS_TO_TICKS(30000));  /* 30 seconds */
    }
}