通过轻量消息队列（原 MNS）通知机制实现不合规资源的自动修复

应用场景

数据规划

操作流程

操作步骤

1. 登录配置审计控制台，设置资源合规事件投递到轻量消息队列（原 MNS）的指定主题（Topic），例如：MNSTestConfig。

   具体操作，请参见设置投递数据到轻量消息队列（原 MNS）。

2. 新建服务。

   1. 登录函数计算控制台。

   2. 在左侧导航栏，单击服务及函数。

   3. 在顶部菜单栏，选择地域，例如：华东2（上海）。

   4. 在服务列表页面，单击创建服务。

   5. 在创建服务面板，名称输入resource_repair。

   6. 单击确定。

3. 授权目标服务修改OSS存储空间的权限。

   1. 在服务resource_repair的左侧导航栏，单击服务详情。

   2. 在角色配置区域，单击编辑。

   3. 选择一个包含权限策略AliyunOSSFullAccess的服务角色。

      如果您没有符合要求的服务角色，请单击创建新的服务角色，在访问控制RAM控制台创建角色，该角色的可信实体类型必须选择阿里云服务、授信服务必须选择函数计算，权限策略必须选择AliyunOSSFullAccess。具体操作，请参见创建可信实体为阿里云服务的RAM角色。

   4. 单击保存。

4. 新建函数。

   1. 在服务resource_repair的左侧导航栏，单击函数管理。

   2. 单击创建函数。

   3. 在创建函数页面，函数名称输入oss_repair_acl_trigger，运行环境选择Python 3.6，其他参数保持默认值。

   4. 单击创建。

5. 配置函数的环境变量。

   1. 在函数oss_repair_acl_trigger的函数代码页签，单击函数配置页签。

   2. 在环境变量区域，单击编辑。

   3. 单击添加变量，输入该环境变量的变量名称和变量值。

      • 变量为prepareRuleId。

      • 值为cr-a6129bc09da7009675a0为配置审计创建的规则ID。

   4. 单击确定。

6. 新建触发器。

   1. 在函数oss_repair_acl_trigger的函数配置页签，单击触发器管理页签。

   2. 单击创建触发器。

   3. 选择触发器类型为轻量消息队列。

   4. 设置轻量消息队列（原 MNS）触发器的相关参数。

      参数设置如下：

      • 名称输入ConfigRuleNonComplianceMNSTrigger。

      • MNS地域选择华东2（上海）。

      • 主题选择MNSTestConfig。

      • Event格式选择STREAM。

      • 角色名称选择AliyunMNSNotificationRole。

   5. 单击确定。

      新建触发器完成后，当配置审计对目标资源进行评估时，您会接收到该资源的不合规事件通知。

7. 配置自动修复代码。

   1. 在函数oss_repair_acl_trigger的触发器管理页签，单击函数代码页签。

   2. 单击文件index.py。

   3. 拷贝并粘贴如下代码至文件index.py。

      # -*- coding: utf-8 -*-
      import logging
      import json
      import os
      import oss2
      
      logger = logging.getLogger()
      
      # 环境变量配置的规则ID（例如：cr-a6129bc09da7009675a0）
      # 注意：这里的值应与阿里云Config规则中的configRuleId一致，
      # 而不是规则名称。这样才能确保函数只对特定的规则触发事件进行处理。
      ENV_RULE_ID = 'prepareRuleId'
      
      # 固定值，用于设置OSS Bucket的ACL为私有
      BUCKET_ACL_PRIVATE = oss2.BUCKET_ACL_PRIVATE
      
      
      def handler(event, context):
          """
          主处理函数，接收来自MNS或直接调用的事件，并根据事件内容决定是否修复OSS Bucket ACL。
      
          :param event: 触发事件数据（JSON字符串）
          :param context: 上下文对象，包含临时凭证等信息
          :return: 'success' 或 'fail'
          """
          logger.info("Received event: %s", event)
      
          try:
              # 尝试解析事件为JSON格式
              notify_json = json.loads(event)
          except json.JSONDecodeError as e:
              # 如果解析失败，则记录错误并返回失败状态
              logger.error("Failed to parse event as JSON: %s", str(e))
              return 'fail'
      
          # 检查通知JSON是否为非空列表
          if not isinstance(notify_json, list) or len(notify_json) == 0:
              logger.error("Invalid event format: expected a non-empty list.")
              return 'fail'
      
          # 获取环境变量中配置的规则ID
          expected_rule_id = os.environ.get(ENV_RULE_ID)
          if not expected_rule_id:
              logger.warning("Environment variable '%s' is not set.", ENV_RULE_ID)
              return 'fail'
      
          for item in notify_json:
              config_rule_id = item.get('configRuleId')
              bucket_name = item.get('resourceId')
              region = item.get('regionId')
      
              logger.info(f"Processing rule ID: {config_rule_id}, Bucket: {bucket_name}, Region: {region}")
      
              # 判断规则ID是否匹配
              if config_rule_id != expected_rule_id:
                  logger.info(f"Rule ID '{config_rule_id}' does not match expected '{expected_rule_id}'. Skip.")
                  continue
      
              # 检查必要的字段是否存在
              if not bucket_name or not region:
                  logger.warning("Missing resourceId or regionId. Skip.")
                  continue
      
              # 执行修复操作
              try:
                  remedy_by_fc_assume(context, region, bucket_name)
              except Exception as ex:
                  logger.exception("Failed to remedy bucket: %s", str(ex))
                  return 'fail'
      
          return 'success'
      
      
      def remedy_by_fc_assume(context, region, bucket_name):
          """
          使用FC提供的临时凭据访问OSS，并修改Bucket ACL为私有。
      
          :param context: 函数计算上下文对象
          :param region: OSS Bucket所在的区域
          :param bucket_name: OSS Bucket名称
          """
          creds = context.credentials
          auth = oss2.StsAuth(
              creds.access_key_id,
              creds.access_key_secret,
              creds.security_token
          )
          endpoint = f'http://oss-{region}.aliyuncs.com'
          bucket = oss2.Bucket(auth, endpoint, bucket_name)
      
          # 设置Bucket ACL为私有
          bucket.put_bucket_acl(BUCKET_ACL_PRIVATE)
          logger.info(f'Bucket {bucket_name} in {region} ACL set to private.')

      说明

      本段代码以规则名称prepareRuleName为例，为您介绍不合规资源的自动修复方法。如需通过其他参数修复，请参见资源不合规事件内容示例。

   4. 单击左上角的部署代码 。

8. 十分钟后，查看修复结果。

   说明

   当资源配置无变更，且审计结果为不合规时，您还需要手动执行审计，然后再执行此步骤。手动执行审计的操作方法，请参见手动执行审计。

   • 通过配置审计控制台查看

   • 通过OSS控制台查看