Iot

Specialist guidance for AWS IoT. Covers IoT Core (MQTT, shadows, rules engine), Greengrass v2 edge compute, fleet provisioning, security, data storage patterns, and fleet management.

Process

1. Identify the IoT workload characteristics: device count, message frequency, payload size, connectivity (always-on vs intermittent), edge processing needs 2. Use the awsknowledge MCP tools (mcp__plugin_aws-dev-toolkit_awsknowledge__aws___search_documentation, mcp__plugin_aws-dev-toolkit_awsknowledge__aws___read_documentation, mcp__plugin_aws-dev-toolkit_awsknowledge__aws___recommend) to verify current IoT Core limits, Greengrass component versions, and service quotas 3. Select the appropriate IoT services using the decision matrix below 4. Design the communication and data ingestion topology (protocols, topics, rules) 5. Configure security (X.509 certificates, IoT policies, fleet provisioning method) 6. Design data storage and analytics pipeline 7. Plan fleet management (jobs, indexing, Device Defender) 8. Recommend operational best practices (monitoring, OTA updates, edge deployments)

IoT Service Selection Decision Matrix

Protocol Selection

MQTT (Default Choice)

Use MQTT for device-to-cloud communication unless there is a specific reason not to. MQTT uses persistent TCP connections with minimal overhead (2-byte header minimum), supports QoS 0 (at most once) and QoS 1 (at least once), and enables server-initiated push to devices via subscriptions.

• QoS 0: Use for high-frequency telemetry where occasional message loss is acceptable (sensor readings every second). Lower overhead because no acknowledgment round-trip.
• QoS 1: Use for commands, configuration changes, and alerts where delivery must be confirmed. The broker retries until PUBACK is received.
• QoS 2 is not supported by AWS IoT Core. If exactly-once semantics are required, implement idempotency in the application layer.

MQTT v5 Features (Prefer When Devices Support It)

• Shared subscriptions: Distribute messages across multiple subscribers for load balancing backend processors, avoiding hot-partition on a single consumer
• Topic aliases: Replace long topic strings with short integer aliases after first publish, reducing per-message overhead for bandwidth-constrained devices
• Message expiry: Set TTL on messages so stale commands are discarded rather than delivered to a device that reconnects hours later
• Session expiry: Control how long the broker holds session state after disconnect, preventing unbounded memory growth from abandoned devices

HTTPS

Use HTTPS only for devices that wake up, send a single reading, and sleep (battery-powered sensors with cellular connectivity). HTTPS does not support subscriptions, so the device cannot receive commands without polling. Every request incurs TLS handshake overhead.

MQTT over WebSocket

Use for browser-based dashboards and mobile apps that need real-time device data. Authenticates with IAM credentials or Cognito identity pools instead of X.509 certificates. Works through corporate proxies and firewalls that block raw TCP on port 8883.

Topic Design

Design topics as a hierarchy with device identity and data type segments. This enables fine-grained IoT policy access control and targeted rules engine subscriptions.

Recommended Structure

{org}/{environment}/{device-type}/{device-id}/{data-category}

Examples:

acme/prod/temperature-sensor/sensor-001/telemetry
acme/prod/temperature-sensor/sensor-001/alerts
acme/prod/temperature-sensor/sensor-001/commands
acme/prod/temperature-sensor/+/telemetry        # Rule subscribes to all sensors

Topic Design Rules

• Include the device ID in the topic so IoT policies can use ${iot:Connection.Thing.ThingName} to restrict each device to its own topics
• Separate telemetry, commands, and alerts into distinct subtopics so rules can target specific data types without parsing payloads
• Use + (single-level) and # (multi-level) wildcards in rules and subscriptions, never in publish topics
• Keep topics under 7 levels deep to stay within IoT Core limits and maintain readability

Basic Ingest

For high-volume telemetry that goes directly to rules engine actions without needing the message broker, use the $aws/rules/<rule-name> topic prefix. Basic Ingest skips the message broker publish cost ($1.00 per million messages), saving significant cost at scale. The tradeoff: messages sent via Basic Ingest cannot be received by other MQTT subscribers.

Device Shadow

Device Shadow maintains a JSON document of desired and reported state for each device. Use shadows when cloud applications need to read or set device state regardless of whether the device is currently connected.

Classic vs Named Shadows

• Classic shadow: One per thing. Use for the primary device state (power on/off, firmware version, connectivity status).
• Named shadows: Up to 10 per thing. Use to separate independent state concerns (e.g., one shadow for configuration, another for diagnostics, another for firmware). Named shadows avoid state conflicts when multiple applications update different aspects of the same device.

Shadow Best Practices

• Keep shadow documents small (<8 KB). Large shadows increase MQTT message size and DynamoDB read/write costs on the shadow service backend.
• Use reported state from the device, desired state from the cloud application. The delta field tells the device what to change.
• Set version-based optimistic locking on updates to prevent stale writes from overwriting newer state.

IoT Rules Engine

The rules engine evaluates SQL statements against incoming MQTT messages and routes matching data to AWS service actions. Every production deployment should have at least one rule for data ingestion and error handling.

Rule SQL Basics

SELECT temperature, humidity, timestamp() as ts, topic(4) as device_id
FROM 'acme/prod/temperature-sensor/+/telemetry'
WHERE temperature > 0 AND temperature < 150

• topic(n) extracts the nth level from the topic string (1-indexed)
• timestamp() adds server-side UTC timestamp
• WHERE clause filters before action execution, reducing downstream processing cost
• Use SELECT * sparingly; extract only the fields needed to minimize action payload size

Action Selection Guide

Error Actions (Always Configure)

Every rule must have an error action. Without one, failed rule actions silently drop data with no notification and no retry. Configure error actions to route failures to S3 or SQS for later reprocessing.

See references/rules-engine-patterns.md for detailed SQL examples and error action configuration.

IoT SiteWise (Industrial IoT)

Use IoT SiteWise instead of raw IoT Core + custom storage when the workload involves industrial equipment with OPC-UA data sources, asset hierarchies, and time-series metrics that need automatic aggregation (min, max, avg, count over time windows).

When to Use IoT SiteWise

• Industrial environments with OPC-UA or Modbus data sources
• Need for asset hierarchy modeling (factory > line > machine > sensor)
• Pre-built portal/dashboard capabilities for operators (SiteWise Monitor)
• Edge data collection and processing via SiteWise Edge gateway

When to Skip IoT SiteWise

• Consumer IoT devices using MQTT natively (use IoT Core directly)
• Custom data formats that do not fit the asset model structure
• Workloads already using Timestream with custom dashboards (Grafana)

IoT Events

Use IoT Events when device telemetry needs to trigger state-machine logic across multiple devices or time windows, and the logic is too complex for simple IoT Rules Engine WHERE clauses.

Detector Models

• Define states (e.g., NORMAL, WARNING, CRITICAL) with transitions based on input conditions
• Each detector instance tracks state for one device independently
• Actions on state entry/exit/transition: send SNS, publish to IoT Core, invoke Lambda, write to DynamoDB
• Use for: equipment health monitoring, multi-sensor correlation, threshold-with-hysteresis alerting (avoid alert flapping by requiring sustained condition before state change)

Fleet Provisioning

Method Selection

Provisioning Best Practices

• Always use a pre-provisioning Lambda hook with fleet provisioning by claim to validate the device identity against an allow-list. Without this, anyone with the claim certificate can provision unlimited devices.
• Scope provisioning templates to create minimal IoT policies. The provisioned policy should grant access only to that device's topics, using ${iot:Connection.Thing.ThingName} policy variables.
• Store device private keys in hardware security modules (HSM) or secure elements when available. Software-stored keys are extractable.

See references/security-provisioning.md for provisioning templates, certificate management, and IoT policy examples.

Security

X.509 Certificates

• Every device must authenticate with a unique X.509 client certificate. Shared certificates across devices make revocation impossible without affecting the entire fleet.
• Use AWS Private CA for production fleets. It provides automated certificate issuance, revocation (CRL), and integration with JITP.
• Rotate certificates before expiry using IoT Jobs to push new certificates and a Lambda to register them. Expired certificates cause immediate connection failure with no grace period.

IoT Policies

IoT policies control what MQTT topics a device can publish/subscribe to and what shadows/jobs it can access. Always use policy variables to scope per-device.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:REGION:ACCOUNT:client/${iot:Connection.Thing.ThingName}"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": "arn:aws:iot:REGION:ACCOUNT:topic/acme/prod/*/${iot:Connection.Thing.ThingName}/*"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": "arn:aws:iot:REGION:ACCOUNT:topicfilter/acme/prod/*/${iot:Connection.Thing.ThingName}/*"
    }
  ]
}

Custom Authorizers

Use custom authorizers when devices cannot use X.509 certificates (e.g., legacy devices with token-based auth or OAuth). The authorizer is a Lambda function that validates the token and returns an IoT policy document. Custom authorizers add latency (Lambda cold start) and cost (per-invocation), so prefer X.509 certificates for new device designs.

Device Defender

• Audit: Scheduled checks for insecure configurations (overly permissive policies, shared certificates, disabled logging). Run at least weekly.
• Detect: Real-time anomaly detection on device metrics (message volume, connection patterns, authorization failures). Alerts when a device deviates from its baseline behavior, indicating compromise or misconfiguration.
• Configure mitigation actions to automatically quarantine compromised devices (move to a restricted thing group with minimal permissions).

Data Storage Patterns

Timestream (Time-Series Telemetry)

• Default choice for telemetry that needs time-range queries (temperature over last 24 hours, average power per hour).
• Automatic tiering: memory store (recent, fast queries) to magnetic store (historical, cheaper).
• Set memory store retention to match your hot-query window (1-24 hours typical). Data beyond this moves to magnetic automatically.
• Cost consideration: Timestream charges per write and per query scan. For very high-frequency telemetry (>1 msg/sec/device across thousands of devices), aggregate at the edge with Greengrass or use Basic Ingest to S3 with Athena for batch queries.

DynamoDB (Device Metadata and State)

• Use for device registry extensions, latest-known state, configuration, and command history.
• Design the partition key as the device ID for even distribution.
• Use TTL to auto-expire old command records and reduce storage cost.
• Do not store raw time-series telemetry in DynamoDB. At 1 msg/sec from 10,000 devices, that is 864 million writes/day, which costs roughly $1,100/day in on-demand WCU charges.

S3 (Bulk Archival)

• Use IoT Rules Engine S3 action with partitioned keys: s3://bucket/year=2026/month=04/day=06/hour=12/device-id.json
• Query archived data with Athena using partition projection for cost-effective ad-hoc analysis.
• Enable S3 Intelligent-Tiering for automatic cost optimization on infrequently accessed telemetry.
• Cheapest option for long-term retention and compliance requirements.

OpenSearch (Search and Analytics)

• Use when operators need full-text search across telemetry fields or real-time dashboards.
• IoT Rules Engine can write directly to OpenSearch Service.
• Cost consideration: OpenSearch clusters run 24/7 with dedicated instances. For intermittent analysis, prefer Athena on S3.

Greengrass v2 (Edge Compute)

When to Use Edge Compute

• Latency: Local control loops that must respond in <100ms (actuator control, safety shutoffs). Cloud round-trip adds 50-200ms minimum.
• Bandwidth: Devices generate more data than the network can upload. Aggregate or filter at the edge, send summaries to cloud.
• Intermittent connectivity: Sites with unreliable internet (remote oil wells, ships, mines). Greengrass buffers data and syncs when connected.
• Local ML inference: Run ML models on edge hardware (image classification, anomaly detection) without sending raw data to cloud.

When to Skip Edge Compute

• Devices with reliable, high-bandwidth connectivity and no latency requirements. Direct MQTT to IoT Core is simpler and eliminates edge infrastructure management.
• Very constrained devices (microcontrollers with <1MB RAM) that cannot run the Greengrass nucleus. Use FreeRTOS with direct IoT Core connectivity instead.

Component Model

Greengrass v2 uses a component model where each capability is a deployable unit (recipe + artifacts). Components can be:

• AWS-provided: Pre-built components for common tasks (stream manager, log manager, MQTT bridge, Docker application manager)
• Custom: Your application logic, packaged as a recipe (YAML/JSON) referencing artifacts (code, binaries, configs)
• Community: Third-party components from the Greengrass component catalog

Stream Manager

Use Stream Manager for reliable edge-to-cloud data transfer. It handles buffering, batching, bandwidth management, and automatic retry. Supports export to Kinesis Data Streams, S3, IoT Analytics, and IoT SiteWise.

• Configure per-stream: storage type (memory or file-system), max size, strategy when full (reject new or overwrite oldest)
• Set bandwidth limits to prevent telemetry uploads from starving control-plane traffic
• Minimum 70 MB RAM overhead for the stream manager component

See references/greengrass-patterns.md for component recipes, deployment configurations, and stream manager setup.

Fleet Management

IoT Jobs (OTA Updates)

• Use Jobs for firmware updates, configuration changes, and certificate rotation across the fleet.
• Continuous jobs: Automatically target new devices added to a thing group. Use for ongoing compliance (all devices in group X must have firmware v2.3+).
• Snapshot jobs: One-time execution against a fixed set of targets.
• Configure rollout rate (max devices per minute) and abort criteria (% failures before halting) to prevent fleet-wide bricking from a bad update.
• Use signed job documents with code signing to prevent tampering.

Fleet Indexing

• Enables SQL-like queries across device registry, shadow, connectivity, and Device Defender violation data.
• Must be explicitly enabled (off by default). Without fleet indexing, you cannot query fleet state at scale.
• Example: thingName:sensor-* AND shadow.reported.firmware:v2.1 AND connectivity.connected:false finds all disconnected sensors on old firmware.
• Use fleet metrics to push aggregated fleet statistics to CloudWatch for dashboards and alarms.

Key Limits (IoT Core)

Anti-Patterns

• Polling instead of MQTT. Devices that HTTP poll for commands waste battery, bandwidth, and IoT Core request costs. A device polling every 5 seconds generates 17,280 requests/day; MQTT keeps a persistent connection with near-zero overhead when idle, and the server pushes commands instantly.
• No error actions on rules. Without an error action, a failed rule action (IAM permission issue, DynamoDB throttle, Lambda error) silently drops the message. There is no retry, no alert, and no way to recover the data. Always route errors to S3 or SQS.
• *Overly permissive IoT policies (iot: on ). A compromised device with `iot: can publish to any topic, read any shadow, and trigger any job. Use policy variables (${iot:Connection.Thing.ThingName}`) to scope each device to its own resources.
• Single MQTT topic for all devices. Publishing everything to devices/telemetry makes it impossible to apply per-device access control, filter rules by device type, or subscribe to a specific device's data. Use hierarchical topics with device identity segments.
• Not using Device Shadow for desired/reported state sync. Without shadows, setting device state requires the device to be online at the exact moment the command is sent. Shadows persist the desired state and deliver it when the device reconnects.
• Storing raw telemetry in DynamoDB. At IoT scale, DynamoDB write costs explode. 10,000 devices at 1 msg/sec = 864M writes/day = ~$1,100/day on-demand. Use Timestream for time-series (10-20x cheaper for write-heavy time-series workloads) or S3 for archival ($0.023/GB/month).
• Ignoring Greengrass for edge preprocessing. Sending raw high-frequency sensor data to the cloud wastes bandwidth and inflates ingestion costs. A Greengrass component that averages 1,000 readings into 1 summary per minute reduces cloud costs by 99.9%.
• Not configuring fleet indexing. Without fleet indexing enabled, you cannot query which devices are running old firmware, which are disconnected, or which have specific shadow states. You are flying blind on fleet health. Enable it proactively.
• Shared X.509 certificates across devices. If one device is compromised, you must revoke the shared certificate, disconnecting all devices that use it. One certificate per device limits the blast radius to a single device.
• No rollout controls on IoT Jobs. Pushing a firmware update to all devices simultaneously risks fleet-wide failure. Always configure max concurrent targets, rollout rate, and abort thresholds (e.g., abort if >5% of devices fail).
• Ignoring Basic Ingest for high-volume telemetry. Standard publish costs $1.00 per million messages. Basic Ingest ($0.00 publish cost, rules actions still charged) saves this entirely for telemetry that only needs to flow to rules engine actions.
• Not setting MQTT session expiry. Default persistent session expiry is 1 hour. Devices that reconnect after longer disconnections lose queued messages. Set session expiry to match the device's expected offline duration (up to 7 days max).

Additional Resources

Reference Files

For detailed operational guidance, consult:

• `references/rules-engine-patterns.md` -- Rule SQL examples for common routing patterns, error action configuration, topic structure best practices, and Basic Ingest setup
• `references/security-provisioning.md` -- X.509 certificate management, fleet provisioning templates (JITP, bulk, by claim), IoT policies with variables, and custom authorizer setup
• `references/greengrass-patterns.md` -- Greengrass v2 component recipes, deployment configurations, stream manager setup, and local MQTT bridge configuration

Related Skills

• `lambda` -- Lambda functions as IoT rule actions and Greengrass components
• `step-functions` -- Orchestrating multi-step device provisioning and remediation workflows
• `dynamodb` -- Device metadata storage design, partition key strategy, TTL configuration
• `s3` -- Telemetry archival, lifecycle policies, Athena integration for batch queries
• `messaging` -- SQS/SNS integration with IoT rules for decoupled processing and alerting
• `observability` -- CloudWatch metrics, alarms, and dashboards for IoT fleet monitoring
• `iam` -- IAM roles for IoT rules engine actions, Greengrass token exchange, and fleet provisioning
• `networking` -- VPC endpoints for IoT Core, private connectivity for Greengrass core devices
• `security-review` -- Security audit of IoT policies, certificate management, and Device Defender configuration

Output Format

When recommending an IoT architecture, include:

Include estimated monthly cost range using the cost-check skill.

Greengrass v2 Patterns

Core Concepts

Greengrass v2 runs on edge devices (called core devices) and uses a component-based architecture. The nucleus is the core runtime. Components are deployable units with a recipe (metadata, dependencies, lifecycle) and artifacts (code, binaries, config).

Installation

# Download and install Greengrass v2 nucleus
# Requires Java 8+ (Corretto recommended) and root/admin access

curl -s https://d2s8p88vqu9w66.cloudfront.net/releases/greengrass-nucleus-latest.zip -o greengrass-nucleus.zip
unzip greengrass-nucleus.zip -d GreengrassInstaller

java -Droot="/greengrass/v2" \
  -Dlog.store=FILE \
  -jar ./GreengrassInstaller/lib/Greengrass.jar \
  --aws-region REGION \
  --thing-name "edge-gateway-001" \
  --thing-group-name "edge-gateways" \
  --thing-policy-name "greengrass-core-policy" \
  --tes-role-name "GreengrassTESRole" \
  --tes-role-alias-name "GreengrassTESRoleAlias" \
  --component-default-user ggc_user:ggc_group \
  --provision true \
  --setup-system-service true

The --provision true flag auto-creates the thing, certificate, and policy in IoT Core. The --setup-system-service true flag registers Greengrass as a systemd service so it starts on boot.

Component Recipes

Custom Telemetry Processor Component

This component reads sensor data, aggregates it, and publishes summaries to IoT Core via the local MQTT bridge.

Recipe (`recipe.yaml`):

---
RecipeFormatVersion: "2020-01-25"
ComponentName: com.acme.telemetry-processor
ComponentVersion: "1.0.0"
ComponentDescription: Aggregates raw sensor telemetry and publishes 1-minute summaries to IoT Core
ComponentPublisher: Acme Corp
ComponentDependencies:
  aws.greengrass.Nucleus:
    VersionRequirement: ">=2.5.0"
    DependencyType: HARD
  aws.greengrass.clientdevices.mqtt.Bridge:
    VersionRequirement: ">=2.2.0"
    DependencyType: HARD
ComponentConfiguration:
  DefaultConfiguration:
    aggregation_interval_seconds: 60
    source_topic: "local/sensors/+/telemetry"
    destination_topic: "acme/prod/edge-gateway-001/aggregated/telemetry"
    accessControl:
      aws.greengrass.ipc.mqttproxy:
        com.acme.telemetry-processor:mqttproxy:1:
          policyDescription: Subscribe to local sensor topics
          operations:
            - "aws.greengrass#SubscribeToIoTCore"
            - "aws.greengrass#PublishToIoTCore"
          resources:
            - "local/sensors/+/telemetry"
            - "acme/prod/edge-gateway-001/aggregated/*"
Manifests:
  - Platform:
      os: linux
    Lifecycle:
      install: "pip3 install -r {artifacts:path}/requirements.txt"
      run:
        script: "python3 {artifacts:path}/telemetry_processor.py"
        RequiresPrivilege: false
    Artifacts:
      - URI: "s3://acme-greengrass-artifacts/telemetry-processor/1.0.0/telemetry_processor.py"
      - URI: "s3://acme-greengrass-artifacts/telemetry-processor/1.0.0/requirements.txt"

ML Inference Component

Runs a pre-trained model at the edge for anomaly detection on sensor data.

Recipe (`recipe.yaml`):

---
RecipeFormatVersion: "2020-01-25"
ComponentName: com.acme.anomaly-detector
ComponentVersion: "1.0.0"
ComponentDescription: Runs anomaly detection ML model on edge sensor data
ComponentPublisher: Acme Corp
ComponentDependencies:
  aws.greengrass.Nucleus:
    VersionRequirement: ">=2.5.0"
    DependencyType: HARD
  aws.greengrass.TokenExchangeService:
    VersionRequirement: ">=2.0.0"
    DependencyType: HARD
ComponentConfiguration:
  DefaultConfiguration:
    model_path: "{artifacts:decompressedPath}/model"
    confidence_threshold: 0.85
    accessControl:
      aws.greengrass.ipc.mqttproxy:
        com.acme.anomaly-detector:mqttproxy:1:
          policyDescription: Subscribe to telemetry, publish anomalies
          operations:
            - "aws.greengrass#SubscribeToIoTCore"
            - "aws.greengrass#PublishToIoTCore"
          resources:
            - "acme/prod/edge-gateway-001/aggregated/telemetry"
            - "acme/prod/edge-gateway-001/anomalies"
Manifests:
  - Platform:
      os: linux
      architecture: aarch64
    Lifecycle:
      install: |
        pip3 install -r {artifacts:path}/requirements.txt
      run:
        script: "python3 {artifacts:path}/anomaly_detector.py --model {configuration:/model_path} --threshold {configuration:/confidence_threshold}"
        RequiresPrivilege: false
    Artifacts:
      - URI: "s3://acme-greengrass-artifacts/anomaly-detector/1.0.0/anomaly_detector.py"
      - URI: "s3://acme-greengrass-artifacts/anomaly-detector/1.0.0/requirements.txt"
      - URI: "s3://acme-greengrass-artifacts/anomaly-detector/1.0.0/model.tar.gz"
        Unarchive: ZIP

Docker Application Component

Runs a containerized application on the Greengrass core device.

Recipe (`recipe.yaml`):

---
RecipeFormatVersion: "2020-01-25"
ComponentName: com.acme.data-dashboard
ComponentVersion: "1.0.0"
ComponentDescription: Local Grafana dashboard for real-time edge data visualization
ComponentPublisher: Acme Corp
ComponentDependencies:
  aws.greengrass.Nucleus:
    VersionRequirement: ">=2.5.0"
    DependencyType: HARD
  aws.greengrass.DockerApplicationManager:
    VersionRequirement: ">=2.0.0"
    DependencyType: HARD
ComponentConfiguration:
  DefaultConfiguration:
    grafana_port: 3000
Manifests:
  - Platform:
      os: linux
    Lifecycle:
      run:
        script: |
          docker run --rm \
            -p {configuration:/grafana_port}:3000 \
            -v /greengrass/v2/work/com.acme.data-dashboard/grafana:/var/lib/grafana \
            grafana/grafana:latest
      shutdown:
        script: "docker stop $(docker ps -q --filter ancestor=grafana/grafana:latest)"
        timeout: 30

Deployment Configuration

Create a Deployment via CLI

aws greengrassv2 create-deployment \
  --target-arn "arn:aws:iot:REGION:ACCOUNT:thinggroup/edge-gateways" \
  --deployment-name "telemetry-processor-v1" \
  --components '{
    "com.acme.telemetry-processor": {
      "componentVersion": "1.0.0",
      "configurationUpdate": {
        "merge": "{\"aggregation_interval_seconds\": 30}"
      }
    },
    "aws.greengrass.clientdevices.mqtt.Bridge": {
      "componentVersion": "2.3.0",
      "configurationUpdate": {
        "merge": "{\"mqttTopicMapping\": {\"telemetryMapping\": {\"topic\": \"local/sensors/+/telemetry\", \"source\": \"LocalMqtt\", \"target\": \"IotCore\"}, \"commandMapping\": {\"topic\": \"acme/prod/+/commands\", \"source\": \"IotCore\", \"target\": \"LocalMqtt\"}}}"
      }
    },
    "aws.greengrass.clientdevices.mqtt.Moquette": {
      "componentVersion": "2.3.0"
    },
    "aws.greengrass.StreamManager": {
      "componentVersion": "2.1.0"
    }
  }' \
  --deployment-policies '{
    "failureHandlingPolicy": "ROLLBACK",
    "componentUpdatePolicy": {
      "timeoutInSeconds": 300,
      "action": "NOTIFY_COMPONENTS"
    }
  }'

Deployment Best Practices

• Always use thing groups as deployment targets, not individual things. This enables automatic deployment to new devices added to the group.
• Set `failureHandlingPolicy` to `ROLLBACK` for production deployments. If any component fails to deploy, the device reverts to the previous configuration instead of running in a degraded state.
• Use `NOTIFY_COMPONENTS` component update policy so running components can gracefully shut down before update, preventing data loss in stream buffers.
• Pin component versions in production deployments. Do not use version ranges (e.g., >=1.0.0) because they may auto-upgrade to untested versions.
• Test deployments on a staging thing group first. Create separate thing groups for staging and production. Deploy to staging, verify via CloudWatch, then deploy to production.

Rollout Configuration

For large fleets, configure deployment rollout to avoid updating all devices simultaneously:

aws greengrassv2 create-deployment \
  --target-arn "arn:aws:iot:REGION:ACCOUNT:thinggroup/edge-gateways" \
  --deployment-name "firmware-update-v2.3" \
  --components '{...}' \
  --iot-job-configuration '{
    "jobExecutionsRolloutConfig": {
      "maximumPerMinute": 10,
      "exponentialRate": {
        "baseRatePerMinute": 5,
        "incrementFactor": 2,
        "rateIncreaseCriteria": {
          "numberOfSucceededThings": 100
        }
      }
    },
    "abortConfig": {
      "criteriaList": [
        {
          "failureType": "FAILED",
          "action": "CANCEL",
          "thresholdPercentage": 5,
          "minNumberOfExecutedThings": 20
        }
      ]
    },
    "timeoutConfig": {
      "inProgressTimeoutInMinutes": 30
    }
  }'

This configuration starts rolling out at 5 devices/minute, doubles the rate after every 100 successes, and aborts the entire deployment if more than 5% of devices fail (after at least 20 have been attempted).

Stream Manager Setup

Configure Stream Manager Component

{
  "aws.greengrass.StreamManager": {
    "componentVersion": "2.1.0",
    "configurationUpdate": {
      "merge": "{\"STREAM_MANAGER_STORE_ROOT_DIR\": \"/greengrass/v2/streams\", \"STREAM_MANAGER_SERVER_PORT\": 8088, \"STREAM_MANAGER_AUTHENTICATE_CLIENT\": true, \"STREAM_MANAGER_EXPORTER_MAX_BANDWIDTH\": 5242880}"
    }
  }
}

Create and Write to a Stream (Python SDK)

from stream_manager import (
    StreamManagerClient,
    MessageStreamDefinition,
    StrategyOnFull,
    ExportDefinition,
    KinesisConfig,
    S3ExportTaskExecutorConfig,
    StatusConfig,
    StatusLevel,
    StatusMessage
)

client = StreamManagerClient()

# Create a stream that exports to Kinesis
client.create_message_stream(
    MessageStreamDefinition(
        name="sensor-telemetry-stream",
        max_size=268435456,  # 256 MB local buffer
        stream_segment_size=16777216,  # 16 MB segments
        strategy_on_full=StrategyOnFull.OverwriteOldestData,
        export_definition=ExportDefinition(
            kinesis=[
                KinesisConfig(
                    identifier="kinesis-export",
                    kinesis_stream_name="iot-telemetry-stream",
                    batch_size=500,
                    batch_interval_millis=5000,
                    priority=10
                )
            ]
        )
    )
)

# Write data to the stream
import json
data = json.dumps({
    "device_id": "sensor-001",
    "temperature": 23.5,
    "humidity": 65.2,
    "timestamp": 1712400000
})
client.append_message("sensor-telemetry-stream", data.encode())

Stream Export Destinations

Stream Manager Best Practices

• Set `strategy_on_full` to `OverwriteOldestData` for telemetry streams where recent data is more valuable than historical. Use RejectNewData for streams where every message must be delivered (alerts, commands).
• Size the local buffer based on expected offline duration. If the site loses connectivity for up to 4 hours and generates 1 MB/min of telemetry, set max_size to at least 240 MB.
• Set batch size and interval together. batch_size=500 with batch_interval_millis=5000 means: send a batch when 500 messages accumulate OR 5 seconds pass, whichever comes first. This balances latency and throughput.
• Monitor stream health via the Greengrass log manager component. Look for ExportTaskFailure log entries.

Local MQTT Bridge Configuration

The MQTT bridge connects local MQTT topics (Moquette broker on the core device) to IoT Core MQTT topics, enabling client devices (sensors, actuators) to communicate with the cloud through the Greengrass core.

Bridge Topic Mapping

{
  "aws.greengrass.clientdevices.mqtt.Bridge": {
    "componentVersion": "2.3.0",
    "configurationUpdate": {
      "merge": "{\"mqttTopicMapping\": {\"sensorTelemetryToCloud\": {\"topic\": \"local/sensors/+/telemetry\", \"source\": \"LocalMqtt\", \"target\": \"IotCore\"}, \"cloudCommandsToLocal\": {\"topic\": \"acme/prod/+/commands\", \"source\": \"IotCore\", \"target\": \"LocalMqtt\"}, \"localDeviceToDevice\": {\"topic\": \"local/actuators/+/control\", \"source\": \"LocalMqtt\", \"target\": \"LocalMqtt\"}}}"
    }
  }
}

Client Device Authentication

Greengrass core authenticates local client devices using their certificates. Configure the client device auth component:

{
  "aws.greengrass.clientdevices.Auth": {
    "componentVersion": "2.4.0",
    "configurationUpdate": {
      "merge": "{\"deviceGroups\": {\"formatVersion\": \"2021-03-05\", \"definitions\": {\"localSensors\": {\"selectionRule\": \"thingName: sensor-*\", \"policyName\": \"localSensorPolicy\"}}, \"policies\": {\"localSensorPolicy\": {\"AllowPublish\": {\"statementDescription\": \"Allow sensors to publish telemetry\", \"operations\": [\"mqtt:publish\"], \"resources\": [\"local/sensors/${iot:clientId}/telemetry\"]}, \"AllowSubscribe\": {\"statementDescription\": \"Allow sensors to receive commands\", \"operations\": [\"mqtt:subscribe\"], \"resources\": [\"local/sensors/${iot:clientId}/commands\"]}}}}}"
    }
  }
}

Components to Deploy Together

For a typical edge gateway setup, deploy these components together:

Monitoring Greengrass Deployments

CloudWatch Logs

Deploy the Log Manager component to ship Greengrass logs to CloudWatch:

{
  "aws.greengrass.LogManager": {
    "componentVersion": "2.3.0",
    "configurationUpdate": {
      "merge": "{\"logsUploaderConfiguration\": {\"systemLogsConfiguration\": {\"uploadToCloudWatch\": true, \"minimumLogLevel\": \"INFO\", \"diskSpaceLimit\": 10, \"diskSpaceLimitUnit\": \"MB\"}, \"componentLogsConfigurationMap\": {\"com.acme.telemetry-processor\": {\"minimumLogLevel\": \"INFO\", \"diskSpaceLimit\": 25, \"diskSpaceLimitUnit\": \"MB\"}}}}"
    }
  }
}

Health Check via CLI

# Check the status of all components on a core device
aws greengrassv2 list-installed-components \
  --core-device-thing-name "edge-gateway-001"

# Check the status of a specific deployment
aws greengrassv2 get-deployment \
  --deployment-id "DEPLOYMENT_ID"

# List core devices and their status
aws greengrassv2 list-core-devices \
  --status HEALTHY

Key Metrics to Monitor

IoT Rules Engine Patterns

Topic Structure Best Practices

Standard Topic Hierarchy

{org}/{env}/{device-type}/{device-id}/{data-category}

Reserved Prefixes

• $aws/things/{thingName}/shadow/ -- Device Shadow MQTT topics (do not use for custom data)
• $aws/things/{thingName}/jobs/ -- IoT Jobs MQTT topics
• $aws/rules/{ruleName} -- Basic Ingest prefix (bypasses message broker)
• $aws/events/ -- Lifecycle events (connect, disconnect, subscribe)

Rule SQL Examples

Route Telemetry to Timestream

SELECT
  topic(4) as device_id,
  topic(3) as device_type,
  temperature,
  humidity,
  pressure,
  timestamp() as time
FROM 'acme/prod/+/+/telemetry'
WHERE temperature IS NOT NULL

Timestream action configuration:

{
  "timestream": {
    "roleArn": "arn:aws:iam::ACCOUNT:role/iot-timestream-role",
    "databaseName": "iot_telemetry",
    "tableName": "sensor_data",
    "dimensions": [
      { "name": "device_id", "value": "${device_id}" },
      { "name": "device_type", "value": "${device_type}" }
    ],
    "timestamp": {
      "value": "${time}",
      "unit": "MILLISECONDS"
    }
  }
}

Route Alerts to Lambda for Enrichment

SELECT
  topic(4) as device_id,
  *
FROM 'acme/prod/+/+/alerts'
WHERE severity >= 3

Use this pattern when alerts need enrichment (look up device owner, location, maintenance history) before sending notifications. The Lambda function queries DynamoDB for device metadata and publishes to SNS.

Write Latest State to DynamoDB

SELECT
  topic(4) as device_id,
  state.reported as reported_state,
  timestamp() as last_updated
FROM '$aws/things/+/shadow/update/documents'

DynamoDBv2 action configuration:

{
  "dynamoDBv2": {
    "roleArn": "arn:aws:iam::ACCOUNT:role/iot-dynamodb-role",
    "putItem": {
      "tableName": "device_state"
    }
  }
}

The DynamoDBv2 action writes the entire SQL SELECT result as a DynamoDB item. The device_id field becomes the partition key (configure the table with device_id as the partition key).

Buffer High-Volume Data in Kinesis

SELECT
  topic(4) as device_id,
  *
FROM 'acme/prod/+/+/telemetry'

Kinesis action configuration:

{
  "kinesis": {
    "roleArn": "arn:aws:iam::ACCOUNT:role/iot-kinesis-role",
    "streamName": "iot-telemetry-stream",
    "partitionKey": "${device_id}"
  }
}

Use Kinesis when downstream consumers (Lambda, Kinesis Data Analytics, custom applications) need to process telemetry in real-time with ordering guarantees per device. The partition key ensures all messages from the same device go to the same shard.

Archive Raw Telemetry to S3

SELECT * FROM 'acme/prod/+/+/telemetry'

S3 action configuration:

{
  "s3": {
    "roleArn": "arn:aws:iam::ACCOUNT:role/iot-s3-role",
    "bucketName": "acme-iot-telemetry-archive",
    "key": "year=${parse_time('yyyy', timestamp())}/month=${parse_time('MM', timestamp())}/day=${parse_time('dd', timestamp())}/${topic(4)}/${timestamp()}.json",
    "cannedAcl": "private"
  }
}

Partition the S3 key by date and device ID for efficient Athena queries with partition projection.

Republish Filtered Data to Another Topic

SELECT
  topic(4) as device_id,
  temperature,
  'HIGH_TEMP' as alert_type
FROM 'acme/prod/temp-sensor/+/telemetry'
WHERE temperature > 100

Republish action configuration:

{
  "republish": {
    "roleArn": "arn:aws:iam::ACCOUNT:role/iot-republish-role",
    "topic": "acme/prod/temp-sensor/${topic(4)}/alerts",
    "qos": 1
  }
}

Use republish to generate derived topics. Downstream applications subscribe to the alert topic without processing raw telemetry.

Send Notifications via SNS

SELECT
  topic(4) as device_id,
  concat('Device ', topic(4), ' battery critically low: ', cast(battery_pct as String), '%') as message
FROM 'acme/prod/+/+/telemetry'
WHERE battery_pct < 10

SNS action configuration:

{
  "sns": {
    "roleArn": "arn:aws:iam::ACCOUNT:role/iot-sns-role",
    "targetArn": "arn:aws:sns:REGION:ACCOUNT:iot-device-alerts",
    "messageFormat": "RAW"
  }
}

Trigger IoT Events Detector

SELECT
  topic(4) as device_id,
  temperature,
  vibration,
  timestamp() as ts
FROM 'acme/prod/motor/+/telemetry'

IoT Events action configuration:

{
  "iotEvents": {
    "roleArn": "arn:aws:iam::ACCOUNT:role/iot-events-role",
    "inputName": "motor_telemetry",
    "messageId": "${newuuid()}"
  }
}

Error Action Configuration

Error Action to S3 (Recommended Default)

Every rule should have an error action. S3 is the cheapest destination for error capture and allows batch reprocessing later.

{
  "errorAction": {
    "s3": {
      "roleArn": "arn:aws:iam::ACCOUNT:role/iot-error-action-role",
      "bucketName": "acme-iot-rule-errors",
      "key": "errors/${ruleName}/${parse_time('yyyy/MM/dd/HH', timestamp())}/${newuuid()}.json",
      "cannedAcl": "private"
    }
  }
}

The error payload includes:

• ruleName: Which rule failed
• topic: The original MQTT topic
• clientId: The device that published
• base64OriginalPayload: The original message (base64 encoded)
• failures[]: Array of failed actions with error messages

Error Action to SQS (For Automated Reprocessing)

Use SQS when you want a Lambda function to automatically retry failed messages:

{
  "errorAction": {
    "sqs": {
      "roleArn": "arn:aws:iam::ACCOUNT:role/iot-error-sqs-role",
      "queueUrl": "https://sqs.REGION.amazonaws.com/ACCOUNT/iot-rule-errors",
      "useBase64": true
    }
  }
}

Wire a Lambda function to the SQS queue to inspect the failure reason, fix the issue (e.g., create a missing DynamoDB table, fix IAM permissions), and republish the original message.

Error Action to CloudWatch Logs (For Debugging)

Use during development or when you need searchable error logs:

{
  "errorAction": {
    "cloudwatchLogs": {
      "roleArn": "arn:aws:iam::ACCOUNT:role/iot-error-cw-role",
      "logGroupName": "/aws/iot/rules/errors",
      "batchMode": true
    }
  }
}

Basic Ingest Setup

When to Use Basic Ingest

Use Basic Ingest for telemetry that only needs rules engine processing (not consumed by other MQTT subscribers). It eliminates the message broker publish charge ($1.00 per million messages).

How It Works

Devices publish to $aws/rules/<rule-name>/<custom-topic> instead of the custom topic directly. The message goes straight to the named rule, bypassing the message broker.

Example

Device publishes to:

$aws/rules/telemetry-to-timestream/acme/prod/temp-sensor/sensor-001/telemetry

The rule SQL references the custom topic portion:

SELECT
  topic(4) as device_id,
  temperature,
  humidity
FROM '$aws/rules/telemetry-to-timestream/acme/prod/+/+/telemetry'

Note: topic() function indexes from the custom topic portion, not from $aws/rules/rule-name.

Basic Ingest Limitations

• Messages are not published to the MQTT broker, so other subscribers cannot receive them
• Cannot use MQTT retained messages with Basic Ingest
• The rule name in the topic must match an existing rule
• Still charges for rule actions (Lambda invocations, Timestream writes, etc.)

IAM Role for Rules Engine

Every rule action needs an IAM role that grants the rules engine permission to invoke the target service. Use a single role per rule (not per action) with least-privilege permissions.

Example: Timestream + S3 Error Action Role

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "timestream:WriteRecords",
        "timestream:DescribeEndpoints"
      ],
      "Resource": "arn:aws:timestream:REGION:ACCOUNT:database/iot_telemetry/table/sensor_data"
    },
    {
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::acme-iot-rule-errors/*"
    }
  ]
}

Trust policy:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "iot.amazonaws.com"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "aws:SourceAccount": "ACCOUNT_ID"
        }
      }
    }
  ]
}

Always include the aws:SourceAccount condition to prevent cross-account confused deputy attacks.

Monitoring Rules

CloudWatch Metrics to Alarm On

Enable IoT Core logging (set to INFO for development, ERROR for production) to get detailed rule execution logs in CloudWatch Logs at /aws/iot/logs.

IoT Security and Provisioning

X.509 Certificate Management

Certificate Hierarchy

AWS Private CA (Root CA)
  └── Subordinate CA (per environment or region)
       └── Device Certificates (one per device)

Use a subordinate CA per environment (prod, staging) so you can revoke an entire environment's CA without affecting others.

Register a CA Certificate

# 1. Generate the CA certificate (or use AWS Private CA)
aws iot register-ca-certificate \
  --ca-certificate file://ca-cert.pem \
  --verification-certificate file://verification-cert.pem \
  --set-as-active \
  --allow-auto-registration

The --allow-auto-registration flag enables JITP: any device presenting a certificate signed by this CA will be automatically registered on first connection.

Register a Device Certificate

# Register and activate a specific device certificate
aws iot register-certificate \
  --certificate-pem file://device-cert.pem \
  --ca-certificate-pem file://ca-cert.pem \
  --set-as-active

# Attach the certificate to a thing
aws iot attach-thing-principal \
  --thing-name "sensor-001" \
  --principal "arn:aws:iot:REGION:ACCOUNT:cert/CERT_ID"

# Attach an IoT policy to the certificate
aws iot attach-policy \
  --policy-name "sensor-telemetry-policy" \
  --target "arn:aws:iot:REGION:ACCOUNT:cert/CERT_ID"

Certificate Rotation

Rotate certificates before expiry using IoT Jobs. The process:

1. Generate new certificate (via AWS Private CA or your PKI) 2. Create an IoT Job that pushes the new certificate to the device 3. Device stores new certificate, acknowledges the job 4. Lambda function registers the new certificate and deactivates the old one 5. Device reconnects with the new certificate 6. After confirmation, revoke and delete the old certificate

# Deactivate old certificate
aws iot update-certificate \
  --certificate-id OLD_CERT_ID \
  --new-status INACTIVE

# Delete after grace period
aws iot delete-certificate \
  --certificate-id OLD_CERT_ID \
  --force-delete

Certificate Revocation

# Revoke a compromised certificate immediately
aws iot update-certificate \
  --certificate-id COMPROMISED_CERT_ID \
  --new-status REVOKED

# Move the device to a quarantine thing group
aws iot add-thing-to-thing-group \
  --thing-name "compromised-device" \
  --thing-group-name "quarantine"

The quarantine thing group should have a group policy that denies all actions except connecting and receiving new certificates (for remediation).

Fleet Provisioning Templates

Just-in-Time Provisioning (JITP) Template

Register this template with your CA certificate. When a device connects with a certificate signed by this CA, IoT Core auto-creates the thing and attaches the policy.

{
  "templateBody": {
    "Parameters": {
      "AWS::IoT::Certificate::CommonName": { "Type": "String" },
      "AWS::IoT::Certificate::Id": { "Type": "String" }
    },
    "Resources": {
      "thing": {
        "Type": "AWS::IoT::Thing",
        "Properties": {
          "ThingName": { "Ref": "AWS::IoT::Certificate::CommonName" },
          "ThingGroups": ["auto-provisioned"],
          "AttributePayload": {
            "provisioning_method": "JITP",
            "provisioned_at": "{{timestamp}}"
          }
        }
      },
      "certificate": {
        "Type": "AWS::IoT::Certificate",
        "Properties": {
          "CertificateId": { "Ref": "AWS::IoT::Certificate::Id" },
          "Status": "ACTIVE"
        }
      },
      "policy": {
        "Type": "AWS::IoT::Policy",
        "Properties": {
          "PolicyName": "device-scoped-policy"
        }
      }
    }
  }
}

Fleet Provisioning by Claim Template

For devices without pre-installed unique certificates. The device uses a shared claim certificate to request a unique identity.

{
  "Parameters": {
    "SerialNumber": { "Type": "String" },
    "DeviceType": { "Type": "String" }
  },
  "Resources": {
    "thing": {
      "Type": "AWS::IoT::Thing",
      "Properties": {
        "ThingName": { "Fn::Join": ["-", [{ "Ref": "DeviceType" }, { "Ref": "SerialNumber" }]] },
        "ThingGroups": [{ "Ref": "DeviceType" }],
        "AttributePayload": {
          "serial_number": { "Ref": "SerialNumber" },
          "device_type": { "Ref": "DeviceType" }
        }
      },
      "OverrideSettings": {
        "ThingGroups": "MERGE"
      }
    },
    "certificate": {
      "Type": "AWS::IoT::Certificate",
      "Properties": {
        "CertificateId": { "Ref": "AWS::IoT::Certificate::Id" },
        "Status": "ACTIVE"
      }
    },
    "policy": {
      "Type": "AWS::IoT::Policy",
      "Properties": {
        "PolicyName": "device-scoped-policy"
      }
    }
  }
}

Create the Provisioning Template

# Create the provisioning template
aws iot create-provisioning-template \
  --template-name "sensor-provisioning" \
  --template-body file://provisioning-template.json \
  --provisioning-role-arn "arn:aws:iam::ACCOUNT:role/iot-provisioning-role" \
  --enabled \
  --pre-provisioning-hook '{
    "targetArn": "arn:aws:lambda:REGION:ACCOUNT:function:validate-device",
    "payloadVersion": "2020-04-01"
  }'

Pre-Provisioning Hook Lambda

This Lambda validates the device identity before allowing provisioning. Critical for fleet provisioning by claim to prevent unauthorized device registration.

import json
import boto3

dynamodb = boto3.resource('dynamodb')
allow_list = dynamodb.Table('device-allow-list')

def handler(event, context):
    serial_number = event['parameters']['SerialNumber']
    device_type = event['parameters']['DeviceType']

    # Check if the device serial number is in the allow list
    response = allow_list.get_item(
        Key={'serial_number': serial_number}
    )

    if 'Item' not in response:
        return {
            'allowProvisioning': False
        }

    # Verify the device type matches the expected type
    if response['Item'].get('device_type') != device_type:
        return {
            'allowProvisioning': False
        }

    return {
        'allowProvisioning': True
    }

Claim Certificate Policy (Minimal)

The claim certificate should only have permission to connect and call the fleet provisioning APIs. Nothing else.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Publish", "iot:Receive"],
      "Resource": [
        "arn:aws:iot:REGION:ACCOUNT:topic/$aws/certificates/create/*",
        "arn:aws:iot:REGION:ACCOUNT:topic/$aws/provisioning-templates/sensor-provisioning/provision/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": [
        "arn:aws:iot:REGION:ACCOUNT:topicfilter/$aws/certificates/create/*",
        "arn:aws:iot:REGION:ACCOUNT:topicfilter/$aws/provisioning-templates/sensor-provisioning/provision/*"
      ]
    }
  ]
}

IoT Policies with Variables

Per-Device Scoped Policy (Production Default)

This policy uses ${iot:Connection.Thing.ThingName} to dynamically scope permissions to the connected device's own resources.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:REGION:ACCOUNT:client/${iot:Connection.Thing.ThingName}",
      "Condition": {
        "Bool": { "iot:Connection.Thing.IsAttached": "true" }
      }
    },
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": [
        "arn:aws:iot:REGION:ACCOUNT:topic/acme/prod/*/${iot:Connection.Thing.ThingName}/telemetry",
        "arn:aws:iot:REGION:ACCOUNT:topic/acme/prod/*/${iot:Connection.Thing.ThingName}/alerts",
        "arn:aws:iot:REGION:ACCOUNT:topic/acme/prod/*/${iot:Connection.Thing.ThingName}/status"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": [
        "arn:aws:iot:REGION:ACCOUNT:topicfilter/acme/prod/*/${iot:Connection.Thing.ThingName}/commands",
        "arn:aws:iot:REGION:ACCOUNT:topicfilter/$aws/things/${iot:Connection.Thing.ThingName}/shadow/*",
        "arn:aws:iot:REGION:ACCOUNT:topicfilter/$aws/things/${iot:Connection.Thing.ThingName}/jobs/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "iot:Receive",
      "Resource": [
        "arn:aws:iot:REGION:ACCOUNT:topic/acme/prod/*/${iot:Connection.Thing.ThingName}/commands",
        "arn:aws:iot:REGION:ACCOUNT:topic/$aws/things/${iot:Connection.Thing.ThingName}/shadow/*",
        "arn:aws:iot:REGION:ACCOUNT:topic/$aws/things/${iot:Connection.Thing.ThingName}/jobs/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": [
        "iot:GetThingShadow",
        "iot:UpdateThingShadow"
      ],
      "Resource": "arn:aws:iot:REGION:ACCOUNT:thing/${iot:Connection.Thing.ThingName}"
    }
  ]
}

Key Policy Variables

Policy Best Practices

• Always require iot:Connection.Thing.IsAttached condition on the Connect action. Without it, a certificate not attached to any thing can still connect.
• Separate Publish and Subscribe/Receive permissions. Devices should publish to telemetry/alerts topics but only subscribe to commands/shadow/jobs topics.
• Never use wildcards in the account or region segments of ARNs.
• Test policies using the IoT Policy Simulator before deploying to production devices.

Custom Authorizer Setup

Use custom authorizers when devices authenticate with tokens instead of X.509 certificates (legacy protocols, shared infrastructure, third-party devices).

Create the Authorizer Lambda

import json

def handler(event, context):
    token = event.get('token', '')
    # event also contains: protocolData, connectionMetadata

    # Validate the token (check against your auth system)
    if not validate_token(token):
        raise Exception('Unauthorized')

    # Extract device identity from token
    device_id = extract_device_id(token)

    return {
        'isAuthenticated': True,
        'principalId': device_id,
        'disconnectAfterInSeconds': 86400,
        'refreshAfterInSeconds': 3600,
        'policyDocuments': [
            json.dumps({
                'Version': '2012-10-17',
                'Statement': [
                    {
                        'Effect': 'Allow',
                        'Action': 'iot:Connect',
                        'Resource': f'arn:aws:iot:REGION:ACCOUNT:client/{device_id}'
                    },
                    {
                        'Effect': 'Allow',
                        'Action': ['iot:Publish', 'iot:Subscribe', 'iot:Receive'],
                        'Resource': f'arn:aws:iot:REGION:ACCOUNT:topic/acme/prod/*/{device_id}/*'
                    }
                ]
            })
        ]
    }

def validate_token(token):
    # Implement your token validation logic
    # Check JWT signature, expiry, issuer, etc.
    pass

def extract_device_id(token):
    # Extract device identity from the token payload
    pass

Register the Custom Authorizer

# Create the authorizer
aws iot create-authorizer \
  --authorizer-name "token-authorizer" \
  --authorizer-function-arn "arn:aws:lambda:REGION:ACCOUNT:function:iot-custom-auth" \
  --token-key-name "x-auth-token" \
  --token-signing-public-keys "FirstKey=file://public-key.pem" \
  --signing-disabled \
  --status ACTIVE

# Grant IoT permission to invoke the Lambda
aws lambda add-permission \
  --function-name iot-custom-auth \
  --principal iot.amazonaws.com \
  --statement-id iot-invoke \
  --action lambda:InvokeFunction \
  --source-arn "arn:aws:iot:REGION:ACCOUNT:authorizer/token-authorizer"

Custom Authorizer Caching

• Enable caching to reduce Lambda invocations and latency. Set refreshAfterInSeconds in the Lambda response.
• Cache TTL should balance security (shorter = faster revocation) and cost (longer = fewer Lambda invocations).
• For production: 300-3600 seconds is typical. For high-security environments: 60-300 seconds.

Device Defender Configuration

Enable Audit

# Create an audit role
# (IAM role with iot:DescribeThing, iot:ListThings, etc.)

# Enable audit checks
aws iot update-account-audit-configuration \
  --audit-check-configurations '{
    "DEVICE_CERTIFICATE_SHARED_CHECK": { "enabled": true },
    "CA_CERTIFICATE_EXPIRING_CHECK": { "enabled": true },
    "IOT_POLICY_OVERLY_PERMISSIVE_CHECK": { "enabled": true },
    "LOGGING_DISABLED_CHECK": { "enabled": true },
    "REVOKED_CA_CERTIFICATE_STILL_ACTIVE_CHECK": { "enabled": true },
    "UNAUTHENTICATED_COGNITO_ROLE_OVERLY_PERMISSIVE_CHECK": { "enabled": true }
  }' \
  --role-arn "arn:aws:iam::ACCOUNT:role/iot-device-defender-audit-role"

# Schedule weekly audit
aws iot create-scheduled-audit \
  --scheduled-audit-name "weekly-security-audit" \
  --frequency WEEKLY \
  --day-of-week MON \
  --target-check-names \
    DEVICE_CERTIFICATE_SHARED_CHECK \
    CA_CERTIFICATE_EXPIRING_CHECK \
    IOT_POLICY_OVERLY_PERMISSIVE_CHECK \
    LOGGING_DISABLED_CHECK

Enable Detect (Anomaly Detection)

# Create a security profile for all devices
aws iot create-security-profile \
  --security-profile-name "baseline-behavior" \
  --behaviors '[
    {
      "name": "message-volume",
      "metric": { "name": "aws:num-messages-sent" },
      "criteria": {
        "comparisonOperator": "less-than",
        "value": { "count": 1000 },
        "durationInSeconds": 300
      }
    },
    {
      "name": "auth-failures",
      "metric": { "name": "aws:num-authorization-failures" },
      "criteria": {
        "comparisonOperator": "less-than",
        "value": { "count": 5 },
        "durationInSeconds": 300
      }
    },
    {
      "name": "connection-attempts",
      "metric": { "name": "aws:num-connection-attempts" },
      "criteria": {
        "comparisonOperator": "less-than",
        "value": { "count": 10 },
        "durationInSeconds": 300
      }
    }
  ]' \
  --alert-targets '{
    "SNS": {
      "alertTargetArn": "arn:aws:sns:REGION:ACCOUNT:iot-security-alerts",
      "roleArn": "arn:aws:iam::ACCOUNT:role/iot-defender-sns-role"
    }
  }'

# Attach the security profile to all things
aws iot attach-security-profile \
  --security-profile-name "baseline-behavior" \
  --security-profile-target-arn "arn:aws:iot:REGION:ACCOUNT:all/things"

Mitigation Actions

# Create a mitigation action to quarantine compromised devices
aws iot create-mitigation-action \
  --action-name "quarantine-device" \
  --action-params '{
    "addThingsToThingGroupParams": {
      "thingGroupNames": ["quarantine"],
      "overrideDynamicGroups": true
    }
  }' \
  --role-arn "arn:aws:iam::ACCOUNT:role/iot-mitigation-role"

The quarantine thing group should have a restrictive group policy that: 1. Allows only iot:Connect (so the device can be reached for remediation) 2. Allows subscribe/receive only on the jobs topic (to receive certificate rotation or firmware update) 3. Denies all publish except to a quarantine status topic