Python连接Elasticsearch遇SSL证书验证失败问题求助

运行Python脚本导入数据到Docker部署的Elasticsearch时出现以下错误：

elastic_transport.TlsError: TLS error caused by: TlsError(TLS error caused by: SSLError([SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: CA cert does not include key usage extension (_ssl.c:1020)))

该脚本9个月前可正常运行，用于从Telegram API拉取群组数据并导入Elasticsearch。已尝试将容器中的证书复制到系统受信任根证书颁发机构，但无效果，寻求代码层面的解决方案。

脚本代码如下：

groups_file = open('telegram_channels.txt', 'r')
# enter your channel group id -100 after this digit

# Set the time range to get messages from
start_time = datetime.now() - timedelta(hours=24)
flag = 0
groups = groups_file.readlines()


def greet():
    while True:
        days = int(input('Choose the number of days that represent the period within which data will be'
              ' collected (e.g. last 5 days) (Max. 14 days)\n>>>'))
        if days > 0 and days < 15:
            break
    return days


async def get_group_messages(days):
    # Create a Telegram client with the specified API ID, API hash and phone number
    client = TelegramClient('session_name', int(api_id), api_hash)
    await client.connect()

    # Check if the user is already authorized, otherwise prompt the user to authorize the client
    if not await client.is_user_authorized():
        await client.send_code_request(phone_number)
        await client.sign_in(phone_number, input('Enter the code: '))

    data = []

    # Get the ID of the specified group
    for index in range(len(groups)):
        group = await client.get_entity(groups[index])
        date_today = datetime.now()
        lower_bound = date_today - timedelta(days=days)

        # below commented code is used for specified time range
        async for message in client.iter_messages(group, min_id=1):
            if str(message.date) < str(lower_bound):
                break

            message_reactions = message.reactions
            reactions = []
            if message_reactions:
                reaction_counts = message_reactions.results

                for reaction_count in reaction_counts:
                    emoji = reaction_count.reaction.emoticon
                    count = reaction_count.count
                    reactions.append({'emoji': emoji, 'count': count})

            data_object = {"channel": group.title, "date": message.date, "text": message.text,
                           "views": message.views,
                           "reactions": reactions}

            data.append(data_object)

    return data


def index_messages_to_elasticsearch(messages):
    es = Elasticsearch([{'host': es_host, 'port': es_port, 'scheme': es_scheme}],
                       basic_auth=(es_user, es_pass))

    for message in messages:
        es.index(index='telegram', body=message)


async def main():
    days = greet()
    messages = await get_group_messages(days)
    index_messages_to_elasticsearch(messages)

# Run the main function
asyncio.run(main())

代码层面解决方案 #

方案1：临时跳过SSL证书验证（仅用于调试，不推荐生产环境） #

修改index_messages_to_elasticsearch函数中的Elasticsearch客户端初始化代码，添加verify_certs=False参数关闭证书验证：

def index_messages_to_elasticsearch(messages):
    es = Elasticsearch([{'host': es_host, 'port': es_port, 'scheme': es_scheme}],
                       basic_auth=(es_user, es_pass),
                       verify_certs=False)

    for message in messages:
        es.index(index='telegram', body=message)

方案2：指定正确的CA证书路径 #

如果需要保留证书验证，确保使用Elasticsearch容器生成的正确CA证书，将证书路径通过ca_certs参数传入客户端：

def index_messages_to_elasticsearch(messages):
    # 替换为你的CA证书实际路径，比如从Docker容器中复制的certs/http_ca.crt
    ca_cert_path = "/path/to/http_ca.crt"
    es = Elasticsearch([{'host': es_host, 'port': es_port, 'scheme': es_scheme}],
                       basic_auth=(es_user, es_pass),
                       ca_certs=ca_cert_path)

    for message in messages:
        es.index(index='telegram', body=message)

方案3：重新生成符合要求的证书 #

如果上述方案无效，可能是Elasticsearch的CA证书缺少必要的密钥使用扩展。针对Docker部署的ES，可按以下步骤重新生成证书：

1. 停止Elasticsearch容器
2. 删除挂载的原证书目录（通常为certs目录）
3. 重新启动容器，让Elasticsearch自动生成包含完整扩展的新证书；或使用elasticsearch-certutil工具手动生成带有keyUsage和extendedKeyUsage扩展的证书
4. 将新证书复制到脚本可访问路径，再用方案2的代码指定路径

额外注意事项 #

• 确认es_host、es_port、es_scheme配置正确，启用TLS时scheme必须为https
• 检查Python环境是否更新过相关依赖库，新版本的elastic_transport或urllib3可能对证书验证更严格，可尝试降级相关库（如pip install elastic_transport==7.17.0）

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