208 lines
8.3 KiB
Python
208 lines
8.3 KiB
Python
# coding : utf-8
|
|
import datetime
|
|
import sys
|
|
|
|
import pytz
|
|
|
|
try:
|
|
from urllib import urlencode
|
|
except:
|
|
from urllib.parse import urlencode
|
|
|
|
from volcengine.auth.MetaData import MetaData
|
|
from volcengine.util.Util import Util
|
|
from volcengine.base.Request import Request
|
|
from volcengine.auth.SignResult import SignResult
|
|
|
|
class SignerV4(object):
|
|
@staticmethod
|
|
def sign(request, credentials):
|
|
if request.path == '':
|
|
request.path = '/'
|
|
if request.method != 'GET' and not ('Content-Type' in request.headers):
|
|
request.headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=utf-8'
|
|
|
|
format_date = SignerV4.get_current_format_date()
|
|
request.headers['X-Date'] = format_date
|
|
if credentials.session_token != '':
|
|
request.headers['X-Security-Token'] = credentials.session_token
|
|
|
|
md = MetaData()
|
|
md.set_algorithm('HMAC-SHA256')
|
|
md.set_service(credentials.service)
|
|
md.set_region(credentials.region)
|
|
md.set_date(format_date[:8])
|
|
|
|
hashed_canon_req = SignerV4.hashed_canonical_request_v4(request, md)
|
|
md.set_credential_scope('/'.join([md.date, md.region, md.service, 'request']))
|
|
|
|
signing_str = '\n'.join([md.algorithm, format_date, md.credential_scope, hashed_canon_req])
|
|
signing_key = SignerV4.get_signing_secret_key_v4(credentials.sk, md.date, md.region, md.service)
|
|
sign = Util.to_hex(Util.hmac_sha256(signing_key, signing_str))
|
|
request.headers['Authorization'] = SignerV4.build_auth_header_v4(sign, md, credentials)
|
|
return
|
|
|
|
@staticmethod
|
|
def sign_url(request, credentials):
|
|
format_date = SignerV4.get_current_format_date()
|
|
date = format_date[:8]
|
|
|
|
md = MetaData()
|
|
md.set_date(date)
|
|
md.set_service(credentials.service)
|
|
md.set_region(credentials.region)
|
|
md.set_signed_headers('')
|
|
md.set_algorithm('HMAC-SHA256')
|
|
md.set_credential_scope('/'.join([md.date, md.region, md.service, 'request']))
|
|
|
|
query = request.query
|
|
query['X-Date'] = format_date
|
|
query['X-NotSignBody'] = ''
|
|
query['X-Credential'] = credentials.ak + '/' + md.credential_scope
|
|
query['X-Algorithm'] = md.algorithm
|
|
query['X-SignedHeaders'] = md.signed_headers
|
|
query['X-SignedQueries'] = ''
|
|
query['X-SignedQueries'] = ';'.join(sorted(query.keys()))
|
|
if credentials.session_token != '':
|
|
query['X-Security-Token'] = credentials.session_token
|
|
|
|
hashed_canon_req = SignerV4.hashed_simple_canonical_request_v4(request, md)
|
|
signing_str = '\n'.join([md.algorithm, format_date, md.credential_scope, hashed_canon_req])
|
|
signing_key = SignerV4.get_signing_secret_key_v4(credentials.sk, md.date, md.region, md.service)
|
|
sign = SignerV4.signature_v4(signing_key, signing_str)
|
|
|
|
query['X-Signature'] = sign
|
|
return urlencode(query)
|
|
|
|
@staticmethod
|
|
def sign_only(param, credentials):
|
|
request = Request()
|
|
request.host = param.host
|
|
request.method = param.method
|
|
request.path = param.path
|
|
request.body = param.body
|
|
request.query = param.query
|
|
request.headers = param.header_list
|
|
|
|
format_date = param.date.strftime("%Y%m%dT%H%M%SZ")
|
|
date = format_date[:8]
|
|
request.headers['X-Date'] = format_date
|
|
md = MetaData()
|
|
md.set_algorithm('HMAC-SHA256')
|
|
md.set_service(credentials.service)
|
|
md.set_region(credentials.region)
|
|
md.set_date(date)
|
|
md.set_credential_scope('/'.join([md.date, md.region, md.service, 'request']))
|
|
|
|
if param.is_sign_url:
|
|
md.set_signed_headers('')
|
|
md.set_credential_scope('/'.join([md.date, md.region, md.service, 'request']))
|
|
query = request.query
|
|
query['X-Date'] = format_date
|
|
query['X-NotSignBody'] = ''
|
|
query['X-Credential'] = credentials.ak + '/' + md.credential_scope
|
|
query['X-Algorithm'] = md.algorithm
|
|
query['X-SignedHeaders'] = md.signed_headers
|
|
query['X-SignedQueries'] = ''
|
|
query['X-SignedQueries'] = ';'.join(sorted(query.keys()))
|
|
if credentials.session_token != '':
|
|
query['X-Security-Token'] = credentials.session_token
|
|
hashed_canon_req = SignerV4.hashed_simple_canonical_request_v4(request, md)
|
|
else:
|
|
if credentials.session_token != '':
|
|
request.headers['X-Security-Token'] = credentials.session_token
|
|
hashed_canon_req = SignerV4.hashed_canonical_request_v4(request, md)
|
|
|
|
signing_str = '\n'.join([md.algorithm, format_date, md.credential_scope, hashed_canon_req])
|
|
signing_key = SignerV4.get_signing_secret_key_v4(credentials.sk, md.date, md.region, md.service)
|
|
sign = SignerV4.signature_v4(signing_key, signing_str)
|
|
|
|
result = SignResult()
|
|
result.xdate = format_date
|
|
result.xAlgorithm = md.algorithm
|
|
if param.is_sign_url:
|
|
result.xSignedQueries = request.query['X-SignedQueries']
|
|
result.xSignedHeaders = md.signed_headers
|
|
result.xCredential = credentials.ak + '/' + md.credential_scope
|
|
result.xSignature = sign
|
|
result.xContextSha256 = request.headers['X-Content-Sha256']
|
|
result.authorization = result.xAlgorithm + " Credential=" + result.xCredential + ", SignedHeaders=" + md.signed_headers + ", Signature=" + result.xSignature
|
|
result.xSecurityToken = credentials.session_token
|
|
|
|
return result
|
|
|
|
@staticmethod
|
|
def hashed_simple_canonical_request_v4(request, meta):
|
|
body = bytes()
|
|
# if sys.version_info[0] == 3:
|
|
# body_hash = Util.sha256(body.decode('utf-8'))
|
|
# else:
|
|
body_hash = Util.sha256(body)
|
|
|
|
if request.path == '':
|
|
request.path = '/'
|
|
|
|
canoncial_request = '\n'.join(
|
|
[request.method, Util.norm_uri(request.path), Util.norm_query(request.query), '\n',
|
|
meta.signed_headers, body_hash])
|
|
# if sys.version_info[0] == 3:
|
|
# return Util.sha256(canoncial_request.decode('utf-8'))
|
|
# else:
|
|
return Util.sha256(canoncial_request)
|
|
|
|
@staticmethod
|
|
def hashed_canonical_request_v4(request, meta):
|
|
# if sys.version_info[0] == 3:
|
|
# body_hash = Util.sha256(request.body.decode('utf-8'))
|
|
# else:
|
|
body_hash = Util.sha256(request.body)
|
|
request.headers['X-Content-Sha256'] = body_hash
|
|
|
|
signed_headers = dict()
|
|
for key in request.headers:
|
|
if key in ['Content-Type', 'Content-Md5', 'Host'] or key.startswith('X-'):
|
|
signed_headers[key.lower()] = request.headers[key]
|
|
|
|
if 'host' in signed_headers:
|
|
v = signed_headers['host']
|
|
if v.find(':') != -1:
|
|
split = v.split(':')
|
|
port = split[1]
|
|
if str(port) == '80' or str(port) == '443':
|
|
signed_headers['host'] = split[0]
|
|
|
|
signed_str = ''
|
|
for key in sorted(signed_headers.keys()):
|
|
signed_str += key + ':' + signed_headers[key] + '\n'
|
|
|
|
meta.set_signed_headers(';'.join(sorted(signed_headers.keys())))
|
|
|
|
canoncial_request = '\n'.join(
|
|
[request.method, Util.norm_uri(request.path), Util.norm_query(request.query), signed_str,
|
|
meta.signed_headers, body_hash])
|
|
|
|
return Util.sha256(canoncial_request)
|
|
|
|
@staticmethod
|
|
def signature_v4(signing_key, signing_str):
|
|
return Util.to_hex(Util.hmac_sha256(signing_key, signing_str))
|
|
|
|
@staticmethod
|
|
def get_signing_secret_key_v4(sk, date, region, service):
|
|
if sys.version_info[0] == 3:
|
|
kdate = Util.hmac_sha256(bytes(sk, encoding='utf-8'), date)
|
|
else:
|
|
kdate = Util.hmac_sha256(sk.encode('utf-8'), date)
|
|
kregion = Util.hmac_sha256(kdate, region)
|
|
kservice = Util.hmac_sha256(kregion, service)
|
|
return Util.hmac_sha256(kservice, 'request')
|
|
|
|
@staticmethod
|
|
def build_auth_header_v4(signature, meta, credentials):
|
|
credential = credentials.ak + '/' + meta.credential_scope
|
|
return meta.algorithm + ' Credential=' + credential + ', SignedHeaders=' + meta.signed_headers + ', Signature=' + signature
|
|
|
|
@staticmethod
|
|
def get_current_format_date():
|
|
return datetime.datetime.now(tz=pytz.timezone('UTC')).strftime("%Y%m%dT%H%M%SZ")
|