"""State module for managing AWS S3 bucket notifications."""
import copy
from dataclasses import field
from dataclasses import make_dataclass
from typing import Any
from typing import Dict
from typing import List
__contracts__ = ["resource"]
TREQ = {
"present": {"require": ["aws.s3.bucket.present"]},
}
BUCKET_NOT_EXISTS_ERROR_MESSAGE = "ClientError: An error occurred (404) when calling the HeadBucket operation: Not Found"
STATE_NAME = "aws.s3.bucket_notification"
[docs]async def present(
hub,
ctx,
name: str,
notifications: List[
make_dataclass(
"LambdaFunctionConfiguration",
[
("LambdaFunctionArn", str),
("Events", List[str]),
("Id", str, field(default=None)),
(
"Filter",
make_dataclass(
"NotificationConfigurationFilter",
[
(
"Key",
make_dataclass(
"S3KeyFilter",
[
(
"FilterRules",
List[
make_dataclass(
"FilterRuleList",
[
(
"Name",
str,
field(default=None),
),
(
"Value",
str,
field(default=None),
),
],
)
],
field(default=None),
)
],
),
field(default=None),
)
],
),
field(default=None),
),
],
)
],
resource_id: str = None,
) -> Dict[str, Any]:
"""Creates a bucket notification for an S3 bucket resource.
Bucket Notification allows user to receive notifications when certain events happen in the S3 bucket. To enable
notifications, add a notification configuration that identifies the events that you want Amazon S3 to publish.
Make sure that it also identifies the destinations where you want Amazon S3 to send the notifications. You store
this configuration in the notification subresource that's associated with a bucket
Amazon S3 can publish notifications for the following events:
a) New object created events
b) Object removal events
c) Restore object events
d) Reduced Redundancy Storage (RRS) object lost events
e) Replication events
f) S3 Lifecycle expiration events
g) S3 Lifecycle transition events
h) S3 Intelligent-Tiering automatic archival events
i) Object tagging events
j) Object ACL PUT events
Amazon S3 can send event notification messages to the following destinations. User can specify the Amazon Resource
Name (ARN) value of these destinations in the notification configuration.
1) Amazon Simple Notification Service (Amazon SNS) topics
2) Amazon Simple Queue Service (Amazon SQS) queues
3) AWS Lambda function
Args:
name(str):
Name of the bucket on which notification needs to be configured.
notifications(list[dict[str, Any]], Optional):
Describes the Lambda functions to invoke and the events for which to invoke them.
* Id (str, Optional):
An optional unique identifier for configurations in a notification configuration. If you don't
provide one, Amazon S3 will assign an ID.
* LambdaFunctionArn (str):
The Amazon Resource Name (ARN) of the Lambda function that Amazon S3 invokes when the specified event
type occurs.
* Events (list[str]):
The Amazon S3 bucket event for which to invoke the Lambda function. For more information, see
Supported Event Types in the Amazon S3 User Guide.
* Filter (dict[str, Any], Optional):
Specifies object key name filtering rules. For information about key name filtering, see
Configuring Event Notifications in the Amazon S3 User Guide.
* Key (dict[str, Any], Optional):
A container for object key name prefix and suffix filtering rules.
* FilterRules (list[dict[str, Any]], Optional):
A list of containers for the key-value pair that defines the criteria for the filter rule.
* Name (str, Optional):
The object key name prefix or suffix identifying one or more objects to which the filtering rule
applies. The maximum length is 1,024 characters. Overlapping prefixes and suffixes are not
supported. For more information, see Configuring Event Notifications in the Amazon S3 User
Guide.
* Value (str, Optional):
The value that the filter searches for in object key names.
resource_id(str, Optional):
Name of the bucket and 'notifications' keyword with a seperator '-'.
Returns:
dict[str, Any]
Request Syntax:
.. code-block:: yaml
[bucket_name]-notifications:
aws.s3.bucket_notification.present:
- name: [string]
- resource_id: [string]
- notifications
LambdaFunctionConfigurations:
- Events:
- [string]
Filter:
Key:
FilterRules:
- Name: Prefix
Value: ''
- Name: Suffix
Value: [string]
Id: [string]
LambdaFunctionArn: [string]
TopicConfigurations:
Examples:
.. code-block:: yaml
test-bucket-notifications:
aws.s3.bucket_notification.present:
- name: test-bucket
- resource_id: test-bucket-notifications
- notifications
LambdaFunctionConfigurations:
- Events:
- s3:ObjectCreated:*
Filter:
Key:
FilterRules:
- Name: Prefix
Value: ''
- Name: Suffix
Value: .jpeg
Id: test
LambdaFunctionArn: arn:aws:lambda:us-west-1:000000000000:function:test
TopicConfigurations:
"""
result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
if resource_id:
name = resource_id.split("-notifications")[0]
# Check if bucket exists
ret = await hub.exec.boto3.client.s3.head_bucket(ctx, Bucket=name)
if not ret["result"]:
if BUCKET_NOT_EXISTS_ERROR_MESSAGE in ret["comment"]:
result["comment"] += [f"aws.s3.bucket with name: {name} does not exists"]
else:
result["comment"] = ret["comment"]
result["result"] = False
return result
# Get current notification configurations for the bucket
bucket_notification_response = (
await hub.exec.boto3.client.s3.get_bucket_notification_configuration(
ctx, Bucket=name
)
)
if bucket_notification_response["result"]:
result[
"old_state"
] = hub.tool.aws.s3.conversion_utils.convert_raw_bucket_notification_to_present(
bucket_notification_response["ret"], name
)
else:
result["comment"] = ret["comment"]
result["result"] = False
return result
# check and create payload for creating notification configurations to avoid redundant calls
notifications_diff = hub.tool.aws.s3.s3_utils.get_notification_configuration_diff(
notifications, result["old_state"]
)
if notifications_diff:
if ctx.get("test", False):
result["new_state"] = hub.tool.aws.test_state_utils.generate_test_state(
enforced_state={
"name": result["old_state"].get("name"),
"resource_id": result["old_state"].get("resource_id"),
"notifications": result["old_state"].get("notifications"),
},
desired_state={
"name": name,
"resource_id": name + "-notifications",
"notifications": notifications,
},
)
if resource_id:
result["comment"] = hub.tool.aws.comment_utils.would_update_comment(
resource_type=STATE_NAME, name=name
)
else:
result["comment"] = hub.tool.aws.comment_utils.would_create_comment(
resource_type=STATE_NAME, name=name
)
return result
notification_ret = (
await hub.exec.boto3.client.s3.put_bucket_notification_configuration(
ctx, Bucket=name, NotificationConfiguration=notifications
)
)
if not notification_ret["result"]:
result["result"] = False
result["comment"] = notification_ret["comment"]
return result
bucket_notification_result = (
await hub.exec.boto3.client.s3.get_bucket_notification_configuration(
ctx, Bucket=name
)
)
if bucket_notification_result["result"]:
result[
"new_state"
] = hub.tool.aws.s3.conversion_utils.convert_raw_bucket_notification_to_present(
bucket_notification_result["ret"], name
)
else:
result["comment"] = bucket_notification_result["comment"]
result["result"] = False
else:
result["new_state"] = copy.deepcopy(result["old_state"])
return result
[docs]async def absent(
hub,
ctx,
name: str,
resource_id: str = None,
) -> Dict[str, Any]:
"""Deletes Bucket Notification from an S3 bucket resource.
This action enables you to delete multiple notification configurations from a bucket using a single HTTP request.
If you know the object keys that you want to delete, then this action provides a suitable alternative to sending
individual delete requests, reducing per-request overhead. For each configuration, Amazon S3 performs a delete
action and returns the result of that delete, success, or failure, in the response.
Args:
name(str):
Name of the bucket on which notification needs to be configured.
resource_id(str, Optional):
Name of the bucket and 'notifications' keyword with a seperator '-'.
Returns:
dict[str, Any]
Request Syntax:
.. code-block:: yaml
[bucket_name-notification]:
aws.s3.bucket_notification.absent:
- name: [string]
- resource_id: [string]
Examples:
.. code-block:: yaml
test-bucket-notification:
aws.s3.bucket_notification.absent:
- name: test-bucket
- resource_id: test-bucket-notifications
"""
result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
if resource_id:
name = resource_id.split("-notifications")[0]
# check if the bucket exists
ret = await hub.exec.boto3.client.s3.head_bucket(ctx, Bucket=name)
if not ret["result"]:
if BUCKET_NOT_EXISTS_ERROR_MESSAGE in ret["comment"]:
result["comment"] += [f"aws.s3.bucket with name: {name} does not exists"]
else:
result["comment"] = ret["comment"]
result["result"] = False
return result
# Get current notification configurations attached to the bucket
bucket_notification_response = (
await hub.exec.boto3.client.s3.get_bucket_notification_configuration(
ctx, Bucket=name
)
)
if bucket_notification_response["result"]:
result[
"old_state"
] = hub.tool.aws.s3.conversion_utils.convert_raw_bucket_notification_to_present(
bucket_notification_response["ret"], name
)
else:
result["comment"] = bucket_notification_response["comment"]
result["result"] = False
return result
if ctx.get("test", False):
result["comment"] += [
f"Would remove aws.s3.bucket_notification from bucket:'{name}'."
]
return result
if not result["old_state"]:
result["comment"] = hub.tool.aws.comment_utils.already_absent_comment(
resource_type=STATE_NAME, name=name
)
return result
notification_ret = (
await hub.exec.boto3.client.s3.put_bucket_notification_configuration(
ctx, Bucket=name, NotificationConfiguration={}
)
)
if not notification_ret["result"]:
result["result"] = False
result["comment"] = notification_ret["comment"]
return result
after_bucket_notification_response = (
await hub.exec.boto3.client.s3.get_bucket_notification_configuration(
ctx, Bucket=name
)
)
if after_bucket_notification_response["result"]:
result[
"new_state"
] = hub.tool.aws.s3.conversion_utils.convert_raw_bucket_notification_to_present(
after_bucket_notification_response["ret"], name
)
else:
result["comment"] = after_bucket_notification_response["comment"]
result["result"] = False
return result
[docs]async def describe(hub, ctx) -> Dict[str, Dict[str, Any]]:
"""Obtain S3 bucket notifications for each bucket under the given context for any user.
Returns:
dict[str, Any]
Examples:
.. code-block:: bash
$ idem describe aws.s3.bucket_notification
"""
result = {}
# To describe the notification configurations of all the buckets, we first need to list all the buckets,
# then get the notification configurations of each bucket
ret = await hub.exec.boto3.client.s3.list_buckets(ctx)
if not ret["result"]:
hub.log.warning(f"Could not describe S3 buckets {ret['comment']}")
return {}
for bucket in ret["ret"]["Buckets"]:
bucket_name = bucket.get("Name")
# get notification configuration for each bucket
try:
bucket_notification_response = (
await hub.exec.boto3.client.s3.get_bucket_notification_configuration(
ctx, Bucket=bucket_name
)
)
if bucket_notification_response["result"]:
bucket_notification_response["ret"].pop("ResponseMetadata", None)
if bucket_notification_response["ret"]:
translated_resource = hub.tool.aws.s3.conversion_utils.convert_raw_bucket_notification_to_present(
bucket_notification_response["ret"], bucket_name
)
result[bucket_name + "-notifications"] = {
"aws.s3.bucket_notification.present": [
{parameter_key: parameter_value}
for parameter_key, parameter_value in translated_resource.items()
]
}
else:
hub.log.warning(
f"Notifications configuration not present for s3 bucket '{bucket_name}'."
f"Describe will skip this bucket and continue."
)
else:
hub.log.warning(
f"Could not get attached notification configuration for bucket '{bucket_name}' with error"
f" {bucket_notification_response['comment']} . Describe will skip this bucket and continue."
)
except Exception as e:
hub.log.warning(
f"Could not get attached notification configuration for bucket '{bucket_name}' with error"
f" {e} . Describe will skip this bucket and continue."
)
return result