from .base_agent import BaseAgent from ..aws_connector import AWSConnector import time class ConfigAgent(BaseAgent): """ An agent specialized in handling AWS Config tasks. """ def execute(self, command: str, **kwargs): """ Executes a given command related to Config. """ if command == 'smart_enable_config': region = kwargs.get('region', '') s3_bucket = kwargs.get('s3_bucket', '') print(f"You are about to smart-enable AWS Config in {region}, using S3 bucket '{s3_bucket}'.") confirm = input("This will create a configuration recorder and delivery channel. Are you sure you want to proceed? (yes/no): ") if confirm.lower() == 'yes': return self._smart_enable_config(**kwargs) else: return {"status": "cancelled", "message": "Smart Enable AWS Config command cancelled by user."} elif command == 'list_compliance_by_resource': region = kwargs.get('region', '') print(f"You are about to list non-compliant resources in {region}.") confirm = input("This will retrieve compliance data from AWS Config. Are you sure you want to proceed? (yes/no): ") if confirm.lower() == 'yes': return self._list_compliance_by_resource(**kwargs) else: return {"status": "cancelled", "message": "List compliance by resource command cancelled by user."} else: raise NotImplementedError(f"Command '{command}' is not supported by ConfigAgent.") def _smart_enable_config(self, region: str, s3_bucket: str): """ Enables AWS Config in the specified region. - Creates a configuration recorder. - Creates a delivery channel. - Starts the configuration recorder. """ print(f"ConfigAgent: Smart enabling AWS Config in region {region}...") try: config_client = AWSConnector.get_client('config', region_name=region) iam_client = AWSConnector.get_client('iam', region_name=region) s3_client = AWSConnector.get_client('s3', region_name=region) # 1. Create S3 bucket if it doesn't exist try: s3_client.head_bucket(Bucket=s3_bucket) print(f"Using existing S3 bucket: {s3_bucket}") except s3_client.exceptions.ClientError as e: if e.response['Error']['Code'] == '404': print(f"Creating S3 bucket for AWS Config: {s3_bucket}") if region == "us-east-1": s3_client.create_bucket(Bucket=s3_bucket) else: location = {'LocationConstraint': region} s3_client.create_bucket(Bucket=s3_bucket, CreateBucketConfiguration=location) else: raise # 2. Create IAM Role for AWS Config config_role_name = 'aws-config-service-role' assume_role_policy_document = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "config.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } try: create_role_response = iam_client.create_role( RoleName=config_role_name, AssumeRolePolicyDocument=json.dumps(assume_role_policy_document), Description='IAM role for AWS Config' ) config_role_arn = create_role_response['Role']['Arn'] iam_client.attach_role_policy( RoleName=config_role_name, PolicyArn='arn:aws:iam::aws:policy/service-role/AWSConfigRole' ) print(f"Created IAM role: {config_role_arn} and attached policy.") time.sleep(10) # Give IAM time to propagate except iam_client.exceptions.EntityAlreadyExistsException: print(f"IAM role '{config_role_name}' already exists. Fetching ARN...") config_role_arn = iam_client.get_role(RoleName=config_role_name)['Role']['Arn'] # 3. Create Configuration Recorder recorder_name = 'default' try: config_client.put_configuration_recorder( ConfigurationRecorder={ 'name': recorder_name, 'roleARN': config_role_arn, 'recordingGroup': { 'allSupported': True, 'includeGlobalResourceTypes': True } } ) print(f"Configuration recorder '{recorder_name}' created.") except config_client.exceptions.MaxNumberOfConfigurationRecordersExceededException: print(f"Configuration recorder '{recorder_name}' already exists. Using existing.") # 4. Create Delivery Channel delivery_channel_name = 'default' try: config_client.put_delivery_channel( DeliveryChannel={ 'name': delivery_channel_name, 's3BucketName': s3_bucket } ) print(f"Delivery channel '{delivery_channel_name}' created.") except config_client.exceptions.MaxNumberOfDeliveryChannelsExceededException: print(f"Delivery channel '{delivery_channel_name}' already exists. Using existing.") # 5. Start Configuration Recorder config_client.start_configuration_recorder(ConfigurationRecorderName=recorder_name) print(f"Configuration recorder '{recorder_name}' started.") return {"status": "success", "message": f"AWS Config enabled in region {region}, recording to S3 bucket '{s3_bucket}'."} except Exception as e: print(f"Error during smart AWS Config enablement: {e}") return {"status": "error", "message": str(e)} def _list_compliance_by_resource(self, region: str, resource_type: str = None, compliance_type: str = 'NON_COMPLIANT'): """ Lists non-compliant resources based on AWS Config rules. """ print(f"ConfigAgent: Listing {compliance_type} resources in region {region}...") try: config_client = AWSConnector.get_client('config', region_name=region) paginator = config_client.get_paginator('get_compliance_details_by_resource') non_compliant_resources = [] # First, list all non-compliant resources compliance_summary = config_client.get_compliance_summary_by_resource_type() resource_types_to_check = [] if resource_type: resource_types_to_check.append(resource_type) else: for item in compliance_summary['ComplianceSummaryByResourceType']: if item['ComplianceSummary']['ComplianceType'] == 'NON_COMPLIANT': resource_types_to_check.append(item['ResourceType']) for r_type in resource_types_to_check: response = config_client.get_compliance_details_by_resource( ResourceType=r_type, ComplianceTypes=[compliance_type] ) for evaluation in response['EvaluationResults']: non_compliant_resources.append({ 'ResourceType': evaluation['EvaluationResultIdentifier']['EvaluatedResourceDetail']['ResourceType'], 'ResourceId': evaluation['EvaluationResultIdentifier']['EvaluatedResourceDetail']['ResourceId'], 'ComplianceType': evaluation['ComplianceType'], 'ConfigRuleName': evaluation['EvaluationResultIdentifier']['ConfigRuleName'], 'Annotation': evaluation.get('Annotation', 'N/A') }) if not non_compliant_resources: return {"status": "success", "message": f"No {compliance_type} resources found in region {region}."} return {"status": "success", "message": f"Found {len(non_compliant_resources)} {compliance_type} resources in region {region}.", "resources": non_compliant_resources} except Exception as e: print(f"Error listing compliance by resource: {e}") return {"status": "error", "message": str(e)}