How to query your AWS resource configuration states using AWS Config and Amazon Athena

import datetime
import re
import boto3
import os

TABLE_NAME = 'aws_config_configuration_snapshot'
DATABASE_NAME = 'sampledb'
ACCOUNT_ID = None # Determined at runtime
LATEST_PARTITION_VALUE = 'latest'

athena = boto3.client('athena')

def lambda_handler(event, context):
    global ACCOUNT_ID

    object_key = event['Records'][0]['s3']['object']['key']
    match = get_configuration_snapshot_object_key_match(object_key)
    if match is None:
        print('Ignoring event for non-configuration snapshot object key', object_key)
        return
    print('Adding partitions for configuration snapshot object key', object_key)
    
    ACCOUNT_ID = context.invoked_function_arn.split(':')[4]
    object_key_parent = 's3://{bucket_name}/{object_key_parent}/'.format(
        bucket_name=event['Records'][0]['s3']['bucket']['name'],
        object_key_parent=os.path.dirname(object_key))
    configuration_snapshot_accountid = get_configuration_snapshot_accountid(match)
    configuration_snapshot_region = get_configuration_snapshot_region(match)
    configuration_snapshot_date = get_configuration_snapshot_date(match)
    
    drop_partition(configuration_snapshot_accountid, configuration_snapshot_region, LATEST_PARTITION_VALUE)
    add_partition(configuration_snapshot_accountid, configuration_snapshot_region, LATEST_PARTITION_VALUE, object_key_parent)
    add_partition(configuration_snapshot_accountid, configuration_snapshot_region, get_configuration_snapshot_date(match).strftime('%Y-%m-%d'), object_key_parent)
    
def get_configuration_snapshot_object_key_match(object_key):
    # Matches object keys like AWSLogs/123456789012/Config/us-east-1/2018/4/11/ConfigSnapshot/123456789012_Config_us-east-1_ConfigSnapshot_20180411T054711Z_a970aeff-cb3d-4c4e-806b-88fa14702hdb.json.gz
    return re.match('AWSLogs/(\d+)/Config/([\w-]+)/(\d+)/(\d+)/(\d+)/ConfigSnapshot/[^\\\]+$', object_key)

def get_configuration_snapshot_accountid(match):
    print('AccountId:', match.group(1))
    return match.group(1)

def get_configuration_snapshot_region(match):
    return match.group(2)

def get_configuration_snapshot_date(match):
    return datetime.date(int(match.group(3)), int(match.group(4)), int(match.group(5)))
    
def add_partition(accountid_partition_value, region_partition_value, dt_partition_value, partition_location):
    execute_query('ALTER TABLE {table_name} ADD PARTITION {partition} location \'{partition_location}\''.format(
        table_name=TABLE_NAME,
        partition=build_partition_string(accountid_partition_value, region_partition_value, dt_partition_value),
        partition_location=partition_location))
        
def drop_partition(accountid_partition_value, region_partition_value, dt_partition_value):
    execute_query('ALTER TABLE {table_name} DROP PARTITION {partition}'.format(
        table_name=TABLE_NAME,
        partition=build_partition_string(accountid_partition_value, region_partition_value, dt_partition_value)))
        
def build_partition_string(accountid_partition_value, region_partition_value, dt_partition_value):
    return "(accountid='{accountid_partition_value}', dt='{dt_partition_value}', region='{region_partition_value}')".format(
	    accountid_partition_value=accountid_partition_value,
        dt_partition_value=dt_partition_value,
        region_partition_value=region_partition_value)

def execute_query(query):
    print('Executing query:', query)
    query_output_location = 's3://aws-athena-query-results-{account_id}-{region}'.format(
        account_id=ACCOUNT_ID,
        region=os.environ['AWS_REGION'])
    start_query_response = athena.start_query_execution(
        QueryString=query,
        QueryExecutionContext={
            'Database': DATABASE_NAME
        },
        ResultConfiguration={
            'OutputLocation': query_output_location,
        }
    )
    print('Query started')
    
    is_query_running = True
    while is_query_running:
        get_query_execution_response = athena.get_query_execution(
            QueryExecutionId=start_query_response['QueryExecutionId']
        )
        query_state = get_query_execution_response['QueryExecution']['Status']['State']
        is_query_running = query_state in ('RUNNING','QUEUED')
        
        if not is_query_running and query_state != 'SUCCEEDED':
            raise Exception('Query failed')
    print('Query completed')

aws configservice deliver-config-snapshot --delivery-channel-name channel_name --region region_name 

After the first delivery has occurred, we’re ready to perform queries on your AWS resources in Amazon Athena. Let’s start with a simple example. Run the following command in the Athena Query Editor to list all EC2 instances in your account which are of type “t2.micro”:

SELECT configurationItem.resourceType, configurationItem.resourceId as InstanceId
FROM sampledb.aws_config_configuration_snapshot
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'
AND configurationItem.resourceType = 'AWS::EC2::Instance'
AND json_extract_scalar(configurationItem.configuration, '$.instancetype') = 't2.micro';

Here’s the break-down of this command:

• line 1: Specify the properties you want to return in the query results. Here, we’re returning the resource type and resource ID for each EC2 instance found.
• line 3: The configuration snapshot files exported by AWS Config store configuration items inside property “configurationItems”, which is an array. CROSS JOIN UNNEST is used to flatten this array such that each individual configuration item within the array can easily be queried. Specifying “t(configurationItem)” assigns name “configurationItem” to each of the entries within the array. “t” refers to the name we assign to the row and can be ignored.
• line 4: The “dt” partition key with value “latest” is specified in order to scope the query down to only look at the latest configuration snapshot file in your S3 bucket. As mentioned in the previous section, the Lambda function is automatically keeping partition dt=”latest” to point to the latest configuration snapshot file for a Region. You can always specify dt=”latest” in your queries if you are interested in querying the latest configuration states for your resources. You can also specify a value like dt=”2018-02-28″ in order to search your configuration snapshot data from a specific date.
• line 5: Only look at configuration items for EC2 instances.
• line 6: json_extract_scalar() is used to evaluate the JSON value of the “configuration” property. The “configuration” property of the configuration item for a resource is a JSON object of that resource’s configuration. However, if you look back to the SerDes used in the CREATE EXTERNAL TABLE command, “configuration” is defined as a string. The reason for this is that the model of the “configuration” property differs for each resource type recorded by AWS Config. So, there is no single SerDes that can be defined for this property which applies to all configuration items. Therefore, json_extract_scalar() is used to evaluate the JSON despite not having included the model for the configuration JSON in the Athena table SerDes. In this function call, “$.instancetype” is specified as the JSONPath expression in order to refer to the instanceType property of EC2 instance’s configuration. Note that this path must be lower-case due to the fact that column names used by the function are lower cased.

The above command returns results such as the following:

Another example

The following query acts on a number of additional configuration item properties and uses additional Presto functions, performing a lookup for all Amazon DynamoDB tables containing an owner tag of TeamA or TeamB that have an update stream enabled and are keyed based on more than property.

SELECT configurationItem.resourceType, configurationItem.resourceId as TableName
FROM sampledb.aws_config_configuration_snapshot
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'
AND configurationItem.resourceType = 'AWS::DynamoDB::Table'
AND (configurationItem.tags['owner']='TeamA' OR configurationItem.tags['owner']='TeamB')
AND json_extract_scalar(configurationItem.configuration, '$.streamspecification.streamenabled') = 'true'
AND json_array_length(json_extract(configurationItem.configuration, '$.keyschema')) > 1

Custom resource views

The previous sample queries only return resource identifier information in the query results. However, you can also structure your queries to return customized views of your AWS resources. The following query returns a list of all Amazon S3 buckets in your account, alongside a view of each bucket’s setting for Requester Pays, Bucket Acceleration, and Versioning.

SELECT configurationItem.resourceId as BucketName,
if(configurationItem.supplementaryConfiguration['isrequesterpaysenabled'] = 'true', 'Enabled', 'Off') as RequesterPaysStatus,
coalesce(json_extract_scalar(configurationItem.supplementaryConfiguration['bucketaccelerateconfiguration'], '$.status'), 'Off') as BucketAccelerationStatus,
json_extract_scalar(configurationItem.supplementaryConfiguration['bucketversioningconfiguration'], '$.status') as BucketVersioningStatus
FROM sampledb.aws_config_configuration_snapshot
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'
AND configurationItem.resourceType = 'AWS::S3::Bucket'

Results for this query appear as follows:

Querying data from multiple Regions

This blog post thus-far has provided instructions as if you only use AWS Config in a single AWS Region. However, if you use AWS Config in more than one Region, the aws_config_configuration_snapshot Athena table can seamlessly be made to support queries on these additional Regions by simply setting them all up to export configuration snapshot data to the same Amazon S3 bucket. Simply repeat the steps in section “Enable Amazon S3 configuration snapshot delivery in AWS Config” in all Regions in which you use AWS Config to set this up. You will then be able to perform incredibly powerful, cross-Region queries against your AWS resources.

By default, queries executed against the aws_config_configuration_snapshot table will search configuration snapshot data for all Regions contained within your S3 bucket. Whether your Amazon S3 bucket contains configuration snapshot data for a single Region or multiple Regions, queries executed against the Athena table are written and behave exactly the same. If, however, you would like to target your queries to a configuration snapshot in a specific Region, you can easily do this by specifying a Region in your queries. Recall from the CREATE EXTERNAL TABLE command that we added a “region” partition key to the Athena table. This partition key is being automatically set by the Lambda function we added earlier. If you execute a query that doesn’t include this partition key, Athena searches data for all Regions. For example:

Retrieve the list of all resources recorded by AWS Config in all Regions

SELECT configurationItem.resourceType, configurationItem.resourceId
FROM sampledb.aws_config_configuration_snapshot
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'

Retrieve the list of all resources recorded by AWS Config only in Region us-east-1

SELECT configurationItem.resourceType, configurationItem.resourceId
FROM sampledb.aws_config_configuration_snapshot
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'
AND region = 'us-east-1'

Popular use cases

Before wrapping up, here are a few additional sample queries based on popular use cases to help get you started.

List all instances missing a “Name” tag

SELECT configurationItem.resourceId as InstanceId
FROM sampledb.aws_config_configuration_snapshot
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'
AND configurationItem.resourceType = 'AWS::EC2::Instance'
AND configurationItem.tags['name'] IS NULL

List Windows instances running the SSM Agent that are missing Windows Update “KB4035631”

SELECT configurationItem.resourceType, configurationItem.resourceId as InstanceId
FROM default.awsconfig_region
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'
AND configurationItem.resourceType = 'AWS::SSM::ManagedInstanceInventory'
AND json_extract_scalar(configurationItem.configuration, '$.aws:instanceinformation.content["' || configurationItem.resourceId || '"].platformtype') = 'Windows'
AND json_extract(configurationItem.configuration, '$.aws:windowsupdate.content.kb4035631') IS NULL

Sort your Certificate Manager certificates by their expiration date

SELECT configurationItem.resourceType, configurationItem.resourceId as CertificateArn, json_extract_scalar(configurationItem.configuration, '$.notafter') as CertificateExpiration
FROM sampledb.aws_config_configuration_snapshot
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'
AND configurationItem.resourceType = 'AWS::ACM::Certificate'
ORDER BY CertificateExpiration

List IAM users that are using inline policies

SELECT configurationItem.resourceName as User
FROM sampledb.aws_config_configuration_snapshot
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'
AND configurationItem.resourceType = 'AWS::IAM::User'
AND json_array_length(json_extract(configurationItem.configuration, '$.userpolicylist')) > 0

List your default VPC in each Region

SELECT configurationItem.awsRegion as Region, configurationItem.resourceId as VpcId
FROM sampledb.aws_config_configuration_snapshot
CROSS JOIN UNNEST(configurationitems) AS t(configurationItem)
WHERE dt = 'latest'
AND configurationItem.resourceType = 'AWS::EC2::VPC'
AND json_extract_scalar(configurationItem.configuration, '$.isdefault') = 'true'

Conclusion

Combining the query power of Amazon Athena with the configuration data stored by AWS Config allows for extremely powerful and useful lookups on your AWS resources. I hope this blog post has demonstrated this to you and helps you in managing your AWS resources.

About the Author

Marc Labrecque is an engineer on AWS Config based in Seattle, Washington.