AWS S3 - 发送签名请求失败

问题描述 投票：0回答：1

澄清。

使用头的方法对常规（也就是永久）凭证和临时凭证都有效.使用查询字符串（AKS将所有头作为uri的一部分），对永久凭证有效，但对临时凭证则失败，出现以下错误。

The request signature we calculated does not match the signature you provided. Check your key and signing method.

这与任何时钟差异无关，而且我知道临时凭证在这个S3 bucket上可以使用，这是事实。

临时凭证的格式如下

[default]
aws_access_key_id = A********
aws_secret_access_key = U*******
aws_session_token = F******

澄清结束

我正在写一个代码片段，需要处理从一个私有的S3 bucket的get请求.我有一个工作代码，使用永久和临时凭证的头。

我现在的目标是使用查询字符串来实现同样的目标，但我无法让它为临时凭证工作，它的工作就好了，永久的，所以我很肯定问题涉及到canonical_querystring的某个地方，但我已经尝试了似乎没有工作

我不能使用boto3，因为它需要独立于任何外部包（requests包仅用于调试，它不会成为最终代码的一部分），如果有人能告诉我我做错了什么，那将是非常感激的。

她的是我使用查询字符串的尝试

import datetime
import hashlib
import hmac
import re

try:
    import httplib
except ImportError:
    import http.client as httplib
import requests
import urllib


def get_region(url, host):
    conn = httplib.HTTPConnection(url)
    headers = {'Host': host}
    conn.request('HEAD', '/', headers=headers)
    res = conn.getresponse()
    status = res.status
    if 400 <= status:
        return None
    return res.getheader('x-amz-bucket-region')


def _sign(key, msg):
    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()


def _get_signature_key(key, date_stamp, region_name, service_name):
    k_date = _sign(('AWS4' + key).encode('utf-8'), date_stamp)
    k_region = _sign(k_date, region_name)
    k_service = _sign(k_region, service_name)
    k_signing = _sign(k_service, 'aws4_request')
    return k_signing


def get_sign_headers(access_key, secret_key, url, session_token=None, method='GET', request_parameters=''):
    regex = r"^(https://)(([a-zA-z0-9\-]+)\.)((\w+.*)\.amazonaws\.com)([^:^/]*)?(.*)$"
    matches = re.match(regex, url, re.MULTILINE)
    service = 's3'
    host = matches.group(2) + matches.group(4)
    canonical_uri = matches.group(7)
    region = get_region('s3.us-east-2.amazonaws.com', host)
    if matches.group(3) == 's3':
        if not region:
            region = matches.group(5)
        host = service + '.' + region + '.amazonaws.com'
    endpoint = 'https://' + host + matches.group(7)

    t = datetime.datetime.utcnow()
    amz_date = t.strftime('%Y%m%dT%H%M%SZ')  # Format date as YYYYMMDD'T'HHMMSS'Z'
    datestamp = t.strftime('%Y%m%d')  # Date w/o time, used in credential scope
    canonical_headers = 'host:' + host + '\n'
    signed_headers = 'host'
    if session_token:
        canonical_headers += 'x-amz-security-token:' + session_token + '\n'
        signed_headers += ';x-amz-security-token'
    algorithm = 'AWS4-HMAC-SHA256'
    credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request'
    canonical_querystring = 'X-Amz-Algorithm=AWS4-HMAC-SHA256'
    try:
        credential = urllib.quote_plus(access_key + '/' + credential_scope)
    except AttributeError:
        credential = urllib.parse.quote_plus(access_key + '/' + credential_scope)
    canonical_querystring += '&X-Amz-Credential=' + credential
    canonical_querystring += '&X-Amz-Date=' + amz_date
    canonical_querystring += '&X-Amz-Expires=30'
    canonical_querystring += '&X-Amz-SignedHeaders=' + signed_headers
    payload_hash = 'UNSIGNED-PAYLOAD'
    canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + \
                                 '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash
    string_to_sign = algorithm + '\n' + amz_date + '\n' + credential_scope + '\n' + hashlib.sha256(
        canonical_request.encode('utf-8')).hexdigest()
    signing_key = _get_signature_key(secret_key, datestamp, region, service)
    signature = hmac.new(signing_key, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
    if session_token:
        try:
            session_token = urllib.quote_plus(session_token)
        except AttributeError:
            session_token = urllib.parse.quote_plus(session_token)
        canonical_querystring += '&X-Amz-Security-Token={TOKEN}'.format(TOKEN=session_token)
    canonical_querystring += '&X-Amz-Signature=' + signature
    request_url = endpoint + "?" + canonical_querystring

    print('\nBEGIN REQUEST++++++++++++++++++++++++++++++++++++')
    print('Request URL = ' + request_url)
    r = requests.get(request_url)

    print('\nRESPONSE++++++++++++++++++++++++++++++++++++')
    print('Response code: %d\n' % r.status_code)
    print(r.text)

而这是使用headers方法的代码（这两种方法都适用）。

import datetime
import hashlib
import hmac
import re
import requests
try:
    import httplib
except ImportError:
    import http.client as httplib


def get_region(url, host):
    conn = httplib.HTTPConnection(url)
    headers = {'Host': host}
    conn.request('HEAD', '/', headers=headers)
    res = conn.getresponse()
    status = res.status
    if 400 <= status:
        return None
    return res.getheader('x-amz-bucket-region')


def _sign(key, msg):
    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()


def _get_signature_key(key, date_stamp, region_name, service_name):
    k_date = _sign(('AWS4' + key).encode('utf-8'), date_stamp)
    k_region = _sign(k_date, region_name)
    k_service = _sign(k_region, service_name)
    k_signing = _sign(k_service, 'aws4_request')
    return k_signing


def get_sign_headers(access_key, secret_key, url, session_token=None, method='GET', request_parameters=''):
    regex = r"^(https://)(([a-zA-z0-9\-]+)\.)((\w+.*)\.amazonaws\.com)([^:^/]*)?(.*)$"
    matches = re.match(regex, url, re.MULTILINE)
    service = 's3'
    host = matches.group(2) + matches.group(4)
    canonical_uri = matches.group(7)
    region = get_region('s3.us-east-2.amazonaws.com', host)
    if matches.group(3) == 's3':
        if not region:
            region = matches.group(5)
        host = service + '.' + region + '.amazonaws.com'
    endpoint = 'https://' + host
    t = datetime.datetime.utcnow()
    amz_date = t.strftime('%Y%m%dT%H%M%SZ')
    datestamp = t.strftime('%Y%m%d')  # Date w/o time, used in credential scope

    canonical_querystring = request_parameters
    canonical_headers = 'host:' + host + '\n' + 'x-amz-content-sha256:UNSIGNED-PAYLOAD' + \
                        '\n' + 'x-amz-date:' + amz_date + '\n'
    signed_headers = 'host;x-amz-content-sha256;x-amz-date'
    if session_token:
        signed_headers += ';x-amz-security-token'
        canonical_headers += 'x-amz-security-token:' + session_token + '\n'
    payload_hash = 'UNSIGNED-PAYLOAD'
    canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + \
                                 '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash

    algorithm = 'AWS4-HMAC-SHA256'
    credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request'
    string_to_sign = algorithm + '\n' + amz_date + '\n' + credential_scope + '\n' + hashlib.sha256(
        canonical_request.encode('utf-8')).hexdigest()
    signing_key = _get_signature_key(secret_key, datestamp, region, service)
    signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
    authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + credential_scope + ', ' + \
                                       'SignedHeaders=' + signed_headers + ', ' + 'Signature=' + signature

    headers = {'x-amz-date': amz_date, 'x-amz-content-sha256': 'UNSIGNED-PAYLOAD',
               'Authorization': authorization_header}
    if session_token:
        headers['x-amz-security-token'] = session_token
    request_url = endpoint + canonical_uri
​
    print('\nBEGIN REQUEST++++++++++++++++++++++++++++++++++++')
    print('Request URL = ' + request_url)
    print('Headers = {}'.format(headers))
    r = requests.get(request_url, headers=headers)
​
    print('\nRESPONSE++++++++++++++++++++++++++++++++++++')
    print('Response code: %d\n' % r.status_code)
python amazon-web-services amazon-s3
1个回答
0
投票

你有没有验证过你的服务器时间是否准确。即使是几分钟的差异也会导致签名失败。

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.