fix: correct V4 signing - URI encode query params, fix canonical headers format

reallife_asset/rl_volcengine_client.py

@@ -33,20 +33,19 @@ def _get_signature_key(secret_key, date_stamp, region, service):
     return k_signing
 
 
+def _uri_encode(s, encode_slash=True):
+    """URI encode string per V4 spec."""
+    import urllib.parse
+    if encode_slash:
+        return urllib.parse.quote(s, safe='')
+    else:
+        return urllib.parse.quote(s, safe='/')
+
+
 async def call_volcengine_api(vendor, operation, params, api_mapping):
     """
     Call Volcengine Ark API with V4 signing.
-    
-    Args:
-        vendor: vendor identifier (e.g., "volcengine")
-        operation: internal operation name (e.g., "create_session")
-        params: API parameters dict
-        api_mapping: dict mapping operation to API action name
-    
-    Returns:
-        dict: API response or error
     """
-    # Get action name from mapping
     action = api_mapping.get(operation)
     if not action:
         return {"error": f"未配置操作: {operation}"}
@@ -61,43 +60,51 @@ async def call_volcengine_api(vendor, operation, params, api_mapping):
             return {"error": f"供应商配置不存在: {vendor}"}
         rec = recs[0]
         ak = getattr(rec, "ak", "")
-        sk_encrypted = getattr(rec, "sk", "")
+        sk = getattr(rec, "sk", "")
         
-        if not ak or not sk_encrypted:
+        if not ak or not sk:
             return {"error": "AK/SK未配置"}
     
-        # AK/SK are stored in plaintext
+    # Timestamps
-        sk = sk_encrypted
-    
-    # Build signed request
     now = datetime.datetime.utcnow()
     date_stamp = now.strftime("%Y%m%d")
     amz_date = now.strftime("%Y%m%dT%H%M%SZ")
     
-    # Query string with Action and Version
+    # Query string (sorted and URI-encoded)
-    query_params = f"Action={action}&Version={VERSION}"
+    query_params = {
+        "Action": action,
+        "Version": VERSION,
+    }
+    canonical_querystring = "&".join(
+        f"{_uri_encode(k)}={_uri_encode(v)}" 
+        for k, v in sorted(query_params.items())
+    )
     
-    # Headers
+    # Payload
-    content_type = "application/json"
+    payload = json.dumps(params, ensure_ascii=False, separators=(',', ':'))
-    payload = json.dumps(params, ensure_ascii=False)
     payload_hash = hashlib.sha256(payload.encode("utf-8")).hexdigest()
     
+    # Headers to sign (lowercase keys, trimmed values)
     headers_to_sign = {
+        "content-type": "application/json",
         "host": HOST,
-        "x-date": amz_date,
         "x-content-sha256": payload_hash,
-        "content-type": content_type,
+        "x-date": amz_date,
     }
+    
+    # Canonical headers (sorted, each ends with \n)
+    canonical_headers = ""
+    for key in sorted(headers_to_sign.keys()):
+        canonical_headers += f"{key}:{headers_to_sign[key]}\n"
+    
+    # Signed headers list
     signed_headers = ";".join(sorted(headers_to_sign.keys()))
-    canonical_headers = "".join(
-        f"{k}:{v}\n" for k, v in sorted(headers_to_sign.items())
-    )
     
     # Canonical request
     canonical_request = "\n".join([
         "POST",
         "/",
-        query_params,
+        canonical_querystring,
         canonical_headers,
         signed_headers,
         payload_hash,
@@ -127,12 +134,12 @@ async def call_volcengine_api(vendor, operation, params, api_mapping):
     )
     
     # Build request
-    url = f"{BASE_URL}/?{query_params}"
+    url = f"{BASE_URL}/?{canonical_querystring}"
     req_headers = {
         "Host": HOST,
         "X-Date": amz_date,
         "X-Content-Sha256": payload_hash,
-        "Content-Type": content_type,
+        "Content-Type": "application/json",
         "Authorization": authorization,
     }
     
@@ -141,7 +148,6 @@ async def call_volcengine_api(vendor, operation, params, api_mapping):
         hc = StreamHttpClient()
         raw = await hc.request("POST", url, headers=req_headers, data=payload.encode("utf-8"))
         
-        # Parse response
         if isinstance(raw, bytes):
             raw = raw.decode("utf-8")
         result = json.loads(raw) if raw else {}