(method='GET', host='', canonical_uri='/', params=None,
region='us-east-1', cur_headers=None, body=b'')
| 200 | |
| 201 | |
| 202 | def sign_request_v4(method='GET', host='', canonical_uri='/', params=None, |
| 203 | region='us-east-1', cur_headers=None, body=b''): |
| 204 | service = 's3' |
| 205 | if cur_headers is None: |
| 206 | cur_headers = SortedDict(ignore_case = True) |
| 207 | |
| 208 | cfg = Config.Config() |
| 209 | access_key = cfg.access_key |
| 210 | secret_key = cfg.secret_key |
| 211 | |
| 212 | t = datetime.datetime.utcnow() |
| 213 | amzdate = t.strftime('%Y%m%dT%H%M%SZ') |
| 214 | datestamp = t.strftime('%Y%m%d') |
| 215 | |
| 216 | signing_key = getSignatureKey(secret_key, datestamp, region, service) |
| 217 | |
| 218 | |
| 219 | canonical_uri = s3_quote(canonical_uri, quote_backslashes=False, unicode_output=True) |
| 220 | canonical_querystring = format_param_str(params, always_have_equal=True).lstrip('?') |
| 221 | |
| 222 | |
| 223 | if type(body) == type(sha256(b'')): |
| 224 | payload_hash = decode_from_s3(body.hexdigest()) |
| 225 | else: |
| 226 | payload_hash = decode_from_s3(sha256(encode_to_s3(body)).hexdigest()) |
| 227 | |
| 228 | canonical_headers = {'host' : host, |
| 229 | 'x-amz-content-sha256': payload_hash, |
| 230 | 'x-amz-date' : amzdate |
| 231 | } |
| 232 | signed_headers = 'host;x-amz-content-sha256;x-amz-date' |
| 233 | |
| 234 | for header in cur_headers.keys(): |
| 235 | # avoid duplicate headers and previous Authorization |
| 236 | if header == 'Authorization' or header in signed_headers.split(';'): |
| 237 | continue |
| 238 | canonical_headers[header.strip()] = cur_headers[header].strip() |
| 239 | signed_headers += ';' + header.strip() |
| 240 | |
| 241 | # sort headers into a string |
| 242 | canonical_headers_str = '' |
| 243 | for k, v in sorted(canonical_headers.items()): |
| 244 | canonical_headers_str += k + ":" + v + "\n" |
| 245 | |
| 246 | canonical_headers = canonical_headers_str |
| 247 | debug(u"canonical_headers = %s" % canonical_headers) |
| 248 | signed_headers = ';'.join(sorted(signed_headers.split(';'))) |
| 249 | |
| 250 | canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash |
| 251 | debug('Canonical Request:\n%s\n----------------------' % canonical_request) |
| 252 | |
| 253 | algorithm = 'AWS4-HMAC-SHA256' |
| 254 | credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request' |
| 255 | string_to_sign = algorithm + '\n' + amzdate + '\n' + credential_scope + '\n' + decode_from_s3(sha256(encode_to_s3(canonical_request)).hexdigest()) |
| 256 | |
| 257 | signature = decode_from_s3(hmac.new(signing_key, encode_to_s3(string_to_sign), sha256).hexdigest()) |
| 258 | authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + credential_scope + ',' + 'SignedHeaders=' + signed_headers + ',' + 'Signature=' + signature |
| 259 | new_headers = SortedDict(cur_headers.items()) |
no test coverage detected
searching dependent graphs…