Azure存储模拟器Python连接表服务时403认证失败求助

问题场景 #

尝试通过Python手动签名请求连接Azure存储模拟器读取表数据，代码执行后返回403认证失败错误。

原代码 #

import datetime
import base64
import hmac
import hashlib
import requests

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

string_params = {
    'verb': 'GET',
    'Content-MD5': '',
    'Content-Type': '',
    'Date': request_time,
    '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-MD5'] + "\n" +
               string_params['Content-Type'] + "\n" +  
               string_params['Date'] + "\n" +  
               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 = {
    'Date' : request_time,
    'x-ms-version' : api_version,
    'Authorization' : ('SharedKey ' + storage_account_name + ':' + signed_string)
}

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

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

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

错误信息 #

403
b'<?xml version="1.0" encoding="utf-8"?><m:error xmlns:m="http://schemas.microsoft.com/ado/2007/08/dataservices/metadata"><m:code>AuthenticationFailed</m:code><m:message xml:lang="en-US">Server failed to authenticate the request. Make sure the value of Authorization header is formed correctly including the signature.\nRequestId:fdcea5c3-e071-4745-87dd-620848f550de\nTime:2023-03-03T05:07:01.4244873Z</m:message></m:error>'

问题原因及修正 #

代码存在两个核心问题：

• CanonicalizedResource格式错误：表服务的list tables操作，CanonicalizedResource的正确格式应为/{存储账户名}\ncomp:list，你错误地重复添加了存储账户名，导致签名不匹配。
• 请求时间不是UTC时间：datetime.datetime.now()获取的是本地时间，Azure存储服务要求签名中的时间必须是UTC时间，时间不一致会触发认证失败。

修正后的代码 #

import datetime
import base64
import hmac
import hashlib
import requests

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

string_params = {
    'verb': 'GET',
    'Content-MD5': '',
    'Content-Type': '',
    'Date': request_time,
    # 修正CanonicalizedResource格式
    'CanonicalizedResource': '/' + storage_account_name + '\ncomp:list'
}

string_to_sign = (string_params['verb'] + "\n" +
               string_params['Content-MD5'] + "\n" +
               string_params['Content-Type'] + "\n" +  
               string_params['Date'] + "\n" +  
               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 = {
    'Date' : request_time,
    'x-ms-version' : api_version,
    'Authorization' : ('SharedKey ' + storage_account_name + ':' + signed_string)
}

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

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

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

额外建议 #

手动实现签名容易出错，推荐使用官方Azure Storage SDK（如azure-data-tables）来操作存储模拟器，SDK会自动处理签名和请求细节，示例代码如下：

from azure.data.tables import TableServiceClient

connection_string = "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;TableEndpoint=http://127.0.0.1:10002/devstoreaccount1;"

table_service = TableServiceClient.from_connection_string(connection_string)
tables = list(table_service.list_tables())
for table in tables:
    print(table.name)

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