From b59ef704e1a66e9e2e0059fb38e6bbe1a31fa7b3 Mon Sep 17 00:00:00 2001 From: Jason Quesenberry Date: Thu, 4 Sep 2025 10:30:04 -0700 Subject: [PATCH] AWS Config for Python --- .doc_gen/metadata/config_metadata.yaml | 196 ++++++++++ .gitignore | 1 + python/example_code/config/README.md | 21 ++ python/example_code/config/config_basics.py | 356 ++++++++++++++++++ python/example_code/config/config_hello.py | 61 +++ python/example_code/config/config_rules.py | 194 ++++++++++ .../config/test/test_config_hello.py | 62 +++ .../config/test/test_config_rules.py | 85 +++++ scenarios/basics/config/SPECIFICATION.md | 92 +++++ 9 files changed, 1068 insertions(+) create mode 100644 .doc_gen/metadata/config_metadata.yaml create mode 100644 python/example_code/config/config_basics.py create mode 100644 python/example_code/config/config_hello.py create mode 100644 python/example_code/config/test/test_config_hello.py create mode 100644 scenarios/basics/config/SPECIFICATION.md diff --git a/.doc_gen/metadata/config_metadata.yaml b/.doc_gen/metadata/config_metadata.yaml new file mode 100644 index 00000000000..056daa37b70 --- /dev/null +++ b/.doc_gen/metadata/config_metadata.yaml @@ -0,0 +1,196 @@ +# zexi 0.4.0 +config_PutConfigurationRecorder: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.PutConfigurationRecorder + services: + config-service: {PutConfigurationRecorder} +config_PutDeliveryChannel: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.PutDeliveryChannel + services: + config-service: {PutDeliveryChannel} +config_StartConfigurationRecorder: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.StartConfigurationRecorder + services: + config-service: {StartConfigurationRecorder} +config_DescribeConfigurationRecorders: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.DescribeConfigurationRecorders + services: + config-service: {DescribeConfigurationRecorders} +config_DescribeConfigurationRecorderStatus: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.DescribeConfigurationRecorderStatus + services: + config-service: {DescribeConfigurationRecorderStatus} +config_ListDiscoveredResources: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.ListDiscoveredResources + services: + config-service: {ListDiscoveredResources} +config_GetResourceConfigHistory: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.GetResourceConfigHistory + services: + config-service: {GetResourceConfigHistory} +config_StopConfigurationRecorder: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.StopConfigurationRecorder + services: + config-service: {StopConfigurationRecorder} +config_DeleteConfigurationRecorder: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.DeleteConfigurationRecorder + services: + config-service: {DeleteConfigurationRecorder} +config_DeleteDeliveryChannel: + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.DeleteDeliveryChannel + services: + config-service: {DeleteDeliveryChannel} +config_Hello: + title: Hello &Config; + title_abbrev: Hello &Config; + synopsis: get started using &Config; by listing existing configuration recorders. + category: Hello + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.config.HelloConfig + services: + config-service: {DescribeConfigurationRecorders} +config_Scenario: + title: Get started with &Config; using an &AWS; SDK + title_abbrev: Get started with &Config; + synopsis_list: + - Create a configuration recorder to track &AWS; resource configurations. + - Set up a delivery channel to specify where &Config; sends configuration snapshots. + - Start the configuration recorder to begin monitoring resources. + - Monitor configuration recorder status and settings. + - Discover &AWS; resources in your account. + - View configuration history for specific resources. + - Clean up &Config; resources. + category: Scenarios + languages: + Python: + versions: + - sdk_version: 3 + github: python/example_code/config + sdkguide: + excerpts: + - description: Create a wrapper class for &Config; operations. + snippet_tags: + - python.example_code.config.ConfigWrapper + - python.example_code.config.PutConfigurationRecorder + - python.example_code.config.PutDeliveryChannel + - python.example_code.config.StartConfigurationRecorder + - python.example_code.config.DescribeConfigurationRecorders + - python.example_code.config.DescribeConfigurationRecorderStatus + - python.example_code.config.ListDiscoveredResources + - python.example_code.config.GetResourceConfigHistory + - python.example_code.config.StopConfigurationRecorder + - python.example_code.config.DeleteConfigurationRecorder + - python.example_code.config.DeleteDeliveryChannel + - description: Run a comprehensive scenario demonstrating &Config; setup and resource monitoring. + snippet_tags: + - python.example_code.config.Scenario_ConfigBasics + services: + config-service: {PutConfigurationRecorder, PutDeliveryChannel, StartConfigurationRecorder, DescribeConfigurationRecorders, DescribeConfigurationRecorderStatus, ListDiscoveredResources, GetResourceConfigHistory, StopConfigurationRecorder, DeleteConfigurationRecorder, DeleteDeliveryChannel} + diff --git a/.gitignore b/.gitignore index 0b25f6593e2..b5f92e2049f 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,7 @@ Package.resolved build_dir node_modules super-linter.log +debug.log target vendor venv diff --git a/python/example_code/config/README.md b/python/example_code/config/README.md index 56c1ac66c0e..be0864c66b7 100644 --- a/python/example_code/config/README.md +++ b/python/example_code/config/README.md @@ -40,7 +40,28 @@ Code excerpts that show you how to call individual service functions. - [DeleteConfigRule](config_rules.py#L89) - [DescribeConfigRules](config_rules.py#L67) +- [DescribeConfigurationRecorders](config_rules.py#L150) +- [DescribeConfigurationRecorderStatus](config_rules.py#L175) +- [GetResourceConfigHistory](config_rules.py#L225) +- [ListDiscoveredResources](config_rules.py#L200) - [PutConfigRule](config_rules.py#L34) +- [PutConfigurationRecorder](config_rules.py#L105) +- [PutDeliveryChannel](config_rules.py#L130) +- [StartConfigurationRecorder](config_rules.py#L155) +- [StopConfigurationRecorder](config_rules.py#L250) + +### Hello + +Code example that shows you how to get started using the service. + +- [Hello AWS Config](config_hello.py) + +### Scenarios + +Code examples that show you how to accomplish a specific task by calling multiple +functions within the same service. + +- [Get started with AWS Config](config_basics.py) diff --git a/python/example_code/config/config_basics.py b/python/example_code/config/config_basics.py new file mode 100644 index 00000000000..16255ad1443 --- /dev/null +++ b/python/example_code/config/config_basics.py @@ -0,0 +1,356 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Purpose + +Shows how to use the AWS SDK for Python (Boto3) to work with AWS Config. +This scenario demonstrates how to: +- Set up a configuration recorder to track AWS resource configurations +- Create a delivery channel to specify where Config sends configuration snapshots +- Start the configuration recorder to begin monitoring resources +- Monitor configuration recorder status and settings +- Discover AWS resources in the account +- Retrieve configuration history for specific resources +- Clean up resources when done + +This example requires an S3 bucket and IAM role with appropriate permissions. +""" + +import logging +import time +import boto3 +from botocore.exceptions import ClientError + +from config_rules import ConfigWrapper +import sys +import os +sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..')) +import demo_tools.question as q + +logger = logging.getLogger(__name__) + + +# snippet-start:[python.example_code.config-service.Scenario_ConfigBasics] +class ConfigBasicsScenario: + """ + Runs an interactive scenario that shows how to get started with AWS Config. + """ + + def __init__(self, config_wrapper, s3_resource, iam_resource): + """ + :param config_wrapper: An object that wraps AWS Config operations. + :param s3_resource: A Boto3 S3 resource. + :param iam_resource: A Boto3 IAM resource. + """ + self.config_wrapper = config_wrapper + self.s3_resource = s3_resource + self.iam_resource = iam_resource + self.recorder_name = None + self.channel_name = None + self.bucket_name = None + self.role_arn = None + + def run_scenario(self): + """ + Runs the scenario. + """ + print("-" * 88) + print("Welcome to the AWS Config basics scenario!") + print("-" * 88) + + print( + "AWS Config provides a detailed view of the resources associated with your AWS account, " + "including how they are configured, how they are related to one another, and how the " + "configurations and their relationships have changed over time." + ) + print() + + # Setup phase + if not self._setup_resources(): + return + + try: + # Configuration monitoring phase + self._demonstrate_configuration_monitoring() + + # Resource discovery phase + self._demonstrate_resource_discovery() + + # Configuration history phase + self._demonstrate_configuration_history() + + finally: + # Cleanup phase + self._cleanup_resources() + + print("Thanks for watching!") + print("-" * 88) + + def _setup_resources(self): + """ + Sets up the necessary resources for the scenario. + """ + print("\n" + "-" * 60) + print("Setup") + print("-" * 60) + + # Get S3 bucket for delivery channel + self.bucket_name = q.ask( + "Enter the name of an S3 bucket for Config to deliver configuration snapshots " + "(the bucket must exist and have appropriate permissions): ", + q.non_empty + ) + + # Verify bucket exists + try: + self.s3_resource.meta.client.head_bucket(Bucket=self.bucket_name) + print(f"✓ S3 bucket '{self.bucket_name}' found.") + except ClientError as err: + if err.response['Error']['Code'] == '404': + print(f"✗ S3 bucket '{self.bucket_name}' not found.") + return False + else: + print(f"✗ Error accessing S3 bucket: {err}") + return False + + # Get IAM role ARN + self.role_arn = q.ask( + "Enter the ARN of an IAM role that grants AWS Config permissions to access your resources " + "(e.g., arn:aws:iam::123456789012:role/config-role): ", + q.non_empty + ) + + # Verify role exists + try: + role_name = self.role_arn.split('/')[-1] + self.iam_resource.Role(role_name).load() + print(f"✓ IAM role found.") + except ClientError as err: + print(f"✗ Error accessing IAM role: {err}") + return False + + # Create configuration recorder + self.recorder_name = "demo-config-recorder" + print(f"\nCreating configuration recorder '{self.recorder_name}'...") + try: + self.config_wrapper.put_configuration_recorder( + self.recorder_name, + self.role_arn + ) + print("✓ Configuration recorder created successfully.") + except ClientError as err: + if 'MaxNumberOfConfigurationRecordersExceededException' in str(err): + print("✗ Maximum number of configuration recorders exceeded.") + print("You can have only one configuration recorder per region.") + return False + else: + print(f"✗ Error creating configuration recorder: {err}") + return False + + # Create delivery channel + self.channel_name = "demo-delivery-channel" + print(f"\nCreating delivery channel '{self.channel_name}'...") + try: + self.config_wrapper.put_delivery_channel( + self.channel_name, + self.bucket_name, + "config-snapshots/" + ) + print("✓ Delivery channel created successfully.") + except ClientError as err: + print(f"✗ Error creating delivery channel: {err}") + return False + + # Start configuration recorder + print(f"\nStarting configuration recorder '{self.recorder_name}'...") + try: + self.config_wrapper.start_configuration_recorder(self.recorder_name) + print("✓ Configuration recorder started successfully.") + print("AWS Config is now monitoring your resources!") + except ClientError as err: + print(f"✗ Error starting configuration recorder: {err}") + return False + + return True + + def _demonstrate_configuration_monitoring(self): + """ + Demonstrates configuration monitoring capabilities. + """ + print("\n" + "-" * 60) + print("Configuration Monitoring") + print("-" * 60) + + # Show recorder status + print("Checking configuration recorder status...") + try: + statuses = self.config_wrapper.describe_configuration_recorder_status([self.recorder_name]) + if statuses: + status = statuses[0] + print(f"Recorder: {status['name']}") + print(f"Recording: {status['recording']}") + print(f"Last Status: {status.get('lastStatus', 'N/A')}") + if 'lastStartTime' in status: + print(f"Last Started: {status['lastStartTime']}") + except ClientError as err: + print(f"Error getting recorder status: {err}") + + # Show recorder configuration + print("\nConfiguration recorder settings:") + try: + recorders = self.config_wrapper.describe_configuration_recorders([self.recorder_name]) + if recorders: + recorder = recorders[0] + recording_group = recorder.get('recordingGroup', {}) + print(f"Recording all supported resources: {recording_group.get('allSupported', False)}") + print(f"Including global resources: {recording_group.get('includeGlobalResourceTypes', False)}") + + if not recording_group.get('allSupported', True): + resource_types = recording_group.get('resourceTypes', []) + print(f"Specific resource types: {', '.join(resource_types)}") + except ClientError as err: + print(f"Error getting recorder configuration: {err}") + + # Wait a moment for resources to be discovered + print("\nWaiting for AWS Config to discover resources...") + time.sleep(10) + + def _demonstrate_resource_discovery(self): + """ + Demonstrates resource discovery capabilities. + """ + print("\n" + "-" * 60) + print("Resource Discovery") + print("-" * 60) + + # Common resource types to check + resource_types = [ + 'AWS::S3::Bucket', + 'AWS::EC2::Instance', + 'AWS::IAM::Role', + 'AWS::Lambda::Function' + ] + + print("Discovering AWS resources in your account...") + total_resources = 0 + + for resource_type in resource_types: + try: + resources = self.config_wrapper.list_discovered_resources(resource_type, limit=10) + count = len(resources) + total_resources += count + print(f"{resource_type}: {count} resources") + + # Show details for first few resources + if resources and count > 0: + print(f" Sample resources:") + for i, resource in enumerate(resources[:3]): + print(f" {i+1}. {resource.get('resourceId', 'N/A')} ({resource.get('resourceName', 'Unnamed')})") + if count > 3: + print(f" ... and {count - 3} more") + print() + + except ClientError as err: + print(f"Error listing {resource_type}: {err}") + + print(f"Total resources discovered: {total_resources}") + + def _demonstrate_configuration_history(self): + """ + Demonstrates configuration history capabilities. + """ + print("\n" + "-" * 60) + print("Configuration History") + print("-" * 60) + + # Try to get configuration history for the S3 bucket we're using + print(f"Getting configuration history for S3 bucket '{self.bucket_name}'...") + try: + config_items = self.config_wrapper.get_resource_config_history( + 'AWS::S3::Bucket', + self.bucket_name, + limit=5 + ) + + if config_items: + print(f"Found {len(config_items)} configuration item(s):") + for i, item in enumerate(config_items): + print(f"\n Configuration {i+1}:") + print(f" Configuration Item Capture Time: {item.get('configurationItemCaptureTime', 'N/A')}") + print(f" Configuration State Id: {item.get('configurationStateId', 'N/A')}") + print(f" Configuration Item Status: {item.get('configurationItemStatus', 'N/A')}") + print(f" Resource Type: {item.get('resourceType', 'N/A')}") + print(f" Resource Id: {item.get('resourceId', 'N/A')}") + + # Show some configuration details + config_data = item.get('configuration') + if config_data and isinstance(config_data, dict): + print(f" Sample configuration keys: {list(config_data.keys())[:5]}") + else: + print("No configuration history found yet. This is normal for newly monitored resources.") + print("Configuration history will be available after resources are modified.") + + except ClientError as err: + if 'ResourceNotDiscoveredException' in str(err): + print("Resource not yet discovered by AWS Config. This is normal for new setups.") + else: + print(f"Error getting configuration history: {err}") + + def _cleanup_resources(self): + """ + Cleans up resources created during the scenario. + """ + print("\n" + "-" * 60) + print("Cleanup") + print("-" * 60) + + if self.recorder_name: + cleanup = q.ask( + f"Do you want to stop and delete the configuration recorder '{self.recorder_name}'? " + "This will stop monitoring your resources. (y/n): ", + q.is_yesno + ) + + if cleanup: + # Stop the configuration recorder + print(f"Stopping configuration recorder '{self.recorder_name}'...") + try: + self.config_wrapper.stop_configuration_recorder(self.recorder_name) + print("✓ Configuration recorder stopped.") + except ClientError as err: + print(f"Error stopping configuration recorder: {err}") + + # Note: In a real scenario, you might also want to delete the recorder and delivery channel + # However, this example leaves them for the user to manage manually + print("\nNote: The configuration recorder and delivery channel have been left in your account.") + print("You can manage them through the AWS Console or delete them manually if needed.") + else: + print("Configuration recorder left running. You can manage it through the AWS Console.") + + print("\nScenario completed!") + + +# snippet-end:[python.example_code.config-service.Scenario_ConfigBasics] + + +def main(): + """ + Runs the Config basics scenario. + """ + logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") + + print("-" * 88) + print("Welcome to the AWS Config basics scenario!") + print("-" * 88) + + config_wrapper = ConfigWrapper(boto3.client('config')) + s3_resource = boto3.resource('s3') + iam_resource = boto3.resource('iam') + + scenario = ConfigBasicsScenario(config_wrapper, s3_resource, iam_resource) + scenario.run_scenario() + + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/python/example_code/config/config_hello.py b/python/example_code/config/config_hello.py new file mode 100644 index 00000000000..6079138076d --- /dev/null +++ b/python/example_code/config/config_hello.py @@ -0,0 +1,61 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Purpose + +Shows how to use the AWS SDK for Python (Boto3) to get started with AWS Config. +This is a simple example that demonstrates basic connectivity to the service. +""" + +import logging +import boto3 +from botocore.exceptions import ClientError + +logger = logging.getLogger(__name__) + + +# snippet-start:[python.example_code.config-service.Hello] +def hello_config(config_client): + """ + Use the AWS SDK for Python (Boto3) to create an AWS Config client and list + any existing configuration recorders in your account. + This example uses the default settings specified in your shared credentials + and config files. + + :param config_client: A Boto3 AWS Config client. + """ + print("Hello, AWS Config! Let's list your configuration recorders:") + try: + response = config_client.describe_configuration_recorders() + recorders = response.get('ConfigurationRecorders', []) + + if recorders: + print(f"Found {len(recorders)} configuration recorder(s):") + for recorder in recorders: + print(f" - {recorder['name']} (Status: {'Recording' if recorder.get('recordingGroup', {}).get('allSupported', False) else 'Limited'})") + else: + print("No configuration recorders found in your account.") + print("You can create one using the AWS Config basics scenario.") + + except ClientError as err: + if err.response['Error']['Code'] == 'AccessDeniedException': + print("You don't have permission to access AWS Config.") + print("Make sure your AWS credentials have the necessary Config permissions.") + else: + logger.error( + "Couldn't list configuration recorders. Here's why: %s: %s", + err.response['Error']['Code'], err.response['Error']['Message']) + raise + + +# snippet-end:[python.example_code.config-service.Hello] + + +def main(): + config_client = boto3.client('config') + hello_config(config_client) + + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/python/example_code/config/config_rules.py b/python/example_code/config/config_rules.py index 618b811be17..6ecd2cf120e 100644 --- a/python/example_code/config/config_rules.py +++ b/python/example_code/config/config_rules.py @@ -10,6 +10,7 @@ import logging from pprint import pprint +import time import boto3 from botocore.exceptions import ClientError @@ -103,6 +104,199 @@ def delete_config_rule(self, rule_name): # snippet-end:[python.example_code.config-service.DeleteConfigRule] + # snippet-start:[python.example_code.config-service.PutConfigurationRecorder] + def put_configuration_recorder(self, recorder_name, role_arn, resource_types=None): + """ + Creates a configuration recorder to track AWS resource configurations. + + :param recorder_name: The name of the configuration recorder. + :param role_arn: The ARN of the IAM role that grants AWS Config permissions. + :param resource_types: List of resource types to record. If None, records all supported types. + """ + try: + recording_group = { + 'allSupported': resource_types is None, + 'includeGlobalResourceTypes': True + } + + if resource_types: + recording_group['allSupported'] = False + recording_group['resourceTypes'] = resource_types + + self.config_client.put_configuration_recorder( + ConfigurationRecorder={ + 'name': recorder_name, + 'roleARN': role_arn, + 'recordingGroup': recording_group + } + ) + logger.info("Created configuration recorder %s.", recorder_name) + except ClientError: + logger.exception("Couldn't create configuration recorder %s.", recorder_name) + raise + + # snippet-end:[python.example_code.config-service.PutConfigurationRecorder] + + # snippet-start:[python.example_code.config-service.PutDeliveryChannel] + def put_delivery_channel(self, channel_name, bucket_name, bucket_prefix=None): + """ + Creates a delivery channel to specify where AWS Config sends configuration snapshots. + + :param channel_name: The name of the delivery channel. + :param bucket_name: The name of the S3 bucket where Config delivers configuration snapshots. + :param bucket_prefix: The prefix for the S3 bucket (optional). + """ + try: + delivery_channel = { + 'name': channel_name, + 's3BucketName': bucket_name + } + + if bucket_prefix: + delivery_channel['s3KeyPrefix'] = bucket_prefix + + self.config_client.put_delivery_channel(DeliveryChannel=delivery_channel) + logger.info("Created delivery channel %s.", channel_name) + except ClientError: + logger.exception("Couldn't create delivery channel %s.", channel_name) + raise + + # snippet-end:[python.example_code.config-service.PutDeliveryChannel] + + # snippet-start:[python.example_code.config-service.StartConfigurationRecorder] + def start_configuration_recorder(self, recorder_name): + """ + Starts the configuration recorder to begin monitoring resources. + + :param recorder_name: The name of the configuration recorder to start. + """ + try: + self.config_client.start_configuration_recorder( + ConfigurationRecorderName=recorder_name + ) + logger.info("Started configuration recorder %s.", recorder_name) + except ClientError: + logger.exception("Couldn't start configuration recorder %s.", recorder_name) + raise + + # snippet-end:[python.example_code.config-service.StartConfigurationRecorder] + + # snippet-start:[python.example_code.config-service.DescribeConfigurationRecorders] + def describe_configuration_recorders(self, recorder_names=None): + """ + Gets data for configuration recorders. + + :param recorder_names: List of recorder names to describe. If None, describes all recorders. + :return: List of configuration recorder data. + """ + try: + if recorder_names: + response = self.config_client.describe_configuration_recorders( + ConfigurationRecorderNames=recorder_names + ) + else: + response = self.config_client.describe_configuration_recorders() + + recorders = response.get('ConfigurationRecorders', []) + logger.info("Got data for %d configuration recorder(s).", len(recorders)) + return recorders + except ClientError: + logger.exception("Couldn't get configuration recorder data.") + raise + + # snippet-end:[python.example_code.config-service.DescribeConfigurationRecorders] + + # snippet-start:[python.example_code.config-service.DescribeConfigurationRecorderStatus] + def describe_configuration_recorder_status(self, recorder_names=None): + """ + Gets the status of configuration recorders. + + :param recorder_names: List of recorder names to check. If None, checks all recorders. + :return: List of configuration recorder status data. + """ + try: + if recorder_names: + response = self.config_client.describe_configuration_recorder_status( + ConfigurationRecorderNames=recorder_names + ) + else: + response = self.config_client.describe_configuration_recorder_status() + + statuses = response.get('ConfigurationRecordersStatus', []) + logger.info("Got status for %d configuration recorder(s).", len(statuses)) + return statuses + except ClientError: + logger.exception("Couldn't get configuration recorder status.") + raise + + # snippet-end:[python.example_code.config-service.DescribeConfigurationRecorderStatus] + + # snippet-start:[python.example_code.config-service.ListDiscoveredResources] + def list_discovered_resources(self, resource_type, limit=20): + """ + Lists discovered AWS resources of a specific type. + + :param resource_type: The type of resources to list (e.g., 'AWS::S3::Bucket'). + :param limit: Maximum number of resources to return. + :return: List of discovered resources. + """ + try: + response = self.config_client.list_discovered_resources( + resourceType=resource_type, + limit=limit + ) + resources = response.get('resourceIdentifiers', []) + logger.info("Found %d resources of type %s.", len(resources), resource_type) + return resources + except ClientError: + logger.exception("Couldn't list discovered resources of type %s.", resource_type) + raise + + # snippet-end:[python.example_code.config-service.ListDiscoveredResources] + + # snippet-start:[python.example_code.config-service.GetResourceConfigHistory] + def get_resource_config_history(self, resource_type, resource_id, limit=10): + """ + Gets the configuration history for a specific resource. + + :param resource_type: The type of the resource (e.g., 'AWS::S3::Bucket'). + :param resource_id: The ID of the resource. + :param limit: Maximum number of configuration items to return. + :return: List of configuration items showing the resource's history. + """ + try: + response = self.config_client.get_resource_config_history( + resourceType=resource_type, + resourceId=resource_id, + limit=limit + ) + config_items = response.get('configurationItems', []) + logger.info("Got %d configuration items for resource %s.", len(config_items), resource_id) + return config_items + except ClientError: + logger.exception("Couldn't get configuration history for resource %s.", resource_id) + raise + + # snippet-end:[python.example_code.config-service.GetResourceConfigHistory] + + # snippet-start:[python.example_code.config-service.StopConfigurationRecorder] + def stop_configuration_recorder(self, recorder_name): + """ + Stops the configuration recorder. + + :param recorder_name: The name of the configuration recorder to stop. + """ + try: + self.config_client.stop_configuration_recorder( + ConfigurationRecorderName=recorder_name + ) + logger.info("Stopped configuration recorder %s.", recorder_name) + except ClientError: + logger.exception("Couldn't stop configuration recorder %s.", recorder_name) + raise + + # snippet-end:[python.example_code.config-service.StopConfigurationRecorder] + def usage_demo(): print("-" * 88) diff --git a/python/example_code/config/test/test_config_hello.py b/python/example_code/config/test/test_config_hello.py new file mode 100644 index 00000000000..abf5bb573a8 --- /dev/null +++ b/python/example_code/config/test/test_config_hello.py @@ -0,0 +1,62 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Unit tests for config_hello.py. +""" + +import boto3 +from botocore.exceptions import ClientError +import pytest + +import sys +import os +sys.path.append(os.path.join(os.path.dirname(__file__), '..')) +from config_hello import hello_config + + +@pytest.mark.parametrize("error_code", [None, "AccessDeniedException", "TestException"]) +def test_hello_config(make_stubber, capsys, error_code): + config_client = boto3.client("config") + config_stubber = make_stubber(config_client) + + recorders = [ + { + "name": "test-recorder-1", + "recordingGroup": {"allSupported": True} + }, + { + "name": "test-recorder-2", + "recordingGroup": {"allSupported": False} + } + ] + + config_stubber.stub_describe_configuration_recorders(None, recorders, error_code=error_code) + + if error_code is None: + hello_config(config_client) + captured = capsys.readouterr() + assert "Hello, AWS Config!" in captured.out + assert "Found 2 configuration recorder(s)" in captured.out + assert "test-recorder-1" in captured.out + assert "test-recorder-2" in captured.out + elif error_code == "AccessDeniedException": + hello_config(config_client) + captured = capsys.readouterr() + assert "You don't have permission to access AWS Config" in captured.out + else: + with pytest.raises(ClientError) as exc_info: + hello_config(config_client) + assert exc_info.value.response["Error"]["Code"] == error_code + + +def test_hello_config_no_recorders(make_stubber, capsys): + config_client = boto3.client("config") + config_stubber = make_stubber(config_client) + + config_stubber.stub_describe_configuration_recorders(None, [], error_code=None) + + hello_config(config_client) + captured = capsys.readouterr() + assert "Hello, AWS Config!" in captured.out + assert "No configuration recorders found" in captured.out \ No newline at end of file diff --git a/python/example_code/config/test/test_config_rules.py b/python/example_code/config/test/test_config_rules.py index aa7ad33ed3b..28751efa4a0 100644 --- a/python/example_code/config/test/test_config_rules.py +++ b/python/example_code/config/test/test_config_rules.py @@ -8,6 +8,9 @@ from botocore.exceptions import ClientError import pytest +import sys +import os +sys.path.append(os.path.join(os.path.dirname(__file__), '..')) from config_rules import ConfigWrapper @@ -75,3 +78,85 @@ def test_delete_config_rule(make_stubber, error_code): with pytest.raises(ClientError) as exc_info: config.delete_config_rule(rule_name) assert exc_info.value.response["Error"]["Code"] == error_code + + +@pytest.mark.parametrize("error_code", [None, "TestException"]) +def test_put_configuration_recorder(make_stubber, error_code): + config_client = boto3.client("config") + config_stubber = make_stubber(config_client) + config = ConfigWrapper(config_client) + recorder_name = "test-recorder" + role_arn = "arn:aws:iam::123456789012:role/config-role" + + recorder = { + 'name': recorder_name, + 'roleARN': role_arn, + 'recordingGroup': { + 'allSupported': True, + 'includeGlobalResourceTypes': True + } + } + + config_stubber.stub_put_configuration_recorder(recorder, error_code=error_code) + + if error_code is None: + config.put_configuration_recorder(recorder_name, role_arn) + else: + with pytest.raises(ClientError) as exc_info: + config.put_configuration_recorder(recorder_name, role_arn) + assert exc_info.value.response["Error"]["Code"] == error_code + + +@pytest.mark.parametrize("error_code", [None, "TestException"]) +def test_describe_configuration_recorders(make_stubber, error_code): + config_client = boto3.client("config") + config_stubber = make_stubber(config_client) + config = ConfigWrapper(config_client) + recorder_name = "test-recorder" + recorders = [{"name": recorder_name}] + + config_stubber.stub_describe_configuration_recorders([recorder_name], recorders, error_code=error_code) + + if error_code is None: + got_recorders = config.describe_configuration_recorders([recorder_name]) + assert [gr["name"] for gr in got_recorders] == [r["name"] for r in recorders] + else: + with pytest.raises(ClientError) as exc_info: + config.describe_configuration_recorders([recorder_name]) + assert exc_info.value.response["Error"]["Code"] == error_code + + +@pytest.mark.parametrize("error_code", [None, "TestException"]) +def test_start_configuration_recorder(make_stubber, error_code): + config_client = boto3.client("config") + config_stubber = make_stubber(config_client) + config = ConfigWrapper(config_client) + recorder_name = "test-recorder" + + config_stubber.stub_start_configuration_recorder(recorder_name, error_code=error_code) + + if error_code is None: + config.start_configuration_recorder(recorder_name) + else: + with pytest.raises(ClientError) as exc_info: + config.start_configuration_recorder(recorder_name) + assert exc_info.value.response["Error"]["Code"] == error_code + + +@pytest.mark.parametrize("error_code", [None, "TestException"]) +def test_list_discovered_resources(make_stubber, error_code): + config_client = boto3.client("config") + config_stubber = make_stubber(config_client) + config = ConfigWrapper(config_client) + resource_type = "AWS::S3::Bucket" + resources = [{"resourceType": resource_type, "resourceId": "test-bucket"}] + + config_stubber.stub_list_discovered_resources(resource_type, resources, error_code=error_code) + + if error_code is None: + got_resources = config.list_discovered_resources(resource_type) + assert len(got_resources) == len(resources) + else: + with pytest.raises(ClientError) as exc_info: + config.list_discovered_resources(resource_type) + assert exc_info.value.response["Error"]["Code"] == error_code diff --git a/scenarios/basics/config/SPECIFICATION.md b/scenarios/basics/config/SPECIFICATION.md new file mode 100644 index 00000000000..7c43e91c14c --- /dev/null +++ b/scenarios/basics/config/SPECIFICATION.md @@ -0,0 +1,92 @@ +AWS Config Specification + +This document contains a draft proposal for an AWS Config Basics Scenario, generated by the Code Examples SpecGen AI tool. The specifications describe a potential code example scenario based on research, usage data, service information, and AI-assistance. The following should be reviewed for accuracy and correctness before proceeding on to a final specification. + +Relevant documentation + +* Getting Started with AWS Config +* What Is AWS Config? +* AWS Config API Reference +* AWS Config Pricing + +API Actions Used + +* PutConfigurationRecorder +* PutDeliveryChannel +* StartConfigurationRecorder +* DescribeConfigurationRecorders +* DescribeConfigurationRecorderStatus +* ListDiscoveredResources +* GetResourceConfigHistory +* StopConfigurationRecorder + +Proposed example structure + +The output below demonstrates how this example would run for the customer. It includes a Hello service example (included for all services), and the scenario description. The scenario code would also be presented as Action snippets, with a code snippet for each SDK action. + +Hello + +The Hello example is a separate runnable example. - Set up the Config service client - Check if Config is available in the current region - List any existing configuration recorders + +Scenario + +Setup + +* Create a configuration recorder to track AWS resource configurations +* Set up a delivery channel to specify where Config sends configuration snapshots +* Start the configuration recorder to begin monitoring resources + +Configuration Monitoring + +* Verify the configuration recorder is running +* Display recorder status and settings +* Show which resource types are being monitored + +Resource Discovery + +* List discovered AWS resources in the account +* Display resource types and counts +* Show configuration details for specific resources + +Configuration History + +* Retrieve configuration history for a specific resource +* Display configuration changes over time +* Show compliance and configuration drift information + +Cleanup + +* Stop the configuration recorder (with user confirmation) +* Optionally delete the configuration recorder and delivery channel +* Display final status + +Errors + +SDK Code examples include basic exception handling for each action used. The table below describes an appropriate exception which will be handled in the code for each service action. + +Action Error Handling +PutConfigurationRecorder InvalidConfigurationRecorderNameException Validate recorder name format and uniqueness. +PutConfigurationRecorder MaxNumberOfConfigurationRecordersExceededException Notify user of recorder limit and suggest cleanup. +PutDeliveryChannel InvalidDeliveryChannelNameException Validate delivery channel name format. +PutDeliveryChannel InvalidS3BucketNameException Validate S3 bucket name and permissions. +StartConfigurationRecorder NoSuchConfigurationRecorderException Verify recorder exists before starting. +StartConfigurationRecorder NoAvailableDeliveryChannelException Ensure delivery channel is configured. +DescribeConfigurationRecorders NoSuchConfigurationRecorderException Handle case where no recorders exist. +DescribeConfigurationRecorderStatus NoSuchConfigurationRecorderException Handle case where recorder doesn't exist. +ListDiscoveredResources ValidationException Validate resource type and pagination parameters. +GetResourceConfigHistory ResourceNotDiscoveredException Handle case where resource is not tracked by Config. +StopConfigurationRecorder NoSuchConfigurationRecorderException Verify recorder exists before stopping. + +Metadata + +action / scenario metadata file metadata key +PutConfigurationRecorder config_metadata.yaml config_PutConfigurationRecorder +PutDeliveryChannel config_metadata.yaml config_PutDeliveryChannel +StartConfigurationRecorder config_metadata.yaml config_StartConfigurationRecorder +DescribeConfigurationRecorders config_metadata.yaml config_DescribeConfigurationRecorders +DescribeConfigurationRecorderStatus config_metadata.yaml config_DescribeConfigurationRecorderStatus +ListDiscoveredResources config_metadata.yaml config_ListDiscoveredResources +GetResourceConfigHistory config_metadata.yaml config_GetResourceConfigHistory +StopConfigurationRecorder config_metadata.yaml config_StopConfigurationRecorder +AWS Config Basics Scenario config_metadata.yaml config_Scenario +