如何使用elasticsearch-py客户端在AWS环境中创建Elasticsearch用户与角色

方案1：直接使用elasticsearch-py客户端内置的原始请求方法 #

你初始化的elasticsearch-py客户端本身就支持发送任意自定义路径的原始请求，不需要调用封装好的索引相关接口，直接使用perform_request方法即可，示例代码如下：

# 你的角色定义载荷
role_payload = {
    "cluster_permissions" : [
      "indices_monitor",
    ],
    "index_permissions" : [
      {
        "index_patterns" : [
          "*"
        ],
        "dls" : "",
        "fls" : [ ],
        "masked_fields" : [ ],
        "allowed_actions" : [
          "read",
          "indices:monitor/stats"
        ]
      }
    ],
    "tenant_permissions" : [
      {
        "tenant_patterns" : [
          "human_resources"
        ],
        "allowed_actions" : [
          "kibana_all_read"
        ]
      }
    ]
}

# 调用perform_request发送PUT请求
response = elasticsearch.perform_request(
    "PUT",
    "/_opendistro/_security/api/roles/jesperancinha-role",
    body=role_payload
)
# 打印返回结果确认创建成功
print(response)

注意：如果你使用的是较新版本的AWS OpenSearch服务，安全API路径可能已变更为/_plugins/_security/api/roles/jesperancinha-role，如果原路径返回404可以替换路径重试。

方案2：直接使用requests库配合现有AWS4Auth鉴权发送请求 #

如果你不想依赖elasticsearch-py的封装，也可以直接用你已经构造好的awsauth鉴权实例，通过requests直接发送HTTP请求，代码更直观：

import requests

role_payload = {
    # 同上的角色定义
}

url = f"https://{self._host}/_opendistro/_security/api/roles/jesperancinha-role"
response = requests.put(
    url,
    auth=awsauth,
    json=role_payload,
    verify=True
)
# 检查请求状态
print(response.status_code)
print(response.json())

两种方案都可以满足你的需求，不需要额外调整你已经完成的AWS鉴权逻辑。

内容的提问来源于stack exchange，提问作者João Esperancinha