Cloud Config:カスタム関数ルールの定義と実行

ルールは、ルール関数に格納されている論理判断コードのスニペットです。Cloud Config はルールに基づいてリソースを監査します。ルールがトリガーされると、Cloud Config は対応するルール関数を呼び出して関連リソースを評価します。この例では、カスタムルール関数のコードには次の主要な関数が含まれています。

• handler

  handler 関数は、ルールがトリガーされたときに呼び出されるエントリ関数です。Function Compute コンソールで関数を作成するときは、ハンドラ関数を構成する必要があります。詳細については、「ハンドラ」をご参照ください。

  説明

  Cloud Config はイベントハンドラのみをサポートしています。詳細については、「イベントハンドラ」をご参照ください。

• put_evaluations

  PutEvaluations 操作を呼び出すことによって、リソース評価結果が Cloud Config に返されます。

次の Python コード例を使用してルールを作成できます。

# # !/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 evaluation types of resources. 
COMPLIANCE_TYPE_COMPLIANT = 'COMPLIANT'
COMPLIANCE_TYPE_NON_COMPLIANT = 'NON_COMPLIANT'
COMPLIANCE_TYPE_NOT_APPLICABLE = 'NOT_APPLICABLE'
# The types of pushed resource configurations. 
CONFIGURATION_TYPE_COMMON = 'COMMON'
CONFIGURATION_TYPE_OVERSIZE = 'OVERSIZE'
CONFIGURATION_TYPE_NONE = 'NONE'


# The entry function that 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. You can directly copy the following sample code. 
    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')

    # Obtain the resource configurations. A rule must be set to be triggered upon configuration changes. After a rule that you create is triggered or you manually trigger a rule, Cloud Config invokes the rule function to evaluate associated resources one by one. If the configurations of a resource change, Cloud Config automatically invokes the corresponding rule function to evaluate the resource. 
    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 resource configurations that you want to push is no less than 100 KB. If yes, the GetDiscoveredResource operation is called 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 resources. You must implement the evaluation logic based on your business scenario. The following code is only for reference: 
    compliance_type, annotation = evaluate_configuration_item(
        rule_parameters, configuration_item)

    # Specify the compliance evaluation result. The result must use the syntax shown in the following sample: 
    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. You can directly copy the following sample code. 
    put_evaluations(creds, result_token, evaluations)
    return evaluations


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

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

    # Convert the configurations to a JSON object.
    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):
    """
    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 into 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. You can directly copy the following sample code. 
def put_evaluations(creds, result_token, evaluations):
    """
    Call the Cloud Config API to return the compliance evaluation result and write the result to Cloud Config. 
    :param context: the Function Compute context.
    :param result_token: the result token.
    :param evaluations: the compliance evaluation result.
    :return: None
    """
    # Your account must be assigned a service role of Function Compute to which the AliyunConfigFullAccess policy is attached. 
    client = AcsClient(creds.access_key_id, creds.access_key_secret, region_id='ap-southeast-1')

    # Create a request and specify the parameters. In this example, the domain parameter is set 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.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)


# Obtain the details of a resource. You can copy the following sample code based on your business scenario. 
def get_discovered_resource(creds, resource_id, resource_type, region_id):
    """
    Call an API operation to obtain the details of a resource.
    :param context: the Function Compute context.
    :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.
    """
    # Your account must be assigned a service role of Function Compute to which the AliyunConfigFullAccess policy is 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)