Cloud Config allows you to create rules based on Function Compute to audit associated resources. When a rule is triggered, Cloud Config invokes the corresponding rule function to evaluate the associated resources and returns the compliance evaluation results of the resources. This topic provides the JSON examples of the code and input parameters of a custom rule function.

Function code

A rule is a piece of logical judgment code that is stored in a rule function. Cloud Config audits resources based on rules. When a rule is triggered, Cloud Config invokes the corresponding rule function to evaluate the associated resources. In this example, the code of the custom rule function contains the following key functions:
  • handler

    The handler function is the entry function that is invoked when the rule is triggered. You must define the handler function when you develop the code of the rule function.

  • put_evaluations

    The put_evaluations function is invoked in the handler function to return the compliance evaluation results.

You can use the following sample Python code to create a rule:
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
import logging
import json
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.auth.credentials import StsTokenCredential
from aliyunsdkcore.acs_exception.exceptions import ClientException
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkcore.request import CommonRequest
logger = logging.getLogger()

# The compliance evaluation results.
COMPLIACE_TYPE_COMPLIANT = 'COMPLIANT'
COMPLIACE_TYPE_NON_COMPLIANT = 'NON_COMPLIANT'
COMPLIACE_TYPE_NOT_APPLICABLE = 'NOT_APPLICABLE'
COMPLIACE_TYPE_INSUFFICIENT_DATA = 'INSUFFICIENT_DATA'

# The entry function, which orchestrates and processes business logic. 
def handler(event, context):
    """
    Process an event.
    :param event: the event.
    :param context: the context.
    :return: the compliance evaluation result.
    """
    # Check whether the event is valid. The code can be copied. 
    evt = validate_event(event)
    if not evt:
        return None

    rule_parameters = evt.get('ruleParameters')
    result_token = evt.get('resultToken')
    invoking_event = evt.get('invokingEvent')
    ordering_timestamp = evt.get('orderingTimestamp')
    annotation = None

    # Initialize the return values. Set the default compliance evaluation result of the rule based on the business scenario. 
    compliance_type = COMPLIACE_TYPE_NOT_APPLICABLE

    # Obtain the configuration of a resource. If a rule is set to be triggered upon configuration changes, the input parameters are valid. After a rule that you create is triggered or you manually trigger a rule, Cloud Config invokes the corresponding rule function to evaluate associated resources one by one. If the configuration of a resource changes, Cloud Config automatically invokes the corresponding rule function to evaluate the resource whose configuration changes. 
    # If a rule is set to be periodically triggered, the input parameters are empty. In this case, you can query the configuration of a resource to be evaluated by calling the corresponding API operation. 
    configuration_item = invoking_event.get('configurationItem')
    account_id = configuration_item.get('accountId')
    resource_id = configuration_item.get('resourceId')
    resource_type = configuration_item.get('resourceType')
    region_id = configuration_item.get('regionId')

    # Evaluate resources. You must implement the evaluation logic based on the business scenario. The following code is only for reference: 
    compliance_type, annotation = evaluate_configuration_item(
        rule_parameters, configuration_item)

    # Set the compliance evaluation result, which uses the following syntax: 
    evaluations = [
        {
            'accountId': account_id,
            'complianceResourceId': resource_id,
            'complianceResourceType': resource_type,
            'complianceRegionId': region_id,
            'orderingTimestamp': ordering_timestamp,
            'complianceType': compliance_type,
            'annotation': annotation
        }
    ]

# Return the compliance evaluation result and write it to Cloud Config. The code can be copied. 
    put_evaluations(context, result_token, evaluations)

    return evaluations

# Evaluate a resource based on input parameters and resource configuration. You must implement the evaluation logic based on the business scenario. The following code is only for reference: 
def evaluate_configuration_item(rule_parameters, configuration_item):
    """
    Evaluate a resource.
    :param rule_parameters: the input parameters of the rule.
    :param configuration_item: the configuration of the resource.
    :return: the compliance evaluation result.
    """
    # Initialize the return values.
    compliance_type = COMPLIACE_TYPE_NOT_APPLICABLE
    annotation = None

    # Obtain the resource type and resource ID.
    resource_type = configuration_item['resourceType']
    full_configuration = configuration_item['configuration']

    # Check whether the configuration is empty.
    if not full_configuration:
        annotation = 'Configuration is empty.'
        return compliance_type, annotation

    # Convert the configuration to a JSON object.
    configuration = parse_json(full_configuration)
    if not configuration:
        annotation = 'Configuration:{} in invald.'.format(full_configuration)
        return compliance_type, annotation

    return compliance_type, annotation


def validate_event(event):
    """
    Check whether an event is valid.
    :param event: the event.
    :return: the JSON object.
    """
    if not event:
        logger.error('Event is empty.')

    evt = parse_json(event)
    logger.info('Loading event: %s .' % evt)

    if 'resultToken' not in evt:
        logger.error('ResultToken is empty.')
        return None

    if 'ruleParameters' not in evt:
        logger.error('RuleParameters is empty.')
        return None

    if 'invokingEvent' not in evt:
        logger.error('InvokingEvent is empty.')
        return None

    return evt


def parse_json(content):
    """
    Convert a JSON string to a JSON object.
    :param content: the JSON string.
    :return: the JSON object.
    """
    try:
        return json.loads(content)
    except Exception as e:
        logger.error('Parse content:{} to json error:{}.'.format(content, e))
        return None


# Return the compliance evaluation result and write it to Cloud Config. The code can be copied. 
def put_evaluations(context, result_token, evaluations):
    """
    Call the Cloud Config API to return the compliance evaluation result and write it to Cloud Config.
    :param context: the Function Compute context.
    :param result_token: the result token.
    :param evaluations: the compliance evaluation result.
    :return: None
    """
    # Enter the AccessKey ID and AccessKey secret of your Alibaba Cloud account. Your Alibaba Cloud account must have the AliyunConfigFullAccess permission. 
    client = AcsClient(
        'LTAI4FxbrTniVtqg1FZW****',
        'C0GXsd6UU6ECUmrsCgPXA6OEvy****',
        'ap-southeast-1',
    )

    # Create a request and set the required parameters. Set the domain parameter to config.ap-southeast-1.aliyuncs.com. 
    request = CommonRequest()
    request.set_domain('config.ap-southeast-1.aliyuncs.com')
    request.set_version('2019-01-08')
    request.set_action_name('PutEvaluations')
    request.add_body_params('ResultToken', result_token)
    request.add_body_params('Evaluations', evaluations)
    request.set_method('POST')

    try:
        response = client.do_action_with_exception(request)
        logger.info('PutEvaluations with request: {}, response: {}.'.format(request, response))
    except Exception as e:
        logger.error('PutEvaluations error: %s' % e)

                

Input parameters

In an event message, the ruleParameters field stores the input parameters of a rule function. Other fields provide the event information when the custom rule is triggered. The following code shows the fields in an event message in the JSON format:
{
    "orderingTimestamp": "Time when the command is executed",
    "invokingEvent": {
        "messageType": "Message type",
        "configurationItem": {
            "accountId": "Alibaba Cloud account ID",
            "arn": "Resource ARN",
            "availabilityZone": "Zone",
            "regionId": "Region ID",
            "configuration": "Detailed resource configuration",
            "configurationDiff": "Resource configuration details before and after the change",
            "relationship": "Related resource",
            "relationshipDiff": "Relationship change",
            "captureTime": "Timestamp when Cloud Config detects a resource change event and generates a log",
            "resourceCreationTime": "Timestamp when the resource is created",
            "resourceStatus": "Resource status",
            "resourceId": "Resource ID",
            "resourceName": "Resource name",
            "resourceType": "Resource type",
            "supplementaryConfiguration": "Supplementary resource configuration",
            "tags": "Tag"
        }
    },
    "ruleParameters": {
        "key": "value"
    },
    "resultToken": "Result token"
}
context: the context information that is automatically set when the custom rule is triggered.
  • context.credentials.access_key_id:"AccessKey ID"
  • context.credentials.access_key_secret:"AccessKey secret"
  • context.region:"Region"