# AWS IoT Construct Library
<!--BEGIN STABILITY BANNER-->---

> The APIs of higher level constructs in this module are experimental and under active development.
> They are subject to non-backward compatible changes or removal in any future version. These are
> not subject to the [Semantic Versioning](https://semver.org/) model and breaking changes will be
> announced in the release notes. This means that while you may use them, you may need to update
> your source code when upgrading to a newer version of this package.
---
<!--END STABILITY BANNER-->
AWS IoT Core lets you connect billions of IoT devices and route trillions of
messages to AWS services without managing infrastructure.
## `TopicRule`
Create a topic rule that give your devices the ability to interact with AWS services.
You can create a topic rule with an action that invoke the Lambda action as following:
```python
func = lambda_.Function(self, "MyFunction",
runtime=lambda_.Runtime.NODEJS_LATEST,
handler="index.handler",
code=lambda_.Code.from_inline("""
exports.handler = (event) => {
console.log("It is test for lambda action of AWS IoT Rule.", event);
};""")
)
iot.TopicRule(self, "TopicRule",
topic_rule_name="MyTopicRule", # optional
description="invokes the lambda function", # optional
sql=iot.IotSql.from_string_as_ver20160323("SELECT topic(2) as device_id, timestamp() as timestamp FROM 'device/+/data'"),
actions=[actions.LambdaFunctionAction(func)]
)
```
Or, you can add an action after constructing the `TopicRule` instance as following:
```python
# func: lambda.Function
topic_rule = iot.TopicRule(self, "TopicRule",
sql=iot.IotSql.from_string_as_ver20160323("SELECT topic(2) as device_id, timestamp() as timestamp FROM 'device/+/data'")
)
topic_rule.add_action(actions.LambdaFunctionAction(func))
```
You can also supply `errorAction` as following,
and the IoT Rule will trigger it if a rule's action is unable to perform:
```python
import aws_cdk.aws_logs as logs
log_group = logs.LogGroup(self, "MyLogGroup")
iot.TopicRule(self, "TopicRule",
sql=iot.IotSql.from_string_as_ver20160323("SELECT topic(2) as device_id, timestamp() as timestamp FROM 'device/+/data'"),
error_action=actions.CloudWatchLogsAction(log_group)
)
```
If you wanna make the topic rule disable, add property `enabled: false` as following:
```python
iot.TopicRule(self, "TopicRule",
sql=iot.IotSql.from_string_as_ver20160323("SELECT topic(2) as device_id, timestamp() as timestamp FROM 'device/+/data'"),
enabled=False
)
```
See also [@aws-cdk/aws-iot-actions-alpha](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-iot-actions-alpha-readme.html) for other actions.
## Logging
AWS IoT provides a [logging feature](https://docs.aws.amazon.com/iot/latest/developerguide/configure-logging.html) that allows you to monitor and log AWS IoT activity.
You can enable IoT logging with the following code:
```python
iot.Logging(self, "Logging",
log_level=iot.LogLevel.INFO
)
```
**Note**: All logs are forwarded to the `AWSIotLogsV2` log group in CloudWatch.
## Audit
An [AWS IoT Device Defender audit looks](https://docs.aws.amazon.com/iot-device-defender/latest/devguide/device-defender-audit.html) at account- and device-related settings and policies to ensure security measures are in place.
An audit can help you detect any drifts from security best practices or access policies.
### Account Audit Configuration
The IoT audit includes [various audit checks](https://docs.aws.amazon.com/iot-device-defender/latest/devguide/device-defender-audit-checks.html), and it is necessary to configure settings to enable those checks.
You can enable an account audit configuration with the following code:
```python
# Audit notification are sent to the SNS topic
# target_topic: sns.ITopic
iot.AccountAuditConfiguration(self, "AuditConfiguration",
target_topic=target_topic
)
```
By default, all audit checks are enabled, but it is also possible to enable only specific audit checks.
```python
iot.AccountAuditConfiguration(self, "AuditConfiguration",
check_configuration=iot.CheckConfiguration(
# enabled
authenticated_cognito_role_overly_permissive_check=True,
# enabled by default
ca_certificate_expiring_check=undefined,
# disabled
ca_certificate_key_quality_check=False,
conflicting_client_ids_check=False,
device_certificate_age_check=False,
device_certificate_expiring_check=False,
device_certificate_key_quality_check=False,
device_certificate_shared_check=False,
intermediate_ca_revoked_for_active_device_certificates_check=False,
io_tPolicy_potential_mis_configuration_check=False,
iot_policy_overly_permissive_check=False,
iot_role_alias_allows_access_to_unused_services_check=False,
iot_role_alias_overly_permissive_check=False,
logging_disabled_check=False,
revoked_ca_certificate_still_active_check=False,
revoked_device_certificate_still_active_check=False,
unauthenticated_cognito_role_overly_permissive_check=False
)
)
```
To configure [the device certificate age check](https://docs.aws.amazon.com/iot-device-defender/latest/devguide/device-certificate-age-check.html), you can specify the duration for the check:
```python
from aws_cdk import Duration
iot.AccountAuditConfiguration(self, "AuditConfiguration",
check_configuration=iot.CheckConfiguration(
device_certificate_age_check=True,
# The default value is 365 days
# Valid values range from 30 days (minimum) to 3652 days (10 years, maximum)
device_certificate_age_check_duration=Duration.days(365)
)
)
```
### Scheduled Audit
You can create a [scheduled audit](https://docs.aws.amazon.com/iot-device-defender/latest/devguide/AuditCommands.html#device-defender-AuditCommandsManageSchedules) that is run at a specified time interval. Checks must be enabled for your account by creating `AccountAuditConfiguration`.
```python
# config: iot.AccountAuditConfiguration
# Daily audit
daily_audit = iot.ScheduledAudit(self, "DailyAudit",
account_audit_configuration=config,
frequency=iot.Frequency.DAILY,
audit_checks=[iot.AuditCheck.AUTHENTICATED_COGNITO_ROLE_OVERLY_PERMISSIVE_CHECK
]
)
# Weekly audit
weekly_audit = iot.ScheduledAudit(self, "WeeklyAudit",
account_audit_configuration=config,
frequency=iot.Frequency.WEEKLY,
day_of_week=iot.DayOfWeek.SUNDAY,
audit_checks=[iot.AuditCheck.CA_CERTIFICATE_EXPIRING_CHECK
]
)
# Monthly audit
monthly_audit = iot.ScheduledAudit(self, "MonthlyAudit",
account_audit_configuration=config,
frequency=iot.Frequency.MONTHLY,
day_of_month=iot.DayOfMonth.of(1),
audit_checks=[iot.AuditCheck.CA_CERTIFICATE_KEY_QUALITY_CHECK
]
)
```
Raw data
{
"_id": null,
"home_page": "https://github.com/aws/aws-cdk",
"name": "aws-cdk.aws-iot-alpha",
"maintainer": null,
"docs_url": null,
"requires_python": "~=3.9",
"maintainer_email": null,
"keywords": null,
"author": "Amazon Web Services",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/e5/4f/53777c7799c5d884d14a25c6d7ef4d59ad9841aba3ce3f1314b291a5e6b1/aws_cdk_aws_iot_alpha-2.214.0a0.tar.gz",
"platform": null,
"description": "# AWS IoT Construct Library\n\n<!--BEGIN STABILITY BANNER-->---\n\n\n\n\n> The APIs of higher level constructs in this module are experimental and under active development.\n> They are subject to non-backward compatible changes or removal in any future version. These are\n> not subject to the [Semantic Versioning](https://semver.org/) model and breaking changes will be\n> announced in the release notes. This means that while you may use them, you may need to update\n> your source code when upgrading to a newer version of this package.\n\n---\n<!--END STABILITY BANNER-->\n\nAWS IoT Core lets you connect billions of IoT devices and route trillions of\nmessages to AWS services without managing infrastructure.\n\n## `TopicRule`\n\nCreate a topic rule that give your devices the ability to interact with AWS services.\nYou can create a topic rule with an action that invoke the Lambda action as following:\n\n```python\nfunc = lambda_.Function(self, \"MyFunction\",\n runtime=lambda_.Runtime.NODEJS_LATEST,\n handler=\"index.handler\",\n code=lambda_.Code.from_inline(\"\"\"\n exports.handler = (event) => {\n console.log(\"It is test for lambda action of AWS IoT Rule.\", event);\n };\"\"\")\n)\n\niot.TopicRule(self, \"TopicRule\",\n topic_rule_name=\"MyTopicRule\", # optional\n description=\"invokes the lambda function\", # optional\n sql=iot.IotSql.from_string_as_ver20160323(\"SELECT topic(2) as device_id, timestamp() as timestamp FROM 'device/+/data'\"),\n actions=[actions.LambdaFunctionAction(func)]\n)\n```\n\nOr, you can add an action after constructing the `TopicRule` instance as following:\n\n```python\n# func: lambda.Function\n\n\ntopic_rule = iot.TopicRule(self, \"TopicRule\",\n sql=iot.IotSql.from_string_as_ver20160323(\"SELECT topic(2) as device_id, timestamp() as timestamp FROM 'device/+/data'\")\n)\ntopic_rule.add_action(actions.LambdaFunctionAction(func))\n```\n\nYou can also supply `errorAction` as following,\nand the IoT Rule will trigger it if a rule's action is unable to perform:\n\n```python\nimport aws_cdk.aws_logs as logs\n\n\nlog_group = logs.LogGroup(self, \"MyLogGroup\")\n\niot.TopicRule(self, \"TopicRule\",\n sql=iot.IotSql.from_string_as_ver20160323(\"SELECT topic(2) as device_id, timestamp() as timestamp FROM 'device/+/data'\"),\n error_action=actions.CloudWatchLogsAction(log_group)\n)\n```\n\nIf you wanna make the topic rule disable, add property `enabled: false` as following:\n\n```python\niot.TopicRule(self, \"TopicRule\",\n sql=iot.IotSql.from_string_as_ver20160323(\"SELECT topic(2) as device_id, timestamp() as timestamp FROM 'device/+/data'\"),\n enabled=False\n)\n```\n\nSee also [@aws-cdk/aws-iot-actions-alpha](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-iot-actions-alpha-readme.html) for other actions.\n\n## Logging\n\nAWS IoT provides a [logging feature](https://docs.aws.amazon.com/iot/latest/developerguide/configure-logging.html) that allows you to monitor and log AWS IoT activity.\n\nYou can enable IoT logging with the following code:\n\n```python\niot.Logging(self, \"Logging\",\n log_level=iot.LogLevel.INFO\n)\n```\n\n**Note**: All logs are forwarded to the `AWSIotLogsV2` log group in CloudWatch.\n\n## Audit\n\nAn [AWS IoT Device Defender audit looks](https://docs.aws.amazon.com/iot-device-defender/latest/devguide/device-defender-audit.html) at account- and device-related settings and policies to ensure security measures are in place.\nAn audit can help you detect any drifts from security best practices or access policies.\n\n### Account Audit Configuration\n\nThe IoT audit includes [various audit checks](https://docs.aws.amazon.com/iot-device-defender/latest/devguide/device-defender-audit-checks.html), and it is necessary to configure settings to enable those checks.\n\nYou can enable an account audit configuration with the following code:\n\n```python\n# Audit notification are sent to the SNS topic\n# target_topic: sns.ITopic\n\n\niot.AccountAuditConfiguration(self, \"AuditConfiguration\",\n target_topic=target_topic\n)\n```\n\nBy default, all audit checks are enabled, but it is also possible to enable only specific audit checks.\n\n```python\niot.AccountAuditConfiguration(self, \"AuditConfiguration\",\n check_configuration=iot.CheckConfiguration(\n # enabled\n authenticated_cognito_role_overly_permissive_check=True,\n # enabled by default\n ca_certificate_expiring_check=undefined,\n # disabled\n ca_certificate_key_quality_check=False,\n conflicting_client_ids_check=False,\n device_certificate_age_check=False,\n device_certificate_expiring_check=False,\n device_certificate_key_quality_check=False,\n device_certificate_shared_check=False,\n intermediate_ca_revoked_for_active_device_certificates_check=False,\n io_tPolicy_potential_mis_configuration_check=False,\n iot_policy_overly_permissive_check=False,\n iot_role_alias_allows_access_to_unused_services_check=False,\n iot_role_alias_overly_permissive_check=False,\n logging_disabled_check=False,\n revoked_ca_certificate_still_active_check=False,\n revoked_device_certificate_still_active_check=False,\n unauthenticated_cognito_role_overly_permissive_check=False\n )\n)\n```\n\nTo configure [the device certificate age check](https://docs.aws.amazon.com/iot-device-defender/latest/devguide/device-certificate-age-check.html), you can specify the duration for the check:\n\n```python\nfrom aws_cdk import Duration\n\n\niot.AccountAuditConfiguration(self, \"AuditConfiguration\",\n check_configuration=iot.CheckConfiguration(\n device_certificate_age_check=True,\n # The default value is 365 days\n # Valid values range from 30 days (minimum) to 3652 days (10 years, maximum)\n device_certificate_age_check_duration=Duration.days(365)\n )\n)\n```\n\n### Scheduled Audit\n\nYou can create a [scheduled audit](https://docs.aws.amazon.com/iot-device-defender/latest/devguide/AuditCommands.html#device-defender-AuditCommandsManageSchedules) that is run at a specified time interval. Checks must be enabled for your account by creating `AccountAuditConfiguration`.\n\n```python\n# config: iot.AccountAuditConfiguration\n\n\n# Daily audit\ndaily_audit = iot.ScheduledAudit(self, \"DailyAudit\",\n account_audit_configuration=config,\n frequency=iot.Frequency.DAILY,\n audit_checks=[iot.AuditCheck.AUTHENTICATED_COGNITO_ROLE_OVERLY_PERMISSIVE_CHECK\n ]\n)\n\n# Weekly audit\nweekly_audit = iot.ScheduledAudit(self, \"WeeklyAudit\",\n account_audit_configuration=config,\n frequency=iot.Frequency.WEEKLY,\n day_of_week=iot.DayOfWeek.SUNDAY,\n audit_checks=[iot.AuditCheck.CA_CERTIFICATE_EXPIRING_CHECK\n ]\n)\n\n# Monthly audit\nmonthly_audit = iot.ScheduledAudit(self, \"MonthlyAudit\",\n account_audit_configuration=config,\n frequency=iot.Frequency.MONTHLY,\n day_of_month=iot.DayOfMonth.of(1),\n audit_checks=[iot.AuditCheck.CA_CERTIFICATE_KEY_QUALITY_CHECK\n ]\n)\n```\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "The CDK Construct Library for AWS::IoT",
"version": "2.214.0a0",
"project_urls": {
"Homepage": "https://github.com/aws/aws-cdk",
"Source": "https://github.com/aws/aws-cdk.git"
},
"split_keywords": [],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "6d2e447da3071db4562374b32e84158ce943439fddd6414fdb3178ffb436256a",
"md5": "0254a745afab8747f7b96b44ec1dc5d3",
"sha256": "60f6c91a117d0085e4d22abe32e88ce62e6d6647129ab09afe70f88f89247346"
},
"downloads": -1,
"filename": "aws_cdk_aws_iot_alpha-2.214.0a0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0254a745afab8747f7b96b44ec1dc5d3",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "~=3.9",
"size": 104753,
"upload_time": "2025-09-02T12:32:27",
"upload_time_iso_8601": "2025-09-02T12:32:27.385928Z",
"url": "https://files.pythonhosted.org/packages/6d/2e/447da3071db4562374b32e84158ce943439fddd6414fdb3178ffb436256a/aws_cdk_aws_iot_alpha-2.214.0a0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "e54f53777c7799c5d884d14a25c6d7ef4d59ad9841aba3ce3f1314b291a5e6b1",
"md5": "ba0fd8b42bc6c34e9996069c9bc83a79",
"sha256": "c8aae3313cdeb40924cde13378a96cb5f1baed9cf264d1365a7cccc147f268b8"
},
"downloads": -1,
"filename": "aws_cdk_aws_iot_alpha-2.214.0a0.tar.gz",
"has_sig": false,
"md5_digest": "ba0fd8b42bc6c34e9996069c9bc83a79",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "~=3.9",
"size": 106169,
"upload_time": "2025-09-02T12:33:10",
"upload_time_iso_8601": "2025-09-02T12:33:10.740010Z",
"url": "https://files.pythonhosted.org/packages/e5/4f/53777c7799c5d884d14a25c6d7ef4d59ad9841aba3ce3f1314b291a5e6b1/aws_cdk_aws_iot_alpha-2.214.0a0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-02 12:33:10",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "aws",
"github_project": "aws-cdk",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "aws-cdk.aws-iot-alpha"
}