Cloud Config:Custom function rules

A rule is evaluation logic stored in a function. Cloud Config invokes this function during auditing. The function code has two parts:

• handler

  The handler is the entry function invoked when the rule triggers. You must specify the handler when creating a function in the Function Compute console. Handlers.

  Note

  Cloud Config supports only event handlers.

• put_evaluations

  This function calls the PutEvaluations API operation to submit the compliance evaluation result to Cloud Config.

The following code provides a Python example:

# #!/usr/bin/env python
# # -*- encoding: utf-8 -*-
import json
import logging

from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest


logger = logging.getLogger()
# The compliance type of a resource.
COMPLIANCE_TYPE_COMPLIANT = 'COMPLIANT'
COMPLIANCE_TYPE_NON_COMPLIANT = 'NON_COMPLIANT'
COMPLIANCE_TYPE_NOT_APPLICABLE = 'NOT_APPLICABLE'
# The delivery type of a resource configuration.
CONFIGURATION_TYPE_COMMON = 'COMMON'
CONFIGURATION_TYPE_OVERSIZE = 'OVERSIZE'
CONFIGURATION_TYPE_NONE = 'NONE'


# The entry function that orchestrates and processes the business logic.
def handler(event, context):
    """
    The handler function.
    :param event: The event.
    :param context: The context.
    :return: The evaluation result.
    """
    # Validate the event. You can copy this code block.
    evt = validate_event(event)
    if not evt:
        return None
    creds = context.credentials
    rule_parameters = evt.get('ruleParameters')
    result_token = evt.get('resultToken')
    invoking_event = evt.get('invokingEvent')
    ordering_timestamp = evt.get('orderingTimestamp')

    # The configuration of the resource. The trigger of the rule must be set to Configuration Change. When you create or manually run a rule, Cloud Config invokes the function to evaluate all resources one by one. If the configuration of a resource changes, Cloud Config automatically invokes the function to evaluate the resource based on the change information.
    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')

    # Check whether the size of the delivered resource configuration is greater than or equal to 100 KB. If it is, you must call the GetDiscoveredResource operation to query the resource details.
    configuration_type = invoking_event.get('configurationType')
    if configuration_type and configuration_type == CONFIGURATION_TYPE_OVERSIZE:
        resource_result = get_discovered_resource(creds, resource_id, resource_type, region_id)
        resource_json = json.loads(resource_result)
        configuration_item["configuration"] = resource_json["DiscoveredResourceDetail"]["Configuration"]

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

    # Set the evaluation result. The format must comply with the following example.
    evaluations = [
        {
            'accountId': account_id,
            'complianceResourceId': resource_id,
            'complianceResourceType': resource_type,
            'complianceRegionId': region_id,
            'orderingTimestamp': ordering_timestamp,
            'complianceType': compliance_type,
            'annotation': annotation
        }
    ]

    # Submit the evaluation result to Cloud Config. You can copy this code block.
    put_evaluations(creds, result_token, evaluations)
    return evaluations


# Evaluate the resource based on the rule information and resource configuration. You must implement the evaluation logic based on your business requirements. The following code is for reference only.
def evaluate_configuration_item(rule_parameters, configuration_item):
    """
    The evaluation logic.
    :param rule_parameters: The rule information.
    :param configuration_item: The resource configuration.
    :return: The compliance type.
    """
    # Initialize the return value.
    compliance_type = COMPLIANCE_TYPE_NON_COMPLIANT
    annotation = None

    # Obtain the full resource configuration.
    full_configuration = configuration_item['configuration']
    if not full_configuration:
        annotation = 'Configuration is empty.'
        return compliance_type, annotation

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


def validate_event(event):
    """
    Validate the event.
    :param event: The event.
    :return: A 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 data to the JSON format.
    :param content: A JSON string.
    :return: A JSON object.
    """
    try:
        return json.loads(content)
    except Exception as e:
        logger.error('Parse content:{} to json error:{}.'.format(content, e))
        return None


# Submit the evaluation result to Cloud Config. You can copy this code block.
def put_evaluations(creds, result_token, evaluations):
    """
    Call an API operation to return and write the evaluation result.
    :param creds: The credentials of the function.
    :param result_token: The result token.
    :param evaluations: The evaluation results.
    :return: None
    """
    # The service-linked role for Function Compute must have the AliyunConfigFullAccess policy attached.
    client = AcsClient(creds.access_key_id, creds.access_key_secret, region_id='ap-southeast-1')

    # Create a request and set its parameters. The domain name is 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.add_body_params('SecurityToken', creds.security_token)
    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)


# Retrieve the full resource details. You can copy this code block.
def get_discovered_resource(creds, resource_id, resource_type, region_id):
    """
    Call an API operation to obtain the resource configuration details.
    :param creds: The credentials of the function.
    :param resource_id: The resource ID.
    :param resource_type: The resource type.
    :param region_id: The ID of the region in which the resource resides.
    :return: The resource details.
    """
    # The service-linked role for Function Compute must have the AliyunConfigFullAccess policy attached.
    client = AcsClient(creds.access_key_id, creds.access_key_secret, region_id='ap-southeast-1')

    request = CommonRequest()
    request.set_domain('config.ap-southeast-1.aliyuncs.com')
    request.set_version('2020-09-07')
    request.set_action_name('GetDiscoveredResource')
    request.add_query_param('ResourceId', resource_id)
    request.add_query_param('ResourceType', resource_type)
    request.add_query_param('Region', region_id)
    request.add_query_param('SecurityToken', creds.security_token)
    request.set_method('GET')

    try:
        response = client.do_action_with_exception(request)
        resource_result = str(response, encoding='utf-8')
        return resource_result
    except Exception as e:
        logger.error('GetDiscoveredResource error: %s' % e)