pythonazurerestazure-blob-storageazure-storage-emulator

Connect to Azure Storage Emulator via REST API & Python

I cannot connect to the Azure Storage Emulator using REST & Python.

I can using the Python SDK

I want to try and make some REST calls to better understand the functionality, to compare speeds and look at using in cloud functions to reduce my image size.

I've tried to use the code found here https://stackoverflow.com/a/49881347/9201100, however when using with the emulator i'm constantly returned AuthenticationFailed

I found a recent C# project on github https://github.com/Azure-Samples/storage-dotnet-rest-api-with-auth. If i include the emulator account_name and account_key it runs perfectly fine.

So i modified the code a little bit updating the api_version = '2017-04-17' and the canonicalized resource to match but still no good.

please help

import datetime
import hmac
import hashlib
import base64

storage_account_name = "devstoreaccount1"
storage_account_key = "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
api_version = '2017-04-17'
request_time = datetime.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')

string_params = {
    'verb': 'GET',
    'Content-Encoding': '',
    'Content-Language': '',
    'Content-Length': '',
    'Content-MD5': '',
    'Content-Type': '',
    'Date': '',
    'If-Modified-Since': '',
    'If-Match': '',
    'If-None-Match': '',
    'If-Unmodified-Since': '',
    'Range': '',
    'CanonicalizedHeaders': 'x-ms-date:' + request_time + '\nx-ms-version:' + api_version + '\n',
    'CanonicalizedResource': '/' + storage_account_name + '/'  + storage_account_name + '\ncomp=list'
}

string_to_sign = (string_params['verb'] + '\n'
                  + string_params['Content-Encoding'] + '\n'
                  + string_params['Content-Language'] + '\n'
                  + string_params['Content-Length'] + '\n'
                  + string_params['Content-MD5'] + '\n'
                  + string_params['Content-Type'] + '\n'
                  + string_params['Date'] + '\n'
                  + string_params['If-Modified-Since'] + '\n'
                  + string_params['If-Match'] + '\n'
                  + string_params['If-None-Match'] + '\n'
                  + string_params['If-Unmodified-Since'] + '\n'
                  + string_params['Range'] + '\n'
                  + string_params['CanonicalizedHeaders']
                  + string_params['CanonicalizedResource'])


signed_string = base64.b64encode(hmac.new(base64.b64decode(storage_account_key), msg=string_to_sign.encode('utf-8'), digestmod=hashlib.sha256).digest()).decode()
authHV= {'SharedKey ' + storage_account_name + ':' + signed_string}

headers = {
    'x-ms-date' : request_time,
    'x-ms-version' : api_version,
    'Authorization' : authHV
}


url = ('http://localhost:10000/' + storage_account_name + '?comp=list')
#
r = requests.get(url, headers = headers)

print(r.content)
Solution

  • There're 2 errors in your code:

    1.In 'CanonicalizedResource', you should use '\ncomp:list' instead of '\ncomp=ist'

    2.For variable authHV, you should define it as authHV=('SharedKey ' + storage_account_name + ':' + signed_string)

    And also remember import requests module.

    Please use the code below, it works at my side:

    import requests
import datetime
import hmac
import hashlib
import base64

storage_account_name = 'devstoreaccount1'
storage_account_key = 'Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=='
#api_version = '2016-05-31'
api_version = '2017-04-17'
request_time = datetime.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')

string_params = {
    'verb': 'GET',
    'Content-Encoding': '',
    'Content-Language': '',
    'Content-Length': '',
    'Content-MD5': '',
    'Content-Type': '',
    'Date': '',
    'If-Modified-Since': '',
    'If-Match': '',
    'If-None-Match': '',
    'If-Unmodified-Since': '',
    'Range': '',
    'CanonicalizedHeaders': 'x-ms-date:' + request_time + '\nx-ms-version:' + api_version + '\n',
    'CanonicalizedResource': '/' + storage_account_name +'/'+storage_account_name + '\ncomp:list' #note, it should be '\ncomp:list', no '/'
}

string_to_sign = (string_params['verb'] + '\n' 
                  + string_params['Content-Encoding'] + '\n'
                  + string_params['Content-Language'] + '\n'
                  + string_params['Content-Length'] + '\n'
                  + string_params['Content-MD5'] + '\n' 
                  + string_params['Content-Type'] + '\n' 
                  + string_params['Date'] + '\n' 
                  + string_params['If-Modified-Since'] + '\n'
                  + string_params['If-Match'] + '\n'
                  + string_params['If-None-Match'] + '\n'
                  + string_params['If-Unmodified-Since'] + '\n'
                  + string_params['Range'] + '\n'
                  + string_params['CanonicalizedHeaders']
                  + string_params['CanonicalizedResource'])

signed_string = base64.b64encode(hmac.new(base64.b64decode(storage_account_key), msg=string_to_sign.encode('utf-8'), digestmod=hashlib.sha256).digest()).decode()

headers = {
    'x-ms-date' : request_time,
    'x-ms-version' : api_version,
    'Authorization' : ('SharedKey ' + storage_account_name + ':' + signed_string)
}

url = ('http://127.0.0.1:10000/' + storage_account_name + '?comp=list')

r = requests.get(url, headers = headers)

print(r.status_code)
print(r.content)

    The test result as below:

    enter image description here