對安全性群組規則的合規性進行審計修複 - Elastic Compute Service

若安全性群組規則對全網段（0.0.0.0/0）開放22（SSH服務）、3389（RDP）等高危風險連接埠，將會給系統帶來嚴重的安全隱患。您可藉助配置審計持續檢測安全性群組配置並自動修複不合規的配置項，確保系統安全。

應用背景

在企業的雲環境中，安全性群組作為網路流量的核心管控手段，通常用於定義伺服器執行個體的訪問規則。然而，在複雜的多執行個體情境下，因營運疏忽或策略設計缺陷，安全性群組規則可能存在以下典型風險配置：

高危連接埠全網段暴露：例如，面向公網（0.0.0.0/0）開放SSH（22連接埠）、RDP（3389連接埠）或資料庫服務連接埠（如3306、6379），導致執行個體直接暴露於互連網，成為暴力破解、資料泄露等攻擊的首要目標。
內網與公網服務混淆：未區分執行個體業務屬性（如公網Web服務與內網資料庫），錯誤對非公網執行個體開放全量IP存取權限，形成內部網路橫向滲透風險。

解決方案

通過阿里雲配置審計服務建立規則，實現對安全性群組規則變更的持續監控，並設定規則檢測諸如22、3389、3306等高危連接埠是否對外開放。一旦發現新增或修改安全性群組規則允許這些連接埠對公網開放，即觸發合規審計。此時，配置審計將自動啟動Function Compute執行自訂修複邏輯，利用阿里雲SDK調整安全性群組設定，例如刪除風險規則。修複後，系統會重新評估相關規則以確認修複效果。此外，使用者可以通過配置審計控制台查看不合規資源的修正詳情，整個過程透明且可追溯，有效防止未經授權的公網訪問。此方案不僅提高了營運效率，減少了人工幹預，還確保資源配置始終符合安全與合規要求，增強了環境的安全性和穩定性。

建立配置審計及Function Compute修複規則

本方案通過Terraform建立配置審計規則，並結合Function Compute實現不合規資源的自動修複，從而達成雲資源合規性的自動化檢測與管理。

說明

如果當前操作使用者為RAM使用者時，請為RAM使用者授予以下許可權。詳細資料請參見為RAM使用者授權。

RAM權限原則

此自訂權限原則允許使用者管理和操作ECS安全性群組規則及Function Compute服務和函數。

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iacservice:CreateExplorerModuleVersion",
        "iacservice:GetExplorerModule",
        "iacservice:CreateExplorerModule",
        "iacservice:ListExplorerModules",
        "iacservice:UpdateExplorerModuleAttribute",
        "iacservice:DeleteExplorerModule"
      ],
      "Resource": "acs:iacservice:*:*:explorermodule/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "iacservice:CreateExplorerTask",
        "iacservice:UpdateExplorerTaskAttribute",
        "iacservice:GetExplorerTask",
        "iacservice:DeleteExplorerTask"
      ],
      "Resource": "acs:iacservice:*:*:explorertask/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "iacservice:CreateJob",
        "iacservice:GetJob",
        "iacservice:listJobs",
        "iacservice:OperateJob"
      ],
      "Resource": "acs:iacservice:*:*:explorertask/*/job/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "iacservice:ListResources",
        "iacservice:ListExplorerHistories",
        "iacservice:CreateExplorerHistory",
        "iacservice:ExportTerraformCode"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "ecs:RevokeSecurityGroup",
        "ecs:DescribeSecurityGroups",
        "ecs:DescribeSecurityGroupAttributes"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "fc:CreateService",
        "fc:DeleteService",
        "fc:UpdateService",
        "fc:CreateFunction",
        "fc:DeleteFunction",
        "fc:UpdateFunction",
        "fc:InvokeFunction",
        "fc:ListServices",
        "fc:ListFunctions",
        "fc:GetService",
        "fc:GetFunction"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": "config:*",
      "Resource": "*"
    }
  ]
}

說明

本教程所含範例程式碼支援一鍵運行，您可以直接運行代碼。一鍵運行

重要

本方案中採用直接刪除不合規安全性群組方式實現自動修正，可能會影響業務的連續性，請根據實際業務修改Function Compute中的修正代碼。

Terraform代碼

variable "region_id" {
  type    = string
  default = "cn-shenzhen"
}
# main.tf


provider "alicloud" {
  region     = var.region_id
}

resource "local_file" "python_script" {
  content  = <<EOF
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
import sys

sys.path.append('/opt/python')
import json
import logging
import jmespath  # 使用jmespath代替jsonpath
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_openapi.client import Client as OpenApiClient
from alibabacloud_openapi_util.client import Client as OpenApiUtilClient
from alibabacloud_tea_util import models as util_models

logger = logging.getLogger()


def handler(event, context):
    logger.info(f"This is event: {str(event, encoding='utf-8')}")
    get_resources_non_compliant(event, context)


def get_resources_non_compliant(event, context):
    # 擷取不合規的資源資訊
    resources = parse_json(event)
    # 遍曆不合規資源，進行修正操作
    for resource in resources:
        remediation(resource, context)


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


def remediation(resource, context):
    logger.info(f"需要修複的資源資訊: {resource}")
    region_id = resource['regionId']
    account_id = resource['accountId']
    resource_id = resource['resourceId']
    resource_type = resource['resourceType']
    if resource_type == 'ACS::ECS::SecurityGroup':
        # 擷取不合規安全性群組的配置資訊，重新校正，確保不合規安全性群組的評估準確性
        resource_result = get_discovered_resource(context, resource_id, resource_type, region_id)
        configuration = json.loads(resource_result["body"]["DiscoveredResourceDetail"]["Configuration"])
        # 判斷是否是託管的安全性群組
        is_managed_security_group = configuration.get('ServiceManaged')
        # 使用jmespath擷取入方向為接受且授權0.0.0.0/0的安全性群組規則id
        delete_security_group_rule_ids = jmespath.search(
            "Permissions.Permission[?SourceCidrIp=='0.0.0.0/0'].SecurityGroupRuleId",
            configuration
        )
        # 非託管的安全性群組，且授權了0.0.0.0/0的入方向安全性群組規則，則刪除
        if is_managed_security_group is False and delete_security_group_rule_ids:
            logger.info(f"注意：刪除安全性群組規則 {region_id}:{resource_id}:{delete_security_group_rule_ids}")
            revoke_security_group(context, region_id, resource_id, delete_security_group_rule_ids)


def revoke_security_group(context, region_id, resource_id, security_group_rule_ids):
    creds = context.credentials
    config = open_api_models.Config(
        access_key_id=creds.access_key_id,
        access_key_secret=creds.access_key_secret,
        security_token=creds.security_token,
        endpoint=f'ecs.{region_id}.aliyuncs.com'
    )
    client = OpenApiClient(config)
    params = open_api_models.Params(
        style='RPC',  # API風格
        version='2014-05-26',  # API版本號碼
        action='RevokeSecurityGroup',  # API 名稱
        method='POST',  # 要求方法
        pathname='/',  # 介面 PATH，RPC介面預設"/"
        protocol='HTTPS',  # 介面協議,
        auth_type='AK',
        req_body_type='json',  # 介面請求體內容格式,
        body_type='json'  # 介面響應體內容格式,
    )
    query = {'RegionId': region_id, 'SecurityGroupId': resource_id, 'SecurityGroupRuleId': security_group_rule_ids}
    # 建立API請求對象
    request = open_api_models.OpenApiRequest(
        query=OpenApiUtilClient.query(query),
    )
    runtime = util_models.RuntimeOptions()
    response = client.call_api(params, request, runtime)
    logger.info(f"刪除結果: {response}")


# 擷取資源詳情
def get_discovered_resource(context, resource_id, resource_type, region_id):
    """
    調用API擷取資源配置詳情
    :param context：Function Compute上下文
    :param resource_id：資源ID
    :param resource_type：資源類型
    :param region_id：資源所屬地區ID
    :return: 資源詳情
    """
    # 需具備許可權AliyunConfigFullAccess的Function ComputeFC的服務角色。
    creds = context.credentials
    config = open_api_models.Config(
        access_key_id=creds.access_key_id,
        access_key_secret=creds.access_key_secret,
        security_token=creds.security_token,
        endpoint='config.cn-shanghai.aliyuncs.com'
    )
    client = OpenApiClient(config)
    params = open_api_models.Params(
        style='RPC',  # API風格
        version='2020-09-07',  # API版本號碼
        action='GetDiscoveredResource',  # API 名稱
        method='POST',  # 要求方法
        pathname='/',  # 介面 PATH，RPC介面預設"/"
        protocol='HTTPS',  # 介面協議,
        auth_type='AK',
        req_body_type='json',  # 介面請求體內容格式,
        body_type='json'  # 介面響應體內容格式,
    )
    query = {'ResourceId': resource_id, 'ResourceType': resource_type, 'Region': region_id}
    # 建立API請求對象
    request = open_api_models.OpenApiRequest(
        query=OpenApiUtilClient.query(query),
    )
    runtime = util_models.RuntimeOptions()
    try:
        response = client.call_api(params, request, runtime)
        return response
    except Exception as e:
        logger.error('GetDiscoveredResource error: %s' % e)

EOF
  filename = "${path.module}/python/index.py"
}

resource "local_file" "requirements_txt" {
  content  = <<EOF
  alibabacloud-tea-openapi
  jmespath>= 0.10.0
  EOF
  filename = "${path.module}/python/requests/requirements.txt"
}
locals {
  code_dir       = "${path.module}/python/"
  archive_output = "${path.module}/code.zip"
  base64_output  = "${path.module}/code_base64.txt"
}

data "archive_file" "code_package" {
  type        = "zip"
  source_dir  = local.code_dir
  output_path = local.archive_output

  depends_on = [
    local_file.python_script,
    local_file.requirements_txt,
  ]
}

resource "null_resource" "upload_code" {
  provisioner "local-exec" {
    command = <<EOT
    base64 -w 0 ${local.archive_output} > ${local.base64_output}
    EOT

    interpreter = ["sh", "-c"]
  }

  depends_on = [data.archive_file.code_package]
}

data "local_file" "base64_encoded_code" {
  filename   = local.base64_output
  depends_on = [null_resource.upload_code]
}
resource "alicloud_fcv3_function" "fc_function" {
  runtime       = "python3.10"
  handler       = "index.handler"
  function_name = "HHM-FC-TEST"
  role          = alicloud_ram_role.role.arn

  code {
    zip_file = data.local_file.base64_encoded_code.content
  }
  lifecycle {
    ignore_changes = [
      code
    ]
  }

  # 顯式設定 log_config 為空白
  log_config {}

  depends_on = [data.local_file.base64_encoded_code]
}

resource "alicloud_config_rule" "default" {
  rule_name    = "SPM0014安全性群組不允許對全部網段開啟風險連接埠"
  description  = "禁止安全性群組對所有網段開放風險連接埠22, 3389"
  source_owner = "ALIYUN"
  # （必需，ForceNew）指定是您還是阿里雲擁有並管理該規則。有效值： CUSTOM_FC: 該規則是自訂規則，您擁有該規則 ● ALIYUN: 該規則是託管規則，阿里雲擁有該規則。
  source_identifier = "sg-risky-ports-check"
  #配置審計ARN（必需，ForceNew）規則的標識符。對於託管規則，值為託管規則的名稱。對於自訂規則，值為自訂規則的ARN。
  resource_types_scope = ["ACS::ECS::SecurityGroup"]
  #規則監控被排除的資源ID，多個ID用逗號分隔，僅適用於基於託管規則建立的規則，定製規則此欄位為空白。
  config_rule_trigger_types = "ConfigurationItemChangeNotification" #規則在配置更改時被觸發
  #有效值包括：One_Hour、Three_Hours、Six_Hours、Twelve_Hours、TwentyFour_Hours。
  risk_level = 1 #    ● 1: 嚴重 ● 2: 警告● 3: 資訊

  input_parameters = {
    "ports" : "22,3389"
  }
}

resource "alicloud_config_remediation" "default" {
  config_rule_id          = alicloud_config_rule.default.id
  remediation_template_id = alicloud_fcv3_function.fc_function.function_arn
  remediation_source_type = "CUSTOM"
  invoke_type             = "AUTO_EXECUTION"
  params                  = "{}"
  remediation_type        = "FC"
}

resource "random_integer" "default" {
  min = 10000
  max = 99999
}

resource "alicloud_ram_role" "role" {
  name        = "tf-example-role-${random_integer.default.result}"
  document    = <<EOF
   {
    "Statement": [
       {
        "Action": "sts:AssumeRole",
        "Effect": "Allow",
        "Principal": {
          "Service": [
            "fc.aliyuncs.com"
          ]
        }
      }
    ],
    "Version": "1"
  }
  EOF
  description = "Ecs ram role."
  force       = true
}
resource "alicloud_ram_policy" "policy" {
  policy_name     = "tf-example-ram-policy-${random_integer.default.result}"
  policy_document = <<EOF
   {
    "Statement": [
       {
        "Action": [
          "config:GetDiscoveredResource",
          "ecs:RevokeSecurityGroup"
        ],
        "Effect":  "Allow",
        "Resource": ["*"]
      }
    ],
    "Version": "1"
  }
  EOF
  description     = "this is a policy test"
  force           = true
}

resource "alicloud_ram_role_policy_attachment" "attach" {
  policy_name = alicloud_ram_policy.policy.policy_name
  policy_type = "Custom"
  role_name   = alicloud_ram_role.role.name
}