# -*- coding: utf-8 -*- # This file is auto-generated, don't edit it. Thanks. from alibabacloud_darabonba_encode_util.encoder import Encoder from Tea.exceptions import TeaException from alibabacloud_darabonba_signature_util.signer import Signer from Tea.core import TeaCore from typing import Any, Dict, List from alibabacloud_gateway_spi.client import Client as SPIClient from alibabacloud_gateway_spi import models as spi_models from alibabacloud_tea_util.client import Client as UtilClient from alibabacloud_openapi_util.client import Client as OpenApiUtilClient from alibabacloud_darabonba_string.client import Client as StringClient from alibabacloud_darabonba_map.client import Client as MapClient from alibabacloud_darabonba_array.client import Client as ArrayClient class Client(SPIClient): def __init__(self): super().__init__() def modify_configuration( self, context: spi_models.InterceptorContext, attribute_map: spi_models.AttributeMap, ) -> None: config = context.configuration config.endpoint = self.get_endpoint(config.network, config.endpoint) async def modify_configuration_async( self, context: spi_models.InterceptorContext, attribute_map: spi_models.AttributeMap, ) -> None: config = context.configuration config.endpoint = await self.get_endpoint_async(config.network, config.endpoint) def modify_request( self, context: spi_models.InterceptorContext, attribute_map: spi_models.AttributeMap, ) -> None: request = context.request config = context.configuration date = UtilClient.get_date_utcstring() request.headers = TeaCore.merge({ 'date': date, 'host': config.endpoint, 'x-acs-version': request.version, 'x-acs-action': request.action, 'user-agent': request.user_agent, 'x-acs-signature-nonce': UtilClient.get_nonce(), 'accept': 'application/json' }, request.headers) signature_algorithm = UtilClient.default_string(request.signature_algorithm, 'ACS4-HMAC-SHA256') signature_version = UtilClient.default_string(request.signature_version, 'v1') hashed_request_payload = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(''), signature_algorithm)) if not UtilClient.is_unset(request.stream): tmp = UtilClient.read_as_bytes(request.stream) hashed_request_payload = Encoder.hex_encode(Encoder.hash(tmp, signature_algorithm)) request.stream = tmp request.headers['content-type'] = 'application/octet-stream' else: if not UtilClient.is_unset(request.body): if UtilClient.equal_string(request.req_body_type, 'json'): json_obj = UtilClient.to_jsonstring(request.body) hashed_request_payload = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(json_obj), signature_algorithm)) request.stream = json_obj request.headers['content-type'] = 'application/json; charset=utf-8' else: m = UtilClient.assert_as_map(request.body) form_obj = OpenApiUtilClient.to_form(m) hashed_request_payload = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(form_obj), signature_algorithm)) request.stream = form_obj request.headers['content-type'] = 'application/x-www-form-urlencoded' if StringClient.equals(signature_version, 'v4'): if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): request.headers['x-acs-content-sm3'] = hashed_request_payload else: request.headers['x-acs-content-sha256'] = hashed_request_payload else: request.headers['x-acs-signature-method'] = 'HMAC-SHA1' request.headers['x-acs-signature-version'] = '1.0' if not UtilClient.equal_string(request.auth_type, 'Anonymous') and not UtilClient.is_unset(request.credential): credential = request.credential credential_model = credential.get_credential() auth_type = credential_model.type if UtilClient.equal_string(auth_type, 'bearer'): bearer_token = credential_model.bearer_token request.headers['x-acs-bearer-token'] = bearer_token request.headers['Authorization'] = f'Bearer {bearer_token}' else: access_key_id = credential_model.access_key_id access_key_secret = credential_model.access_key_secret security_token = credential_model.security_token if not UtilClient.empty(security_token): request.headers['x-acs-security-token'] = security_token headers = {} if not UtilClient.is_unset(request.headers.get('content-type')): headers = request.headers elif StringClient.equals(request.req_body_type, 'formData') and StringClient.equals(request.action, 'DownloadFile') and StringClient.equals(request.pathname, '/v2/file/download'): headers_array = MapClient.key_set(request.headers) for key in headers_array: headers[key] = request.headers.get(key) headers['content-type'] = 'application/x-www-form-urlencoded; charset=UTF-8' else: headers = request.headers if StringClient.equals(signature_version, 'v4'): date_new = StringClient.sub_string(date, 0, 10) region = self.get_region(config.endpoint) signingkey = self.get_signingkey(signature_algorithm, access_key_secret, region, date_new) request.headers['Authorization'] = self.get_authorization_v4(request.pathname, request.method, request.query, headers, signature_algorithm, hashed_request_payload, access_key_id, signingkey, request.product_id, region, date_new) else: request.headers['Authorization'] = self.get_authorization(request.pathname, request.method, request.query, headers, access_key_id, access_key_secret) async def modify_request_async( self, context: spi_models.InterceptorContext, attribute_map: spi_models.AttributeMap, ) -> None: request = context.request config = context.configuration date = UtilClient.get_date_utcstring() request.headers = TeaCore.merge({ 'date': date, 'host': config.endpoint, 'x-acs-version': request.version, 'x-acs-action': request.action, 'user-agent': request.user_agent, 'x-acs-signature-nonce': UtilClient.get_nonce(), 'accept': 'application/json' }, request.headers) signature_algorithm = UtilClient.default_string(request.signature_algorithm, 'ACS4-HMAC-SHA256') signature_version = UtilClient.default_string(request.signature_version, 'v1') hashed_request_payload = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(''), signature_algorithm)) if not UtilClient.is_unset(request.stream): tmp = await UtilClient.read_as_bytes_async(request.stream) hashed_request_payload = Encoder.hex_encode(Encoder.hash(tmp, signature_algorithm)) request.stream = tmp request.headers['content-type'] = 'application/octet-stream' else: if not UtilClient.is_unset(request.body): if UtilClient.equal_string(request.req_body_type, 'json'): json_obj = UtilClient.to_jsonstring(request.body) hashed_request_payload = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(json_obj), signature_algorithm)) request.stream = json_obj request.headers['content-type'] = 'application/json; charset=utf-8' else: m = UtilClient.assert_as_map(request.body) form_obj = OpenApiUtilClient.to_form(m) hashed_request_payload = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(form_obj), signature_algorithm)) request.stream = form_obj request.headers['content-type'] = 'application/x-www-form-urlencoded' if StringClient.equals(signature_version, 'v4'): if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): request.headers['x-acs-content-sm3'] = hashed_request_payload else: request.headers['x-acs-content-sha256'] = hashed_request_payload else: request.headers['x-acs-signature-method'] = 'HMAC-SHA1' request.headers['x-acs-signature-version'] = '1.0' if not UtilClient.equal_string(request.auth_type, 'Anonymous') and not UtilClient.is_unset(request.credential): credential = request.credential credential_model = await credential.get_credential_async() auth_type = credential_model.type if UtilClient.equal_string(auth_type, 'bearer'): bearer_token = credential_model.bearer_token request.headers['x-acs-bearer-token'] = bearer_token request.headers['Authorization'] = f'Bearer {bearer_token}' else: access_key_id = credential_model.access_key_id access_key_secret = credential_model.access_key_secret security_token = credential_model.security_token if not UtilClient.empty(security_token): request.headers['x-acs-security-token'] = security_token headers = {} if not UtilClient.is_unset(request.headers.get('content-type')): headers = request.headers elif StringClient.equals(request.req_body_type, 'formData') and StringClient.equals(request.action, 'DownloadFile') and StringClient.equals(request.pathname, '/v2/file/download'): headers_array = MapClient.key_set(request.headers) for key in headers_array: headers[key] = request.headers.get(key) headers['content-type'] = 'application/x-www-form-urlencoded; charset=UTF-8' else: headers = request.headers if StringClient.equals(signature_version, 'v4'): date_new = StringClient.sub_string(date, 0, 10) region = self.get_region(config.endpoint) signingkey = await self.get_signingkey_async(signature_algorithm, access_key_secret, region, date_new) request.headers['Authorization'] = await self.get_authorization_v4_async(request.pathname, request.method, request.query, headers, signature_algorithm, hashed_request_payload, access_key_id, signingkey, request.product_id, region, date_new) else: request.headers['Authorization'] = await self.get_authorization_async(request.pathname, request.method, request.query, headers, access_key_id, access_key_secret) def modify_response( self, context: spi_models.InterceptorContext, attribute_map: spi_models.AttributeMap, ) -> None: request = context.request response = context.response if UtilClient.is_4xx(response.status_code) or UtilClient.is_5xx(response.status_code): _res = UtilClient.read_as_json(response.body) err = UtilClient.assert_as_map(_res) headers = response.headers request_id = headers.get('x-ca-request-id') err['statusCode'] = response.status_code err['requestId'] = request_id raise TeaException({ 'code': f"{self.default_any(err.get('Code'), err.get('code'))}", 'message': f"{self.default_any(err.get('Message'), err.get('message'))}", 'data': err }) if not UtilClient.is_unset(response.body): if UtilClient.equal_number(response.status_code, 204): UtilClient.read_as_string(response.body) elif UtilClient.equal_string(request.body_type, 'binary'): response.deserialized_body = response.body elif UtilClient.equal_string(request.body_type, 'byte'): byt = UtilClient.read_as_bytes(response.body) response.deserialized_body = byt elif UtilClient.equal_string(request.body_type, 'string'): str = UtilClient.read_as_string(response.body) response.deserialized_body = str elif UtilClient.equal_string(request.body_type, 'json'): response.deserialized_body = UtilClient.read_as_json(response.body) elif UtilClient.equal_string(request.body_type, 'array'): response.deserialized_body = UtilClient.read_as_json(response.body) else: response.deserialized_body = UtilClient.read_as_string(response.body) async def modify_response_async( self, context: spi_models.InterceptorContext, attribute_map: spi_models.AttributeMap, ) -> None: request = context.request response = context.response if UtilClient.is_4xx(response.status_code) or UtilClient.is_5xx(response.status_code): _res = await UtilClient.read_as_json_async(response.body) err = UtilClient.assert_as_map(_res) headers = response.headers request_id = headers.get('x-ca-request-id') err['statusCode'] = response.status_code err['requestId'] = request_id raise TeaException({ 'code': f"{self.default_any(err.get('Code'), err.get('code'))}", 'message': f"{self.default_any(err.get('Message'), err.get('message'))}", 'data': err }) if not UtilClient.is_unset(response.body): if UtilClient.equal_number(response.status_code, 204): await UtilClient.read_as_string_async(response.body) elif UtilClient.equal_string(request.body_type, 'binary'): response.deserialized_body = response.body elif UtilClient.equal_string(request.body_type, 'byte'): byt = await UtilClient.read_as_bytes_async(response.body) response.deserialized_body = byt elif UtilClient.equal_string(request.body_type, 'string'): str = await UtilClient.read_as_string_async(response.body) response.deserialized_body = str elif UtilClient.equal_string(request.body_type, 'json'): response.deserialized_body = await UtilClient.read_as_json_async(response.body) elif UtilClient.equal_string(request.body_type, 'array'): response.deserialized_body = await UtilClient.read_as_json_async(response.body) else: response.deserialized_body = await UtilClient.read_as_string_async(response.body) def get_endpoint( self, network: str, endpoint: str, ) -> str: real_endpoint = 'api.aliyunpds.com' if not UtilClient.empty(endpoint): real_endpoint = endpoint if not UtilClient.empty(network) and StringClient.equals(network, 'vpc'): real_endpoint = StringClient.replace(real_endpoint, 'api.aliyunpds.com', 'api-vpc.aliyunpds.com', None) real_endpoint = StringClient.replace(real_endpoint, 'admin.aliyunpds.com', 'admin-vpc.aliyunpds.com', None) return real_endpoint async def get_endpoint_async( self, network: str, endpoint: str, ) -> str: real_endpoint = 'api.aliyunpds.com' if not UtilClient.empty(endpoint): real_endpoint = endpoint if not UtilClient.empty(network) and StringClient.equals(network, 'vpc'): real_endpoint = StringClient.replace(real_endpoint, 'api.aliyunpds.com', 'api-vpc.aliyunpds.com', None) real_endpoint = StringClient.replace(real_endpoint, 'admin.aliyunpds.com', 'admin-vpc.aliyunpds.com', None) return real_endpoint def default_any( self, input_value: Any, default_value: Any, ) -> Any: if UtilClient.is_unset(input_value): return default_value return input_value def get_authorization( self, pathname: str, method: str, query: Dict[str, str], headers: Dict[str, str], ak: str, secret: str, ) -> str: signature = self.get_signature(pathname, method, query, headers, secret) return f'acs {ak}:{signature}' async def get_authorization_async( self, pathname: str, method: str, query: Dict[str, str], headers: Dict[str, str], ak: str, secret: str, ) -> str: signature = await self.get_signature_async(pathname, method, query, headers, secret) return f'acs {ak}:{signature}' def get_signature( self, pathname: str, method: str, query: Dict[str, str], headers: Dict[str, str], secret: str, ) -> str: string_to_sign = '' canonicalized_resource = self.build_canonicalized_resource(pathname, query) canonicalized_headers = self.build_canonicalized_headers(headers) string_to_sign = f'{method}\n{canonicalized_headers}{canonicalized_resource}' signature = Signer.hmac_sha1sign(string_to_sign, secret) return Encoder.base_64encode_to_string(signature) async def get_signature_async( self, pathname: str, method: str, query: Dict[str, str], headers: Dict[str, str], secret: str, ) -> str: string_to_sign = '' canonicalized_resource = await self.build_canonicalized_resource_async(pathname, query) canonicalized_headers = await self.build_canonicalized_headers_async(headers) string_to_sign = f'{method}\n{canonicalized_headers}{canonicalized_resource}' signature = Signer.hmac_sha1sign(string_to_sign, secret) return Encoder.base_64encode_to_string(signature) def build_canonicalized_resource( self, pathname: str, query: Dict[str, str], ) -> str: canonicalized_resource = pathname if not UtilClient.is_unset(query): query_array = MapClient.key_set(query) sorted_query_array = ArrayClient.asc_sort(query_array) separator = '?' for key in sorted_query_array: canonicalized_resource = f'{canonicalized_resource}{separator}{key}' if not UtilClient.empty(query.get(key)): canonicalized_resource = f'{canonicalized_resource}={query.get(key)}' separator = '&' return canonicalized_resource async def build_canonicalized_resource_async( self, pathname: str, query: Dict[str, str], ) -> str: canonicalized_resource = pathname if not UtilClient.is_unset(query): query_array = MapClient.key_set(query) sorted_query_array = ArrayClient.asc_sort(query_array) separator = '?' for key in sorted_query_array: canonicalized_resource = f'{canonicalized_resource}{separator}{key}' if not UtilClient.empty(query.get(key)): canonicalized_resource = f'{canonicalized_resource}={query.get(key)}' separator = '&' return canonicalized_resource def build_canonicalized_headers( self, headers: Dict[str, str], ) -> str: accept = headers.get('accept') if UtilClient.is_unset(accept): accept = '' content_md_5 = headers.get('content-md5') if UtilClient.is_unset(content_md_5): content_md_5 = '' content_type = headers.get('content-type') if UtilClient.is_unset(content_type): content_type = '' date = headers.get('date') if UtilClient.is_unset(date): date = '' canonicalized_headers = f'{accept}\n{content_md_5}\n{content_type}\n{date}\n' sorted_headers = self.get_signed_headers(headers) for header in sorted_headers: value = headers.get(header) value_trim = StringClient.trim(value) canonicalized_headers = f'{canonicalized_headers}{header}:{value_trim}\n' return canonicalized_headers async def build_canonicalized_headers_async( self, headers: Dict[str, str], ) -> str: accept = headers.get('accept') if UtilClient.is_unset(accept): accept = '' content_md_5 = headers.get('content-md5') if UtilClient.is_unset(content_md_5): content_md_5 = '' content_type = headers.get('content-type') if UtilClient.is_unset(content_type): content_type = '' date = headers.get('date') if UtilClient.is_unset(date): date = '' canonicalized_headers = f'{accept}\n{content_md_5}\n{content_type}\n{date}\n' sorted_headers = await self.get_signed_headers_async(headers) for header in sorted_headers: value = headers.get(header) value_trim = StringClient.trim(value) canonicalized_headers = f'{canonicalized_headers}{header}:{value_trim}\n' return canonicalized_headers def get_signed_headers( self, headers: Dict[str, str], ) -> List[str]: headers_array = MapClient.key_set(headers) sorted_headers_array = ArrayClient.asc_sort(headers_array) tmp = '' separator = '' for key in sorted_headers_array: lower_key = StringClient.to_lower(key) if StringClient.has_prefix(lower_key, 'x-acs-'): if not StringClient.contains(tmp, lower_key): tmp = f'{tmp}{separator}{lower_key}' separator = ';' return StringClient.split(tmp, ';', None) async def get_signed_headers_async( self, headers: Dict[str, str], ) -> List[str]: headers_array = MapClient.key_set(headers) sorted_headers_array = ArrayClient.asc_sort(headers_array) tmp = '' separator = '' for key in sorted_headers_array: lower_key = StringClient.to_lower(key) if StringClient.has_prefix(lower_key, 'x-acs-'): if not StringClient.contains(tmp, lower_key): tmp = f'{tmp}{separator}{lower_key}' separator = ';' return StringClient.split(tmp, ';', None) def get_region( self, endpoint: str, ) -> str: region = 'center' if UtilClient.empty(endpoint): return region if StringClient.contains(endpoint, '.admin.aliyunpds.com'): region = StringClient.replace(endpoint, '.admin.aliyunpds.com', '', None) return region def get_signingkey( self, signature_algorithm: str, secret: str, region: str, date: str, ) -> bytes: sc_1 = f'aliyun_v4{secret}' sc_2 = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): sc_2 = Signer.hmac_sha256sign(date, sc_1) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): sc_2 = Signer.hmac_sm3sign(date, sc_1) sc_3 = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): sc_3 = Signer.hmac_sha256sign_by_bytes(region, sc_2) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): sc_3 = Signer.hmac_sm3sign_by_bytes(region, sc_2) sc_4 = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): sc_4 = Signer.hmac_sha256sign_by_bytes('pds', sc_3) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): sc_4 = Signer.hmac_sm3sign_by_bytes('pds', sc_3) hmac = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): hmac = Signer.hmac_sha256sign_by_bytes('aliyun_v4_request', sc_4) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): hmac = Signer.hmac_sm3sign_by_bytes('aliyun_v4_request', sc_4) return hmac async def get_signingkey_async( self, signature_algorithm: str, secret: str, region: str, date: str, ) -> bytes: sc_1 = f'aliyun_v4{secret}' sc_2 = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): sc_2 = Signer.hmac_sha256sign(date, sc_1) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): sc_2 = Signer.hmac_sm3sign(date, sc_1) sc_3 = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): sc_3 = Signer.hmac_sha256sign_by_bytes(region, sc_2) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): sc_3 = Signer.hmac_sm3sign_by_bytes(region, sc_2) sc_4 = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): sc_4 = Signer.hmac_sha256sign_by_bytes('pds', sc_3) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): sc_4 = Signer.hmac_sm3sign_by_bytes('pds', sc_3) hmac = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): hmac = Signer.hmac_sha256sign_by_bytes('aliyun_v4_request', sc_4) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): hmac = Signer.hmac_sm3sign_by_bytes('aliyun_v4_request', sc_4) return hmac def get_authorization_v4( self, pathname: str, method: str, query: Dict[str, str], headers: Dict[str, str], signature_algorithm: str, payload: str, ak: str, signingkey: bytes, product: str, region: str, date: str, ) -> str: signature = self.get_signature_v4(pathname, method, query, headers, signature_algorithm, payload, signingkey) signed_headers = self.get_signed_headers(headers) signed_headers_str = ArrayClient.join(signed_headers, ';') return f'{signature_algorithm} Credential={ak}/{date}/{region}/{product}/aliyun_v4_request,SignedHeaders={signed_headers_str},Signature={signature}' async def get_authorization_v4_async( self, pathname: str, method: str, query: Dict[str, str], headers: Dict[str, str], signature_algorithm: str, payload: str, ak: str, signingkey: bytes, product: str, region: str, date: str, ) -> str: signature = await self.get_signature_v4_async(pathname, method, query, headers, signature_algorithm, payload, signingkey) signed_headers = await self.get_signed_headers_async(headers) signed_headers_str = ArrayClient.join(signed_headers, ';') return f'{signature_algorithm} Credential={ak}/{date}/{region}/{product}/aliyun_v4_request,SignedHeaders={signed_headers_str},Signature={signature}' def get_signature_v4( self, pathname: str, method: str, query: Dict[str, str], headers: Dict[str, str], signature_algorithm: str, payload: str, signingkey: bytes, ) -> str: string_to_sign = '' canonicalized_resource = self.build_canonicalized_resource(pathname, query) canonicalized_headers = self.build_canonicalized_headers(headers) signed_headers = self.get_signed_headers(headers) signed_headers_str = ArrayClient.join(signed_headers, ';') string_to_sign = f'{method}\n{canonicalized_resource}\n{canonicalized_headers}\n{signed_headers_str}\n{payload}' hex = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(string_to_sign), signature_algorithm)) string_to_sign = f'{signature_algorithm}\n{hex}' signature = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): signature = Signer.hmac_sha256sign_by_bytes(string_to_sign, signingkey) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): signature = Signer.hmac_sm3sign_by_bytes(string_to_sign, signingkey) return Encoder.hex_encode(signature) async def get_signature_v4_async( self, pathname: str, method: str, query: Dict[str, str], headers: Dict[str, str], signature_algorithm: str, payload: str, signingkey: bytes, ) -> str: string_to_sign = '' canonicalized_resource = await self.build_canonicalized_resource_async(pathname, query) canonicalized_headers = await self.build_canonicalized_headers_async(headers) signed_headers = await self.get_signed_headers_async(headers) signed_headers_str = ArrayClient.join(signed_headers, ';') string_to_sign = f'{method}\n{canonicalized_resource}\n{canonicalized_headers}\n{signed_headers_str}\n{payload}' hex = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(string_to_sign), signature_algorithm)) string_to_sign = f'{signature_algorithm}\n{hex}' signature = UtilClient.to_bytes('') if UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SHA256'): signature = Signer.hmac_sha256sign_by_bytes(string_to_sign, signingkey) elif UtilClient.equal_string(signature_algorithm, 'ACS4-HMAC-SM3'): signature = Signer.hmac_sm3sign_by_bytes(string_to_sign, signingkey) return Encoder.hex_encode(signature)