Source code for idem_aws.states.aws.config.rule

"""State module for managing Amazon Config Rule."""
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"]


[docs]async def present( hub, ctx, name: str, resource_id: str = None, scope: make_dataclass( "Scope", [ ("ComplianceResourceTypes", List[str], field(default=None)), ("TagKey", str, field(default=None)), ("TagValue", str, field(default=None)), ("ComplianceResourceId", str, field(default=None)), ], ) = None, source: make_dataclass( "Source", [ ("Owner", str, field(default=None)), ("SourceIdentifier", str, field(default=None)), ( "SourceDetails", List[ make_dataclass( "SourceDetail", [ ("EventSource", str, field(default=None)), ("MessageType", str, field(default=None)), ("MaximumExecutionFrequency", str, field(default=None)), ], ) ], field(default=None), ), ( "CustomPolicyDetails", make_dataclass( "CustomPolicyDetail", [ ("PolicyRuntime", str, field(default=None)), ("PolicyText", str, field(default=None)), ("EnableDebugLogDelivery", bool, field(default=None)), ], ), field(default=None), ), ], ) = None, tags: Dict[str, Any] or List[ make_dataclass( "Tag", [("Key", str, field(default=None)), ("Value", str, field(default=None))], ) ] = None, max_execution_frequency: str = None, input_parameters: str = None, ) -> Dict[str, Any]: """Adds or updates Config rule for evaluating whether your Amazon Web Services resources comply with your desired configurations. Args: name(str): An Idem name of the rule. resource_id(str, Optional): AWS Config Rule Name. scope(dict[str, Any], Optional): Defines which resources can trigger an evaluation for the rule. The scope can include one or more resource types, a combination of one resource type and one resource ID, or a combination of a tag key and value. Specify a scope to constrain the resources that can trigger an evaluation for the rule. If you do not specify a scope, evaluations are triggered when any resource in the recording group changes. The scope can be empty. * ComplianceResourceTypes (list[str], Optional): The resource types of only those Amazon Web Services resources that you want to trigger an evaluation for the rule. You can only specify one type if you also specify a resource ID for ComplianceResourceId. * TagKey (str, Optional): The tag key that is applied to only those Amazon Web Services resources that you want to trigger an evaluation for the rule. * TagValue (str, Optional): The tag value applied to only those Amazon Web Services resources that you want to trigger an evaluation for the rule. If you specify a value for TagValue, you must also specify a value for TagKey. * ComplianceResourceId (str, Optional): The ID of the only Amazon Web Services resource that you want to trigger an evaluation for the rule. If you specify a resource ID, you must specify one resource type for ComplianceResourceTypes. source (dict[str, Any]): Provides the rule owner (Amazon Web Services or customer), the rule identifier, and the notifications that cause the function to evaluate your Amazon Web Services resources. * Owner (str): Indicates whether Amazon Web Services or the customer owns and manages the Config rule. Config Managed Rules are predefined rules owned by Amazon Web Services. For more information, see Config Managed Rules in the Config developer guide. Config Custom Rules are rules that you can develop either with Guard (CUSTOM_POLICY) or Lambda (CUSTOM_LAMBDA). For more information, see Config Custom Rules in the Config developer guide. * SourceIdentifier (str, Optional): For Config Managed rules, a predefined identifier from a list. For example, IAM_PASSWORD_POLICY is a managed rule. To reference a managed rule, see List of Config Managed Rules. For Config Custom Lambda rules, the identifier is the Amazon Resource Name (ARN) of the rule's Lambda function, such as arn:aws:lambda:us-east-2:123456789012:function:custom_rule_name. For Config Custom Policy rules, this field will be ignored. * SourceDetails (list[dict[str, Any]], Optional): Provides the source and the message types that cause Config to evaluate your Amazon Web Services resources against a rule. It also provides the frequency with which you want Config to run evaluations for the rule if the trigger type is periodic. If the owner is set to CUSTOM_POLICY, the only acceptable values for the Config rule trigger message type are ConfigurationItemChangeNotification and OversizedConfigurationItemChangeNotification. * (dict) -- Provides the source and the message types that trigger Config to evaluate your Amazon Web Services resources against a rule. It also provides the frequency with which you want Config to run evaluations for the rule if the trigger type is periodic. You can specify the parameter values for SourceDetail only for custom rules. * EventSource (str, Optional): The source of the event, such as an Amazon Web Services service, that triggers Config to evaluate your Amazon Web Services resources. * MessageType (str, Optional): The type of notification that triggers Config to run an evaluation for a rule. You can specify the following notification types: * ConfigurationItemChangeNotification - Triggers an evaluation when Config delivers a configuration item as a result of a resource change. * OversizedConfigurationItemChangeNotification - Triggers an evaluation when Config delivers an oversized configuration item. Config may generate this notification type when a resource changes and the notification exceeds the maximum size allowed by Amazon SNS. * ScheduledNotification - Triggers a periodic evaluation at the frequency specified for MaximumExecutionFrequency. * ConfigurationSnapshotDeliveryCompleted - Triggers a periodic evaluation when Config delivers a configuration snapshot. If you want your custom rule to be triggered by configuration changes, specify two SourceDetail objects, one for ConfigurationItemChangeNotification and one for OversizedConfigurationItemChangeNotification. * MaximumExecutionFrequency (str, Optional): The frequency at which you want Config to run evaluations for a custom rule with a periodic trigger. If you specify a value for MaximumExecutionFrequency, then MessageType must use the ScheduledNotification value. By default, rules with a periodic trigger are evaluated every 24 hours. To change the frequency, specify a valid value for the MaximumExecutionFrequency parameter. Based on the valid value you choose, Config runs evaluations once for each valid value. For example, if you choose Three_Hours, Config runs evaluations once every three hours. In this case, Three_Hours is the frequency of this rule. * CustomPolicyDetails (dict[str, Any], Optional): Provides the runtime system, policy definition, and whether debug logging is enabled. Required when owner is set to CUSTOM_POLICY. * PolicyRuntime (str): The runtime system for your Config Custom Policy rule. Guard is a policy-as-code language that allows you to write policies that are enforced by Config Custom Policy rules. For more information about Guard, see the Guard GitHub Repository. * PolicyText (str): The policy definition containing the logic for your Config Custom Policy rule. * EnableDebugLogDelivery (bool, Optional): The boolean expression for enabling debug logging for your Config Custom Policy rule. The default value is false. max_execution_frequency(str, Optional): The maximum frequency with which Config runs evaluations for a rule. Default is 24 hours input_parameters(str, Optional): A string, in JSON format, that is passed to the Config rule. tags(dict or list, Optional): Dict in the format of {tag-key: tag-value} or List of tags in the format of [{"Key": tag-key, "Value": tag-value}] to associate with the Config rule. The metadata that you apply to a resource to help you categorize and organize them. * (Key, Optional): One part of a key-value pair that make up a tag. A key is a general label that acts like a category for more specific tag values. * (Value, Optional): The optional part of a key-value pair that make up a tag. A value acts as a descriptor within a tag category (key). Request syntax: .. code-block:: sls [aws-config-rule]: aws.config.rule.present: - name: 'string' - resource_id: 'string' - scope: dict ComplianceResourceTypes: list - source: dict Owner: 'string' SourceIdentifier: 'string' Returns: Dict[str, Any] Examples: .. code-block:: sls ec2-instance-no-public-ip: aws.config.rule.present: - name: ec2-instance-no-public-ip - resource_id: ec2-instance-no-public-ip - tags: - Key: ENV Value: Test - Key: Service Value: TestService - config_rule_name: ec2-instance-no-public-ip - scope: ComplianceResourceTypes: - AWS::EC2::Instance - AWS::EC2::Host - source: Owner: AWS SourceIdentifier: EC2_INSTANCE_NO_PUBLIC_IP """ result = dict(comment=[], old_state=None, new_state=None, name=name, result=True) before = None resource_updated = False update_scope = None if resource_id: try: before = await hub.exec.boto3.client.config.describe_config_rules( ctx, ConfigRuleNames=[resource_id] ) except hub.tool.boto3.exception.ClientError as e: result["comment"] += [f"{e.__class__.__name__}: {e}"] result["result"] = False return result if isinstance(tags, List): tags = hub.tool.aws.tag_utils.convert_tag_list_to_dict(tags) if before: try: result[ "old_state" ] = await hub.tool.aws.config.conversion_utils.convert_raw_config_rule_to_present_async( ctx, raw_resource=before["ret"]["ConfigRules"][0], idem_resource_name=name, ) plan_state = copy.deepcopy(result["old_state"]) # updating scope if scope is not None: update_scope = ( scope if scope is not None else result["old_state"]["scope"] ) update_ret = await hub.tool.aws.config.rule.update_rule( ctx, resource_id=resource_id, before=before["ret"]["ConfigRules"][0], source=source, scope=update_scope, frequency=max_execution_frequency, input_parameters=input_parameters, ) result["comment"] += update_ret["comment"] result["result"] = update_ret["result"] resource_updated = resource_updated or bool(update_ret["ret"]) if update_ret["ret"] and ctx.get("test", False): for key in ["max_execution_frequency", "scope", "input_parameters"]: if key in update_ret["ret"]: plan_state[key] = update_ret["ret"].get(key) if (tags is not None) and tags != result["old_state"].get("tags"): update_tag_ret = await hub.tool.aws.config.tag.update_tags( ctx, result["old_state"]["config_rule_arn"], result["old_state"]["tags"], tags, ) result["result"] = result["result"] and update_tag_ret["result"] result["comment"] += update_tag_ret["comment"] resource_updated = resource_updated or bool(update_tag_ret["result"]) if ctx.get("test", False) and update_tag_ret["ret"] is not None: plan_state["tags"] = update_tag_ret["ret"] if not resource_updated: result["comment"] += [f"{name} already exists"] except hub.tool.boto3.exception.ClientError as e: result["comment"] += [f"{e.__class__.__name__}: {e}"] result["result"] = False else: try: if ctx.get("test", False): desired_state_payload = { "name": name, "config_rule_name": name, "source": source, "tags": tags, } if scope: desired_state_payload["scope"] = scope if input_parameters: desired_state_payload["input_parameters"] = input_parameters if max_execution_frequency: desired_state_payload[ "max_execution_frequency" ] = max_execution_frequency result["new_state"] = hub.tool.aws.test_state_utils.generate_test_state( enforced_state={}, desired_state=desired_state_payload ) result["comment"] += [f"Would create aws.config.rule {name}"] return result config_rule_payload = {"ConfigRuleName": name, "Source": source} if scope: config_rule_payload["Scope"] = scope if max_execution_frequency: config_rule_payload[ "MaximumExecutionFrequency" ] = max_execution_frequency if input_parameters: config_rule_payload["InputParameters"] = input_parameters ret = await hub.exec.boto3.client.config.put_config_rule( ctx, ConfigRule=config_rule_payload, Tags=hub.tool.aws.tag_utils.convert_tag_dict_to_list(tags) if tags else None, ) result["result"] = ret["result"] if not result["result"]: result["comment"] = ret["comment"] + ret["comment"] return result resource_id = name result["comment"] += [f"Created '{name}'"] except hub.tool.boto3.exception.ClientError as e: result["comment"] += f"{e.__class__.__name__}: {e}" result["result"] = False try: if ctx.get("test", False): result["new_state"] = plan_state elif (not before) or resource_updated: after = await hub.exec.boto3.client.config.describe_config_rules( ctx, ConfigRuleNames=[resource_id] ) result[ "new_state" ] = await hub.tool.aws.config.conversion_utils.convert_raw_config_rule_to_present_async( ctx, raw_resource=after["ret"]["ConfigRules"][0], idem_resource_name=name, ) else: result["new_state"] = copy.deepcopy(result["old_state"]) except Exception as e: result["comment"] += [str(e)] result["result"] = False return result
[docs]async def absent( hub, ctx, name: str, resource_id: str = None, ) -> Dict[str, Any]: """Deletes the specified Config rule and all of its evaluation results. Config sets the state of a rule to DELETING until the deletion is complete. You cannot update a rule while it is in this state. If you make a PutConfigRule or DeleteConfigRule request for the rule, you will receive a ResourceInUseException. Args: name(str): An Idem name of the rule. resource_id(str, Optional): AWS Config Rule Name. Idem automatically considers this resource being absent if this field is not specified. Returns: Dict[str, Any] Examples: .. code-block:: sls ec2-instance-no-public-ip: aws.config.rule.absent: - name: ec2-instance-no-public-ip - resource_id: ec2-instance-no-public-ip """ result = dict(comment=[], old_state=None, new_state=None, name=name, result=True) if not resource_id: result["comment"] = hub.tool.aws.comment_utils.already_absent_comment( resource_type="aws.config.rule", name=name ) return result before = await hub.exec.boto3.client.config.describe_config_rules( ctx, ConfigRuleNames=[resource_id] ) if not before: result["comment"] = hub.tool.aws.comment_utils.already_absent_comment( resource_type="aws.config.rule", name=name ) elif ctx.get("test", False): result[ "old_state" ] = await hub.tool.aws.config.conversion_utils.convert_raw_config_rule_to_present_async( ctx, raw_resource=before["ret"]["ConfigRules"][0], idem_resource_name=name ) result["comment"] = hub.tool.aws.comment_utils.would_delete_comment( resource_type="aws.config.rule", name=name ) return result else: try: if before["ret"]["ConfigRules"][0]["ConfigRuleState"] == "DELETING": result["comment"] += [ f"aws.config.rule '{name}' is still in deleting state." ] else: result[ "old_state" ] = await hub.tool.aws.config.conversion_utils.convert_raw_config_rule_to_present_async( ctx, raw_resource=before["ret"]["ConfigRules"][0], idem_resource_name=name, ) ret = await hub.exec.boto3.client.config.delete_config_rule( ctx, ConfigRuleName=resource_id ) result["result"] = ret["result"] if not result["result"]: result["comment"] = ret["comment"] result["result"] = False return result result["comment"] = hub.tool.aws.comment_utils.delete_comment( resource_type="aws.config.rule", name=name ) except hub.tool.boto3.exception.ClientError as e: result["comment"] += [f"{e.__class__.__name__}: {e}"] result["result"] = False return result
[docs]async def describe(hub, ctx) -> Dict[str, Dict[str, Any]]: """Describe the resource in a way that can be recreated/managed with the corresponding "present" function. Return details about your Config rules. Returns: Dict[str, Any] Examples: .. code-block:: bash $ idem describe aws.config.rule """ result = {} ret = await hub.exec.boto3.client.config.describe_config_rules(ctx) if not ret["result"]: hub.log.warning(f"Could not describe Rules {ret['comment']}") return {} for resource in ret["ret"]["ConfigRules"]: resource_id = resource.get("ConfigRuleName") resource_translated = await hub.tool.aws.config.conversion_utils.convert_raw_config_rule_to_present_async( ctx, raw_resource=resource, idem_resource_name=resource_id ) result[resource_id] = { "aws.config.rule.present": [ {parameter_key: parameter_value} for parameter_key, parameter_value in resource_translated.items() ] } return result