青云OSS签名生成方法详解 如何为对象存储API生成合法请求签名 Python示例代码
为什么需要签名?
想象一下,你去银行取钱,总得证明自己吧?签名就是你在API世界里的”身份证”。青云OSS(对象存储服务)和很多云厂商一样,要求你对每个请求进行签名,这样服务器就能确认:(1)你是谁;(2)你的请求没有被中途篡改过。
没有正确签名的请求,青云OSS会直接拒绝,返回403或400错误。下面我会手把手带你把这件事搞明白。
青云OSS的签名算法基础
青云OSS的签名采用的是基于 HMAC-SHA256 的自定义签名方案,跟AWS SigV4有些相似,但也有自己的特点。核心流程大致如下:
签名字符串 = HTTP方法 + "\n"
+ Content-MD5 + "\n"
+ Content-Type + "\n"
+ Date + "\n"
+ Canonicalized青云OSSHeaders
+ CanonicalizedResource
签名 = Base64(HMAC-SHA256(签名字符串, SecretAccessKey))
Authorization头部格式为:
Authorization: QC <AccessKey>:<Signature>
一步步拆解签名的每个部分
第一部分:准备必要的凭证
你需要从青云控制台拿到两个东西:
- Access Key(AK):你的公开标识,类似用户名
- Secret Access Key(SK):你的密钥,类似密码,绝对不能泄露
这两个是你签名的”原材料”。
第二部分:构建签名字符串
这是最关键也最容易出错的地方,我们逐行来看。
HTTP方法:大写的动词,比如 GET、POST、PUT、DELETE。
Content-MD5:请求体的MD5哈希值,用Base64编码后写入。如果你没有发送请求体(比如GET请求),这里传空字符串。
Content-Type:请求的MIME类型,比如 application/json。没有的话传空字符串。
Date:注意! 青云OSS使用的是RFC 1123格式的Date,也就是 Thu, 01 Jan 2026 00:00:00 GMT 这种格式。而且这个Date必须是请求发送时的当前时间,服务器会校验时间戳,超出一定范围(通常5分钟)就会拒绝。
第三部分:构建规范化头部
规范化头部(CanonicalizedQCHeaders)是把所有以 x-qs- 开头的自定义头部按字母排序,然后拼接起来。
比如有以下头部:
x-qs-storage-class: STANDARD
x-qs-acl: public-read
规范化后就是:
x-qs-acl:public-read
x-qs-storage-class:STANDARD
每个头部占一行,格式是 Key:Value,Key和Value之间用冒号分隔,没有空格。
第四部分:构建规范化资源路径
规范化资源路径就是URL中的路径部分,加上请求参数(如果有的话)。
比如对 bucket my-bucket 中的对象 photos/2026/image.jpg 做GET请求,路径是:
/my-bucket/photos/2026/image.jpg
如果带有查询参数(比如 ?uploads),也要包含进去,按字母排序。
Python完整实现
下面是一个完整可运行的签名生成工具类,覆盖了青云OSS最常用的操作:
import hashlib
import hmac
import base64
from datetime import datetime, timezone
from urllib.parse import quote, unquote, urlencode
class QingCloudOSSClient:
"""青云OSS签名生成器"""
def __init__(self, access_key: str, secret_key: str):
"""
初始化客户端
:param access_key: 青云Access Key
:param secret_key: 青云Secret Access Key
"""
self.access_key = access_key
self.secret_key = secret_key
def _get_date_string(self) -> str:
"""生成RFC 1123格式的GMT日期字符串"""
now = datetime.now(timezone.utc)
return now.strftime("%a, %d %b %Y %H:%M:%S GMT")
def _compute_md5(self, body: str = "") -> str:
"""计算请求体的MD5并Base64编码"""
md5_hash = hashlib.md5(body.encode("utf-8")).digest()
return base64.b64encode(md5_hash).decode("utf-8")
def _canonicalize_headers(self, headers: dict) -> str:
"""
规范化自定义头部(只处理x-qs-开头的头部)
规则:
1. 只保留以 x-qs- 开头的头部
2. 按Key字母排序
3. 每个头部格式为 "key:value",用换行符拼接
"""
qs_headers = {
k.lower(): v
for k, v in headers.items()
if k.lower().startswith("x-qs-")
}
if not qs_headers:
return ""
sorted_headers = sorted(qs_headers.items())
return "\n".join(f"{k}:{v}" for k, v in sorted_headers)
def _canonicalize_resource(self, bucket: str, key: str, params: dict = None) -> str:
"""
规范化资源路径
格式: /bucket/key?params
"""
path = f"/{bucket}/{key}" if key else f"/{bucket}"
if params:
sorted_params = sorted(params.items())
query_string = urlencode(sorted_params)
path = f"{path}?{query_string}"
return path
def sign_request(
self,
method: str,
bucket: str,
key: str = "",
body: str = "",
content_type: str = "",
extra_headers: dict = None,
params: dict = None,
) -> dict:
"""
生成请求签名
:param method: HTTP方法,如 GET, POST, PUT, DELETE
:param bucket: 存储桶名称
:param key: 对象键名(对象在桶中的路径)
:param body: 请求体内容
:param content_type: Content-Type,如 application/json
:param extra_headers: 额外的自定义头部(x-qs-开头)
:param params: URL查询参数
:return: 包含签名信息的字典
"""
method = method.upper()
extra_headers = extra_headers or {}
params = params or {}
# 生成日期
date_str = self._get_date_string()
# 计算MD5
content_md5 = self._compute_md5(body)
# 构建签名字符串
canonical_headers = self._canonicalize_headers(extra_headers)
canonical_resource = self._canonicalize_resource(bucket, key, params)
string_to_sign = (
f"{method}\n"
f"{content_md5}\n"
f"{content_type}\n"
f"{date_str}\n"
f"{canonical_headers}"
f"{canonical_resource}"
)
# HMAC-SHA256签名
signature = self._compute_signature(string_to_sign)
# 返回完整的请求信息
return {
"authorization": f"QC {self.access_key}:{signature}",
"date": date_str,
"content-md5": content_md5,
"content-type": content_type,
"x-qs-date": date_str, # 某些API可能需要这个
"string_to_sign": string_to_sign, # 调试用
}
def _compute_signature(self, string_to_sign: str) -> str:
"""
使用HMAC-SHA256计算签名,并返回Base64编码结果
"""
hmac_obj = hmac.new(
self.secret_key.encode("utf-8"),
string_to_sign.encode("utf-8"),
hashlib.sha256,
)
return base64.b64encode(hmac_obj.digest()).decode("utf-8")
实际使用示例
示例一:列出存储桶中的所有对象(GET请求)
import requests
# 初始化客户端
client = QingCloudOSSClient(
access_key="YOUR_ACCESS_KEY",
secret_key="YOUR_SECRET_KEY",
)
bucket = "my-photo-bucket"
# 生成签名
signed = client.sign_request(
method="GET",
bucket=bucket,
params={"max-keys": "100"},
)
# 构建请求URL
url = f"https://{bucket}.qingstor.com/"
# 发送请求(这里假设endpoint格式)
headers = {
"Authorization": signed["authorization"],
"Date": signed["date"],
}
response = requests.get(url, headers=headers)
print(response.status_code)
print(response.text)
示例二:上传文件(PUT请求)
def upload_object(client: QingCloudOSSClient, bucket: str, key: str, file_path: str):
"""
上传对象到青云OSS
:param client: 签名客户端
:param bucket: 存储桶名称
:param key: 对象键名
:param file_path: 本地文件路径
"""
# 读取文件内容
with open(file_path, "r", encoding="utf-8") as f:
body = f.read()
# 生成签名
signed = client.sign_request(
method="PUT",
bucket=bucket,
key=key,
body=body,
content_type="text/plain",
extra_headers={
"x-qs-storage-class": "STANDARD",
# "x-qs-acl": "private", # 如果需要的话
},
)
# 构建URL
url = f"https://{bucket}.qingstor.com/{key}"
headers = {
"Authorization": signed["authorization"],
"Date": signed["date"],
"Content-MD5": signed["content-md5"],
"Content-Type": signed["content-type"],
"x-qs-storage-class": "STANDARD",
}
response = requests.put(url, data=body, headers=headers)
print(f"上传状态: {response.status_code}")
print(f"响应头: {response.headers}")
return response
示例三:删除对象(DELETE请求)
def delete_object(client: QingCloudOSSClient, bucket: str, key: str):
"""
删除对象
"""
signed = client.sign_request(
method="DELETE",
bucket=bucket,
key=key,
)
url = f"https://{bucket}.qingstor.com/{key}"
headers = {
"Authorization": signed["authorization"],
"Date": signed["date"],
}
response = requests.delete(url, headers=headers)
print(f"删除状态: {response.status_code}")
return response
签名错误排查指南
签名不对是新手最常遇到的问题,下面列出一些常见的坑:
1. 时间不同步
服务器会校验请求中的Date时间戳,如果你的本地时间和服务器时间相差超过5分钟,签名就会失败。
# 确保使用UTC时间
from datetime import datetime, timezone
now = datetime.now(timezone.utc)
date_str = now.strftime("%a, %d %b %Y %H:%M:%S GMT")
print(date_str) # 例如: Thu, 01 Jan 2026 08:30:00 GMT
2. Content-MD5计算错误
这是最容易出错的地方。注意是请求体的MD5,不是文件路径的MD5,也不是对象内容的MD5(那是另一回事)。
import hashlib
import base64
body = "hello world"
# 正确做法:对字符串的字节计算MD5,再Base64编码
md5_bytes = hashlib.md5(body.encode("utf-8")).digest()
md5_base64 = base64.b64encode(md5_bytes).decode("utf-8")
print(md5_base64) # qZkM7R4k8lLq5b5v5h5j5g== (示例)
3. 签名字符串拼接顺序错误
青云OSS的签名字符串拼接顺序是固定的:
METHOD\n
Content-MD5\n
Content-Type\n
Date\n
CanonicalizedHeaders
CanonicalizedResource
注意最后一部分,CanonicalizedHeaders 和 CanonicalizedResource 之间没有换行符!这是很多人的踩坑点。
# 正确的拼接方式
string_to_sign = (
f"{method}\n" # 第一行:HTTP方法
f"{content_md5}\n" # 第二行:Content-MD5
f"{content_type}\n" # 第三行:Content-Type
f"{date_str}\n" # 第四行:Date
f"{canonical_headers}" # 第五行:规范化头部(可能有多个\n)
f"{canonical_resource}" # 第六行:规范化资源(紧跟在头部后面)
)
4. 规范化头部排序问题
必须以小写形式排序,而不是原始大小写:
# 错误:按原始大小写排序
headers = {"X-QS-Storage-Class": "STANDARD", "x-qs-acl": "private"}
sorted(headers.items()) # 可能排序结果不对
# 正确:统一转小写后再排序
qs_headers = {k.lower(): v for k, v in headers.items() if k.lower().startswith("x-qs-")}
sorted_headers = sorted(qs_headers.items())
5. 路径编码问题
URL路径中的特殊字符需要正确编码,但某些字符(如 /)不能编码:
from urllib.parse import quote
key = "my photo/2026/image.jpg"
# 编码键名,保留 /
encoded_key = quote(key, safe="/")
print(encoded_key) # my%20photo/2026/image.jpg
一个完整的端到端示例
把上面所有东西串起来,给你一个可以直接运行的脚本:
#!/usr/bin/env python3
"""
青云OSS签名生成完整示例
演示:上传、列举、删除对象
"""
import hashlib
import hmac
import base64
import requests
from datetime import datetime, timezone
from urllib.parse import quote
class QingStorSigner:
"""青云存储签名器"""
def __init__(self, access_key: str, secret_key: str):
self.access_key = access_key
self.secret_key = secret_key
def sign(
self,
method: str,
host: str,
path: str,
body: str = "",
content_type: str = "",
headers: dict = None,
) -> dict:
headers = headers or {}
# 1. 获取当前GMT时间
date_str = datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S GMT")
# 2. 计算Content-MD5
md5_digest = hashlib.md5(body.encode("utf-8")).digest()
content_md5 = base64.b64encode(md5_digest).decode("utf-8")
# 3. 收集并排序 x-qs- 头部
qs_headers = {}
for k, v in headers.items():
if k.lower().startswith("x-qs-"):
qs_headers[k.lower()] = v
sorted_qs = sorted(qs_headers.items())
canonical_headers = "\n".join(f"{k}:{v}" for k, v in sorted_qs)
# 4. 构建签名字符串
string_to_sign = (
f"{method}\n"
f"{content_md5}\n"
f"{content_type}\n"
f"{date_str}\n"
f"{canonical_headers}"
f"{path}"
)
# 5. 计算HMAC-SHA256签名
signature = base64.b64encode(
hmac.new(
self.secret_key.encode("utf-8"),
string_to_sign.encode("utf-8"),
hashlib.sha256,
).digest()
).decode("utf-8")
return {
"authorization": f"QC {self.access_key}:{signature}",
"date": date_str,
"content-md5": content_md5,
"host": host,
"path": path,
}
def demo():
"""演示完整流程"""
# ===== 请替换为你的真实凭证 =====
ACCESS_KEY = "your-access-key"
SECRET_KEY = "your-secret-key"
BUCKET = "my-demo-bucket"
ENDPOINT = f"{BUCKET}.qingstor.com"
signer = QingStorSigner(ACCESS_KEY, SECRET_KEY)
# --- 1. 上传对象 ---
print("=" * 50)
print("1. 上传对象")
print("=" * 50)
upload_key = "test/hello.txt"
upload_body = "Hello, QingStor OSS!"
upload_headers = {
"x-qs-storage-class": "STANDARD",
}
signed = signer.sign(
method="PUT",
host=ENDPOINT,
path=f"/{upload_key}",
body=upload_body,
content_type="text/plain",
headers=upload_headers,
)
url = f"https://{ENDPOINT}/{upload_key}"
resp = requests.put(
url,
data=upload_body,
headers={
"Authorization": signed["authorization"],
"Date": signed["date"],
"Content-MD5": signed["content-md5"],
"Content-Type": "text/plain",
"x-qs-storage-class": "STANDARD",
},
)
print(f"上传状态: {resp.status_code}")
# --- 2. 获取对象 ---
print("\n" + "=" * 50)
print("2. 获取对象")
print("=" * 50)
signed_get = signer.sign(
method="GET",
host=ENDPOINT,
path=f"/{upload_key}",
)
resp = requests.get(
url,
headers={
"Authorization": signed_get["authorization"],
"Date": signed_get["date"],
},
)
print(f"获取状态: {resp.status_code}")
print(f"内容: {resp.text}")
# --- 3. 列举对象 ---
print("\n" + "=" * 50)
print("3. 列举对象")
print("=" * 50)
signed_list = signer.sign(
method="GET",
host=ENDPOINT,
path="/",
)
list_url = f"https://{ENDPOINT}/"
resp = requests.get(
list_url,
headers={
"Authorization": signed_list["authorization"],
"Date": signed_list["date"],
},
)
print(f"列举状态: {resp.status_code}")
# --- 4. 删除对象 ---
print("\n" + "=" * 50)
print("4. 删除对象")
print("=" * 50)
signed_del = signer.sign(
method="DELETE",
host=ENDPOINT,
path=f"/{upload_key}",
)
resp = requests.delete(
url,
headers={
"Authorization": signed_del["authorization"],
"Date": signed_del["date"],
},
)
print(f"删除状态: {resp.status_code}")
if __name__ == "__main__":
demo()
常见错误码对照
| HTTP状态码 | 含义 | 常见原因 |
|---|---|---|
| 400 Bad Request | 请求格式错误 | 签名字符串格式有误 |
| 403 Forbidden | 签名无效 | AccessKey/SecretKey错误,或签名计算错误 |
| 403 Forbidden | 请求超时 | 时间戳与服务器时间偏差超过5分钟 |
| 404 Not Found | 资源不存在 | bucket或key拼写错误 |
调试技巧
签名不对的时候,把签名字符串打印出来,和官方文档要求的格式逐行对比,基本能定位问题:
# 在sign()方法末尾加一行调试输出
print("--- 签名字符串 ---")
print(repr(string_to_sign))
print("--- 签名结果 ---")
print(signature)
把这段输出贴到官方文档的示例对比一下,通常一眼就能看出哪一行不对。
总结
青云OSS的签名核心就记住三句话:
- 签名字符串要严格按顺序拼接,一个换行符都不能多也不能少
- 时间要用UTC格式,RFC 1123,和服务器时间误差别超过5分钟
- x-qs-头部要小写排序,这是最容易踩的坑
理解了这三个要点,剩下的就是照着模板写代码了。祝你和青云OSS相处愉快!
