如何为 Azure Cosmos DB REST API 构造散列令牌签名以列出用户?

How to construct the hashed token signature for Azure Cosmos DB REST API to list users?

根据 Cosmos DB REST API 的文档,每次 API 调用都必须设置授权 header。该值的构造如下所述:https://docs.microsoft.com/en-us/rest/api/cosmos-db/access-control-on-cosmosdb-resources

我在 Python 中实现如下:

def get_authorisation_token(verb, resource_type, resource_id, date, master_key):
     key = base64.b64decode(master_key)

     text = f"""{verb.lower()}\n{resource_type.lower()}\n{resource_id.lower()}\n{date.lower()}\n\n"""

     text_encoded = text.encode('utf-8')

     signature_hash = hmac.new(key, text_encoded, digestmod=hashlib.sha256).digest()
     signature = base64.b64encode(signature_hash).decode()

     key_type = 'master'
     version = '1.0'

     uri = f'type={key_type}&ver={version}&sig={signature}'
     uri_encoded = urllib.parse.quote(uri)

     return uri_encoded

由于这是随每次调用一起发送的,因此身份验证令牌需要 re-created 才能匹配请求 URL。因此,例如要获取数据库列表,必须提供资源类型 dbs 和资源 link/ID 为空字符串 URL:https://{databaseaccount}.documents.azure.com/dbs/

我无法弄清楚的部分是资源类型和资源的正确组合 ID/link 以从特定数据库获取所有用户。可以在此处找到文档:https://docs.microsoft.com/en-us/rest/api/cosmos-db/list-users

我已经尝试了一些组合,但没有 returns 用户,我只是得到一个 401:

{
    "code": "Unauthorized",
    "message": "The input authorization token can't serve the request. Please check that the expected payload is built as per the protocol, and check the key being used. Server used the following payload to sign: 'get\nusers\ndbs/<db_name>\nmon, 09 nov 2020 23:37:24 gmt\n\n'\r\nActivityId: 697a4159-f160-4aab-ae90-6cb5eaadb710, Microsoft.Azure.Documents.Common/2.11.0"
}

问题请参考以下代码

from wsgiref.handlers import format_date_time
from datetime import datetime
from time import mktime
import base64
from urllib.parse import quote
import hmac
from hashlib import sha256
import requests
from azure.cosmos.auth import GetAuthorizationHeader
from azure.cosmos.cosmos_client import CosmosClientConnection

master_key = ''
database_name = ''
key = base64.b64decode(master_key)
verb = 'GET'
resource_type = 'users'
resource_id = f'dbs/{database_name}'
now = datetime.now()
stamp = mktime(now.timetuple())
date = format_date_time(stamp)
print(date)
text = "{verb}\n{resource_type}\n{resource_id}\n{date}\n{other}\n".format(
    verb=(verb.lower() or ''),
    resource_type=(resource_type.lower() or ""),
    resource_id=(resource_id or ""),
    date=date.lower(),
    other="".lower())

body = text.encode("utf-8")
digest = hmac.new(key, body, sha256).digest()
signature = base64.encodebytes(digest).decode("utf-8")
key_type = 'master'
version = '1.0'
uri = f'type={key_type}&ver={version}&sig={signature[:-1]}'
uri_encoded = quote(uri)

url = "https://<>.documents.azure.com:443/dbs/<>/users"

payload = {}
headers = {
    'Authorization': uri_encoded,
    'x-ms-date': date,
    'x-ms-version': '2018-12-31'
}

response = requests.request("GET", url, headers=headers, data=payload)

print(response.text)