Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
raise exceptions.ClientException(error_code.SDK_INVALID_CREDENTIAL,
error_msg.get_msg('SDK_INVALID_CREDENTIAL'))
class SignerV1(Signer):
def __init__(self, access_key, access_secret):
self._access_key = access_key
self._access_secret = access_secret
def sign(self, region_id, request):
header = request.get_signed_header(region_id, self._access_key, self._access_secret)
url = request.get_url(region_id, self._access_key, self._access_secret)
return header, url
class SignerV2(Signer):
_MIN_SESSION_PERIOD = 900
_MAX_SESSION_PERIOD = 3600
_RETRY_DELAY_FAST = 3
_PRIORITY = 1
def __init__(self, public_key_id, private_key, session_period, region_id, debug=False):
if not debug and session_period < self._MIN_SESSION_PERIOD or session_period > self._MAX_SESSION_PERIOD:
raise exceptions.ClientException(
error_code.SDK_INVALID_SESSION_EXPIRATION,
error_msg.get_msg('SDK_INVALID_SESSION_EXPIRATION').format(self._MIN_SESSION_PERIOD,
self._MAX_SESSION_PERIOD))
self._public_key_id = public_key_id
self._private_key = private_key
self._session_period = session_period
self._schedule_interval = session_period if debug else max(session_period * 0.8, 5)
def sign(self, region_id, request):
pass
@staticmethod
def get_signer(credential, debug=False):
if credential['ak'] is not None and credential['secret'] is not None:
return SignerV1(credential['ak'], credential['secret'])
elif credential['public_key_id'] is not None and credential['private_key'] is not None:
return SignerV2(credential['public_key_id'], credential['private_key'], credential['session_period'],
credential['region_id'], debug)
else:
raise exceptions.ClientException(error_code.SDK_INVALID_CREDENTIAL,
error_msg.get_msg('SDK_INVALID_CREDENTIAL'))
class SignerV1(Signer):
def __init__(self, access_key, access_secret):
self._access_key = access_key
self._access_secret = access_secret
def sign(self, region_id, request):
header = request.get_signed_header(region_id, self._access_key, self._access_secret)
url = request.get_url(region_id, self._access_key, self._access_secret)
return header, url
class SignerV2(Signer):
_MIN_SESSION_PERIOD = 900
_MAX_SESSION_PERIOD = 3600
_RETRY_DELAY_FAST = 3
_PRIORITY = 1