Copyright (year) Beijing Volcano Engine Technology Ltd.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
from urllib.parse import quote
# The following parameters vary depending on the service, but are usually consistent within a service.
# Service, Version, Region, Host, ContentType must be consistent with the API documentation.
Host = "rtc.volcengineapi.com"
ContentType = "application/json"
# Request credentials, obtained from IAM or STS service.
# Please replace with your own real AK and SK.
# When using temporary credentials, you need to pass the SessionToken into the Header and include it in the SignedHeader. Please add the X-Security-Token header to the header parameter yourself.
for key in sorted(params.keys()):
if type(params[key]) == list:
query + quote(key, safe="-_. ~") + "=" + quote(k, safe="-_. ~") + "&"
query = (query + quote(key, safe="-_. ~") + "=" + quote(params[key], safe="-_. ~") + "&")
return query.replace("+", "%20")
# Step 1: Prepare helper functions.
def hmac_sha256(key: bytes, content: str):
return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest()
def hash_sha256(content: str):
return hashlib.sha256(content.encode("utf-8")).hexdigest()
# Step 2: Sign request function
def request(method, date, query, header, ak, sk, action, body):
# Step 3: Create credential. The Service and Region fields are fixed. ak and sk represent
# AccessKeyID and SecretAccessKey, respectively. At the same time, the signature structure needs to be initialized. Some properties required for signature calculation are also processed here.
# Initialize credential structure
"secret_access_key": sk,
# Initialize signature structure
"content_type": ContentType,
"query": {"Action": action, "Version": Version, **query},
request_param["body"] = ""
# Step 4: Next, start calculating the signature. Before calculating the signature, prepare the signResult variable to receive the signature calculation result and set some parameters.
# Initialize the structure of the signature result
x_date = request_param["date"].strftime("%Y%m%dT%H%M%SZ")
short_x_date = x_date[:8]
x_content_sha256 = hash_sha256(request_param["body"])
"Host": request_param["host"],
"X-Content-Sha256": x_content_sha256,
"Content-Type": request_param["content_type"],
# Step 5: Calculate the Signature.
signed_headers_str = ";".join(
["content-type", "host", "x-content-sha256", "x-date"]
# signed_headers_str = signed_headers_str + ";x-security-token"
canonical_request_str = "\n".join(
[request_param["method"].upper(),
norm_query(request_param["query"]),
"content-type:" + request_param["content_type"],
"host:" + request_param["host"],
"x-content-sha256:" + x_content_sha256,
# Print the canonical request for debugging comparison
print("CanonicalRequest:\n" + canonical_request_str)
hashed_canonical_request = hash_sha256(canonical_request_str)
# Print the hash value for debugging comparison
print("\nHashedCanonicalRequest:\n" + hashed_canonical_request)
credential_scope = "/".join([short_x_date, credential["region"], credential["service"], "request"])
string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hashed_canonical_request])
# Print the final calculated signature string for debugging comparison
print("\nStringToSign:\n" + string_to_sign)
k_date = hmac_sha256(credential["secret_access_key"].encode("utf-8"), short_x_date)
k_region = hmac_sha256(k_date, credential["region"])
k_service = hmac_sha256(k_region, credential["service"])
k_signing = hmac_sha256(k_service, "request")
signature = hmac_sha256(k_signing, string_to_sign).hex()
sign_result["Authorization"] = "HMAC-SHA256 Credential={}, SignedHeaders={}, Signature={}".format(
credential["access_key_id"] + "/" + credential_scope,
header = {**header, **sign_result}
# header = {**header, **{"X-Security-Token": SessionToken}}
# Step 6: Write the Signature into the HTTP Header and send the HTTP request.
r = requests.request(method=method,
url="https://{}{}".format(request_param["host"], request_param["path"]),
params=request_param["query"],
data=request_param["body"],
# datetime.utcnow() is deprecated in 3.12+, use the following method for compatibility
from datetime import timezone
return datetime.datetime.now(timezone.utc)
class UTC(datetime.tzinfo):
def utcoffset(self, dt):
return datetime.timedelta(0)
return datetime.timedelta(0)
return datetime.datetime.now(UTC())
if __name__ == "__main__":
# Please replace AppId, RoomId, TaskId with your actual situation
response_body = request("POST", now, {}, {}, AK, SK, "StopVoiceChat", json.dumps(body_payload))