如何通过CloudFormation或Lambda为OpenSearch服务添加访问IP地址？

你的现有OpenSearch访问策略配置如下：

{
  "Version": "2008-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "es:*",
      "Resource": "arn:aws:es:us-east-1:1234:domain/esupport/*",
      "Condition": {
        "IpAddress": {
          "aws:SourceIp": [
            "1.2.3.4",
            "5.6.7.8"
          ]
        }
      }
    }
  ]
}

完全可以通过CloudFormation模板或Lambda Function URL实现动态添加IP的需求，以下是具体方案：

方案一：CloudFormation模板管理IP白名单 #

通过CloudFormation将IP列表抽象为参数，后续仅需更新参数即可完成IP添加，无需手动编辑策略：

1. 编写CloudFormation模板

AWSTemplateFormatVersion: '2010-09-09'
Parameters:
  OpenSearchDomainName:
    Type: String
    Default: esupport
  AllowedSourceIps:
    Type: CommaDelimitedList
    Default: "1.2.3.4,5.6.7.8"
Resources:
  OpenSearchDomain:
    Type: AWS::OpenSearchService::Domain
    Properties:
      DomainName: !Ref OpenSearchDomainName
      AccessPolicies:
        !Sub |
          {
            "Version": "2008-10-17",
            "Statement": [
              {
                "Effect": "Allow",
                "Principal": {
                  "AWS": "*"
                },
                "Action": "es:*",
                "Resource": "arn:aws:es:us-east-1:1234:domain/${OpenSearchDomainName}/*",
                "Condition": {
                  "IpAddress": {
                    "aws:SourceIp": ${AllowedSourceIps}
                  }
                }
              }
            ]
          }

2. 更新IP白名单
   当需要添加新IP时，只需更新AllowedSourceIps参数值（例如改为"1.2.3.4,5.6.7.8,9.10.11.12"），然后执行CloudFormation栈更新操作即可自动替换访问策略。

方案二：Lambda Function URL动态添加IP #

通过Lambda函数调用AWS SDK修改OpenSearch的访问策略，并通过Function URL提供HTTP接口实现IP添加：

1. 创建Lambda函数
   使用Python编写函数逻辑，注意提前给Lambda授予es:UpdateDomainConfig权限：

import boto3
import json

client = boto3.client('opensearch')
DOMAIN_NAME = 'esupport'

def lambda_handler(event, context):
    # 从请求体获取要添加的IP
    try:
        request_body = json.loads(event['body'])
        new_ip = request_body.get('ip')
        if not new_ip:
            return {'statusCode': 400, 'body': json.dumps({'error': 'Missing IP parameter'})}
        
        # 获取当前访问策略
        response = client.describe_domain_config(DomainName=DOMAIN_NAME)
        current_policy = json.loads(response['DomainConfig']['AccessPolicies']['Options'])
        
        # 检查IP是否已存在
        existing_ips = current_policy['Statement'][0]['Condition']['IpAddress']['aws:SourceIp']
        if new_ip in existing_ips:
            return {'statusCode': 200, 'body': json.dumps({'message': 'IP already exists'})}
        
        # 添加新IP
        existing_ips.append(new_ip)
        updated_policy = json.dumps(current_policy)
        
        # 更新域名配置
        client.update_domain_config(
            DomainName=DOMAIN_NAME,
            AccessPolicies=updated_policy
        )
        
        return {'statusCode': 200, 'body': json.dumps({'message': 'IP added successfully', 'current_ips': existing_ips})}
    except Exception as e:
        return {'statusCode': 500, 'body': json.dumps({'error': str(e)})}

2. 配置Lambda Function URL
   在Lambda控制台为该函数创建Function URL，配置合适的授权方式（例如选择AWS IAM授权限制调用者，或设置自定义访问控制），即可通过HTTP请求调用该接口添加IP。

内容的提问来源于stack exchange，提问作者shantanuo