Elasticsearch检查索引是否存在时触发认证错误求助

嘿，我来帮你搞定这个问题！你遇到的认证错误，核心原因很明确：当Elasticsearch配置了Cognito用户池认证时，直接通过http_auth传用户名密码的基本HTTP认证是完全无效的——Cognito采用的是OAuth2/OpenID Connect的认证体系，得用它的专属流程来获取有效令牌，或者结合AWS IAM的签名认证来访问ES。

下面给你两种最常用的解决方案，你可以根据自己的场景选：

方案1：用Cognito用户名密码获取ID Token认证 #

这个方案适合你想用指定Cognito用户身份访问ES的场景，步骤是先调用Cognito的API拿到用户的ID Token，然后把这个Token作为Bearer令牌放到请求头里去连接ES。

直接给你改好的代码，替换掉对应的配置项就能用：

from __future__ import print_function
import sys
import boto3
from elasticsearch import Elasticsearch, RequestsHttpConnection

# 替换成你的实际配置
COGNITO_USER_POOL_ID = 'us-east-1_XXXXXXX'  # 你的Cognito用户池ID
COGNITO_CLIENT_ID = 'XXXXXXXXXXXXXXXXXXXX'  # 用户池里创建的客户端ID
USERNAME = 'your-cognito-username'
PASSWORD = 'your-cognito-password'
ES_ENDPOINT = 'your-es-domain.us-east-1.es.amazonaws.com'  # ES域名，不要加https://
INDEX_NAME = 'index_1'

# 初始化Cognito客户端
client = boto3.client('cognito-idp', region_name='us-east-1')  # 替换成你的AWS区域

try:
    # 调用Cognito接口获取认证令牌
    auth_response = client.initiate_auth(
        ClientId=COGNITO_CLIENT_ID,
        AuthFlow='USER_PASSWORD_AUTH',
        AuthParameters={
            'USERNAME': USERNAME,
            'PASSWORD': PASSWORD
        }
    )
    id_token = auth_response['AuthenticationResult']['IdToken']
    print("成功获取Cognito ID Token")
except client.exceptions.NotAuthorizedException:
    print("用户名或密码不对，请检查！")
    sys.exit(1)
except Exception as e:
    print(f"获取Cognito令牌失败：{str(e)}")
    sys.exit(1)

# 用ID Token连接Elasticsearch
es_client = Elasticsearch(
    hosts=[{'host': ES_ENDPOINT, 'port': 443}],
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection,
    headers={'Authorization': f'Bearer {id_token}'}
)

# 检查并创建索引
try:
    index_exists = es_client.indices.exists(INDEX_NAME)
    print(f"索引 {INDEX_NAME} 是否存在：{index_exists}")
    if not index_exists:
        mappings_rds = {
            "settings": {
                "number_of_shards": 2,
                "number_of_replicas": 1
            },
            "mappings": {
                "properties" : {
                    "tableName": { "type": "keyword" },
                    "tableRows": { "type": "integer" },
                    "updatedTime": { "type": "date", "format":"date_optional_time||yyyy-MM-dd'T'HH:mm:ss" },
                    "created_timestamp":{"type": "date", "format":"date_optional_time||yyyy-MM-dd'T'HH:mm:ss"}
                }
            }
        }
        create_result = es_client.indices.create(INDEX_NAME, body=mappings_rds, ignore=400)
        print(f"索引创建结果：{create_result}")
except Exception as e:
    print(f"操作索引失败：{str(e)}")

方案2：用AWS IAM角色/用户进行SigV4签名认证 #

如果你的ES集群同时配置了IAM认证（很多时候Cognito和IAM会配合使用），这个方案更适合服务器端脚本——不需要用户名密码，直接用AWS的访问密钥生成SigV4签名来认证，权限由IAM政策控制。

示例代码如下：

from __future__ import print_function
import boto3
from elasticsearch import Elasticsearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth

# 替换成你的实际配置
REGION = 'us-east-1'
ES_ENDPOINT = 'your-es-domain.us-east-1.es.amazonaws.com'
INDEX_NAME = 'index_1'

# 获取当前AWS环境的凭证（本地环境用~/.aws/credentials，EC2/ECS用实例角色）
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, REGION, 'es', session_token=credentials.token)

# 连接Elasticsearch
es_client = Elasticsearch(
    hosts=[{'host': ES_ENDPOINT, 'port': 443}],
    http_auth=awsauth,
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection
)

# 检查并创建索引
try:
    index_exists = es_client.indices.exists(INDEX_NAME)
    print(f"索引 {INDEX_NAME} 是否存在：{index_exists}")
    if not index_exists:
        mappings_rds = {
            "settings": {
                "number_of_shards": 2,
                "number_of_replicas": 1
            },
            "mappings": {
                "properties" : {
                    "tableName": { "type": "keyword" },
                    "tableRows": { "type": "integer" },
                    "updatedTime": { "type": "date", "format":"date_optional_time||yyyy-MM-dd'T'HH:mm:ss" },
                    "created_timestamp":{"type": "date", "format":"date_optional_time||yyyy-MM-dd'T'HH:mm:ss"}
                }
            }
        }
        create_result = es_client.indices.create(INDEX_NAME, body=mappings_rds, ignore=400)
        print(f"索引创建结果：{create_result}")
except Exception as e:
    print(f"操作索引失败：{str(e)}")

几个重要的注意点 #

• 不管用哪个方案，都要确保你的Cognito用户或者IAM实体有对应的ES访问权限：比如要给用户/角色配置允许es:ESHttpGet、es:ESHttpPut、es:ESHttpPost等动作的政策。
• 安装依赖包：这两个方案都需要额外的包，执行pip install boto3 requests-aws4auth elasticsearch就能搞定。
• 如果是在AWS托管的服务（比如EC2、Lambda）上运行脚本，优先用IAM角色，不要硬编码密钥，更安全。

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