pythonazureazure-cosmosdbunauthorized

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


According to the documentation for the Cosmos DB REST API, with every API call, the Authorization header must be set. The value for this is constructed as described here: https://learn.microsoft.com/en-us/rest/api/cosmos-db/access-control-on-cosmosdb-resources

I am implementing this in Python as follows:

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

Since this is sent with every call, the auth token needs to be re-created to match the request URL. So, for example to get a list of databases, one must provide the resource type to be dbs and the resource link/ID to be an empty string with the URL: https://{databaseaccount}.documents.azure.com/dbs/

The part I cannot figure out, is the correct combination of resource type and resource ID/link to get all users from a particular database. Documentation can be found here: https://learn.microsoft.com/en-us/rest/api/cosmos-db/list-users

I have tried some combinations but nothing returns the users, I just get a 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"
}

Solution

  • Regarding the issue, please refer to the following code

    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)
    
    

    enter image description here