青云对象存储OSS签名生成配置与使用全攻略
嘿,朋友,你是不是正在折腾青云对象存储,却被那个”签名”搞得一头雾水?别急,今天咱们就好好唠唠这件事。说实话,第一次搞云服务签名的时候,我也懵了好久——为啥非得算个东西才能上传?为啥算错了就不让访问?其实吧,签名这东西说白了就是云服务商给你的一个”身份证”,证明”嘿,这确实是我在操作”。
先给你打个预防针:这篇文章不玩虚的,咱们直接从”你拿到了AK/SK之后该怎么干”开始讲,中间会穿插我踩过的坑和总结出来的套路,保证让你读完就能上手用。
一、先搞清楚:青云OSS的签名到底是啥
青云对象存储的签名机制,跟AWS S3的签名V4协议基本是同宗同源的。简单说,它做的事情是:
把你的请求信息 + 你的Secret Key,通过一套固定算法,算出一个字符串(Signature)。青云那边拿到这个字符串,用你对应的Secret Key再算一遍,如果两遍结果一样,就认为请求合法。
听起来有点抽象?我举个例子你就懂了。
想象一下你去银行办业务,柜台阿姨不只看身份证(Access Key),还要你对暗号(Signature)。暗号就是根据当天日期 + 你的私钥 + 请求内容算出来的。你报对暗号,人家才给你办事;报错了,对不起,请回吧。
青云OSS签名的核心公式长这样:
StringToSign = HTTPMethod + "\n"
+ Content-MD5 + "\n"
+ Content-Type + "\n"
+ Date + "\n"
+ CanonicalizedHeaders + CanonicalizedResource
然后:
Signature = HMAC-SHA256(YourSecretKey, StringToSign)
最后把签名塞到请求头里,通常是这样的:
Authorization: Qiniu <AccessKey>:<Signature>
等等,这里我要纠正一个常见的误解——青云的签名头格式不是”Qiniu”,而是”QingStor”。很多人看文档看混了,结果调试半天发现Authorization格式都错了。这一点我后来才搞清楚,专门给你标出来。
二、准备工作:你需要的东西
在动手写代码之前,你先确认这几样东西准备好了:
- Access Key ID(AK) —— 就是你的用户名,公开信息,可以暴露在代码里
- Secret Key(SK) —— 就是你的密码,绝对、永远、必须保密,不要提交到GitHub,不要写在日志里
- Endpoint(端点) —— 就是你bucket所在的数据中心地址,比如
https://qingstor.com或者区域性的https://pek3a.qingstor.com - Bucket 名称 —— 你要操作的那个存储桶
这四个东西,青云控制台里都能找到。如果你还没有,先去 青云控制台 申请一对AK/SK,然后创建一个Bucket,万事俱备。
三、签名算法的详细拆解
我知道很多人看算法文档就头大,所以我把它拆碎了讲。青云OSS签名用的是 HMAC-SHA256 算法,整个过程分这几个步骤:
步骤一:准备待签名字符串
这一步是签名最核心的部分,也是最容易出错的地方。你需要拼接一个字符串,格式是:
VERB + "\n"
Content-MD5 + "\n"
Content-Type + "\n"
Date + "\n"
CanonicalizedHeaders + CanonicalizedResource
每一个字段都有讲究:
VERB 就是HTTP方法,大写:GET、PUT、POST、DELETE。
Content-MD5 是你请求体的MD5值,用Base64编码。如果是GET请求没有body,就传空字符串的Base64 MD5,也就是 1B2M2Y8AsgTpgAmY7PhCfg==。这个值我也记不住,每次都去算。
Content-Type 就是请求的MIME类型,比如 application/octet-stream,没有就传空字符串。
Date 是请求的UTC时间,格式必须是 Wdy, DD Mon YYYY HH:MM:SS GMT,比如 Tue, 15 Nov 1994 08:12:31 GMT。注意这个格式要求非常严格,少一个字母都不行。
步骤二:处理 CanonicalizedHeaders
如果请求里有自定义Header,需要对它们进行规范化。规则是:
- Header名统一转小写
- 以
x-qs-或x-amz-开头的Header需要参与签名计算 - 按字母顺序排序
- 每个Header格式为
key:value\n,多个用\n连接 - key和value之间用单个冒号分隔,value前后不能有多余空格
举个例子,假设你有两个自定义Header:
x-qs-acl: public-read
x-qs-storage-class: STANDARD
规范化后就是:
x-qs-acl:public-read
x-qs-storage-class:STANDARD
步骤三:处理 CanonicalizedResource
这个指的是你请求的资源路径,格式是:
/BucketName/ObjectName
如果有查询参数(比如 ?uploads、?acl),也要拼上去。注意:路径一定要以 / 开头。
比如你要对一个叫 my-bucket 的Bucket里的 images/photo.jpg 文件做操作,CanonicalizedResource就是:
/my-bucket/images/photo.jpg
如果要列出bucket里的对象,加上查询参数:
/my-bucket/?prefix=&max-keys=100
步骤四:计算签名
把上面拼好的 StringToSign 和你的 Secret Key 做 HMAC-SHA256,然后Base64编码:
import hmac
import hashlib
import base64
string_to_sign = "GET\n\n\nTue, 15 Nov 1994 08:12:31 GMT\n/my-bucket/images/photo.jpg"
secret_key = "your-secret-key-here"
signature = base64.b64encode(
hmac.new(
secret_key.encode('utf-8'),
string_to_sign.encode('utf-8'),
hashlib.sha256
).digest()
).decode('utf-8')
print(signature)
出来的结果就是一个签名字符串,把它放进Authorization头就可以了:
Authorization: QingStor <AK>:<Signature>
四、完整代码示例:用Python实现全套操作
光说不练假把式,下面我给你一个完整的Python实现,涵盖了创建签名、上传文件、下载文件、列出对象等常用操作。代码我写得很详细,每个函数都有注释,你可以直接复制用。
import hmac
import hashlib
import base64
import hashlib
import time
import requests
from datetime import datetime, timezone
from urllib.parse import quote, unquote
class QingStorClient:
"""
青云对象存储客户端
封装了签名生成和常用OSS操作
"""
def __init__(self, access_key, secret_key, endpoint, region='pek3a'):
"""
初始化客户端
参数:
access_key: 你的Access Key ID
secret_key: 你的Secret Key(务必保密!)
endpoint: 服务终端节点,例如 'qingstor.com'
region: 区域代码,例如 'pek3a'(北京A)、'sh1a'(上海A)等
"""
self.access_key = access_key
self.secret_key = secret_key
self.endpoint = endpoint
self.region = region
self.base_url = f'https://{region}.{endpoint}'
def _get_utc_time_str(self):
"""获取格式化的UTC时间字符串"""
now = datetime.now(timezone.utc)
return now.strftime('%a, %d %b %Y %H:%M:%S GMT')
def _get_date_str(self):
"""获取日期字符串(用于Date header)"""
now = datetime.now(timezone.utc)
return now.strftime('%Y%m%d')
def _get_canonicalized_headers(self, headers=None):
"""
规范化自定义Header
只保留以 x-qs- 开头的Header,并按字母序排列
"""
canonical = ''
if not headers:
return canonical
# 筛选出 x-qs- 开头的header
qs_headers = {
k.lower(): v.strip()
for k, v in headers.items()
if k.lower().startswith('x-qs-')
}
# 按key排序
for key in sorted(qs_headers.keys()):
canonical += f'{key}:{qs_headers[key]}\n'
return canonical
def _get_canonicalized_resource(self, bucket, key='', query_string=''):
"""
构建规范化的资源路径
参数:
bucket: Bucket名称
key: 对象Key(文件名路径)
query_string: 查询参数(如 '?uploads')
"""
# 对key进行URL编码,但保留 / 不被编码
encoded_key = ''
if key:
# 按 / 分割,逐段编码
parts = key.split('/')
encoded_key = '/'.join(quote(p, safe='') for p in parts)
resource = f'/{bucket}{encoded_key}'
if query_string:
resource += query_string
return resource
def sign_request(self, method, bucket, key='', headers=None, query_string=''):
"""
生成请求签名
参数:
method: HTTP方法 (GET/PUT/POST/DELETE)
bucket: Bucket名称
key: 对象Key
headers: 额外请求头
query_string: 查询字符串
返回:
签名字符串和完整的请求头字典
"""
# 1. 准备基础信息
content_md5 = ''
content_type = ''
date = self._get_utc_time_str()
if headers:
content_md5 = headers.get('Content-MD5', '')
content_type = headers.get('Content-Type', '')
# 2. 构建规范化Header
canonical_headers = self._get_canonicalized_headers(headers)
# 3. 构建规范化资源路径
canonical_resource = self._get_canonicalized_resource(bucket, key, query_string)
# 4. 拼接待签名字符串
string_to_sign = (
f'{method}\n'
f'{content_md5}\n'
f'{content_type}\n'
f'{date}\n'
f'{canonical_headers}'
f'{canonical_resource}'
)
# 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')
# 6. 构建完整的请求头
request_headers = {
'Date': date,
'Authorization': f'QingStor {self.access_key}:{signature}',
}
if content_md5:
request_headers['Content-MD5'] = content_md5
if content_type:
request_headers['Content-Type'] = content_type
# 合并自定义header
if headers:
for k, v in headers.items():
if k.lower() not in ('content-md5', 'content-type', 'date', 'authorization'):
request_headers[k] = v
return signature, request_headers, string_to_sign
def put_object(self, bucket, key, body, content_type='application/octet-stream',
storage_class='STANDARD', acl='private'):
"""
上传对象到OSS
参数:
bucket: Bucket名称
key: 对象Key(文件路径)
body: 文件内容(字节或字符串)
content_type: MIME类型
storage_class: 存储类型
acl: 访问权限
返回:
HTTP响应对象
"""
# 准备body
if isinstance(body, str):
body_bytes = body.encode('utf-8')
else:
body_bytes = body
# 计算MD5
md5_hash = hashlib.md5(body_bytes).digest()
content_md5 = base64.b64encode(md5_hash).decode('utf-8')
# 准备headers
headers = {
'Content-Type': content_type,
'Content-MD5': content_md5,
'x-qs-storage-class': storage_class,
'x-qs-acl': acl,
}
# 生成签名
signature, request_headers, string_to_sign = self.sign_request(
method='PUT',
bucket=bucket,
key=key,
headers=headers
)
# 发送请求
url = f'{self.base_url}/{bucket}/{key}'
response = requests.put(url, data=body_bytes, headers=request_headers)
return response
def get_object(self, bucket, key):
"""
从OSS下载对象
参数:
bucket: Bucket名称
key: 对象Key
返回:
HTTP响应对象
"""
headers = {}
signature, request_headers, string_to_sign = self.sign_request(
method='GET',
bucket=bucket,
key=key,
headers=headers
)
url = f'{self.base_url}/{bucket}/{key}'
response = requests.get(url, headers=request_headers)
return response
def list_objects(self, bucket, prefix='', max_keys=100):
"""
列出Bucket中的对象
参数:
bucket: Bucket名称
prefix: 对象Key前缀过滤
max_keys: 最多返回数量
返回:
HTTP响应对象,解析后为JSON
"""
query_string = f'?prefix={quote(prefix, safe="")}&max-keys={max_keys}'
headers = {}
signature, request_headers, string_to_sign = self.sign_request(
method='GET',
bucket=bucket,
headers=headers,
query_string=query_string
)
url = f'{self.base_url}/{bucket}{query_string}'
response = requests.get(url, headers=request_headers)
return response
def delete_object(self, bucket, key):
"""
删除OSS中的对象
参数:
bucket: Bucket名称
key: 对象Key
返回:
HTTP响应对象
"""
headers = {}
signature, request_headers, string_to_sign = self.sign_request(
method='DELETE',
bucket=bucket,
key=key,
headers=headers
)
url = f'{self.base_url}/{bucket}/{key}'
response = requests.delete(url, headers=request_headers)
return response
def create_bucket(self, bucket, storage_class='STANDARD', acl='private'):
"""
创建Bucket
参数:
bucket: Bucket名称
storage_class: 存储类型
acl: 访问权限
返回:
HTTP响应对象
"""
headers = {
'x-qs-storage-class': storage_class,
'x-qs-acl': acl,
}
signature, request_headers, string_to_sign = self.sign_request(
method='PUT',
bucket=bucket,
headers=headers
)
url = f'{self.base_url}/{bucket}'
response = requests.put(url, headers=request_headers)
return response
# ==================== 使用示例 ====================
if __name__ == '__main__':
# 你的青云OSS配置(从控制台获取)
ACCESS_KEY = 'your-access-key-here'
SECRET_KEY = 'your-secret-key-here'
ENDPOINT = 'qingstor.com'
REGION = 'pek3a' # 北京A
# 创建客户端实例
oss_client = QingStorClient(
access_key=ACCESS_KEY,
secret_key=SECRET_KEY,
endpoint=ENDPOINT,
region=REGION
)
BUCKET_NAME = 'my-test-bucket'
# 示例1:创建Bucket
print('=== 创建Bucket ===')
resp = oss_client.create_bucket(BUCKET_NAME)
print(f'状态码: {resp.status_code}')
print(f'响应内容: {resp.text}')
# 示例2:上传文件
print('\n=== 上传文件 ===')
test_content = 'Hello, QingStor OSS! 这是一段测试文本。'
resp = oss_client.put_object(
bucket=BUCKET_NAME,
key='hello.txt',
body=test_content,
content_type='text/plain; charset=utf-8'
)
print(f'状态码: {resp.status_code}')
print(f'响应内容: {resp.text}')
# 示例3:下载文件
print('\n=== 下载文件 ===')
resp = oss_client.get_object(BUCKET_NAME, 'hello.txt')
print(f'状态码: {resp.status_code}')
print(f'文件内容: {resp.text}')
# 示例4:列出对象
print('\n=== 列出对象 ===')
resp = oss_client.list_objects(BUCKET_NAME)
print(f'状态码: {resp.status_code}')
print(f'响应内容: {resp.text}')
# 示例5:删除文件
print('\n=== 删除文件 ===')
resp = oss_client.delete_object(BUCKET_NAME, 'hello.txt')
print(f'状态码: {resp.status_code}')
print(f'响应内容: {resp.text}')
上面的代码我写得尽量详细,你复制过去之后只需要把 ACCESS_KEY 和 SECRET_KEY 替换成你自己的,就能跑起来了。
五、常见错误及排错指南
写代码的人都知道,跑通了是爹,跑不通是祖宗。下面是我踩过的坑,给你列出来,你就不用再踩了:
错误一:SignatureDoesNotMatch
这是最常见的错误,几乎每个新手都会遇到。原因通常是:
- 时间不同步——青云服务器会用请求头里的Date时间和服务器时间做比对,如果你的服务器时间跟青云的差了超过15分钟,签名就直接无效了。解决办法是同步NTP时间,或者在代码里强制用青云的日期。
# Linux上同步时间
sudo ntpdate pool.ntp.org
待签名字符串拼接错误——注意换行符
\n的数量和位置,一个都不能错。特别是CanonicalizedHeaders后面如果没有自定义Header,就是空字符串,但也得有那部分。Secret Key不对——确认你用的SK跟AK是配对的,一个AK对应一个SK,别混了。
错误二:403 Forbidden
这个错误一般是权限问题:
- Bucket的ACL设置不允许你操作(比如你设了
private但你没签名) - 你请求的方法不对(比如用GET请求一个只有PUT权限的资源)
- IP白名单限制(如果控制台开了IP限制)
错误三:MD5不匹配
上传文件的时候如果报了MD5校验失败,检查一下:
- Content-MD5是不是Base64编码的
- MD5是不是对整个body计算的(不是对字符串)
- 上传的时候Content-MD5头和实际body是不是一致的
# 正确的MD5计算方式
import hashlib
import base64
body = b'hello world'
md5_digest = hashlib.md5(body).digest() # 得到的是原始字节
content_md5 = base64.b64encode(md5_digest).decode('utf-8') # 再Base64编码
print(content_md5) # 结果:XrY7u+Ae7tCTyyK7j1rNww==
错误四:日期格式不对
Date header的格式必须严格符合 RFC 1123,差一个逗号都不行:
正确的格式:
Tue, 15 Nov 1994 08:12:31 GMT
错误的格式(会报错):
2024-11-15T08:12:31Z # ISO格式不行
Tue, 15 November 1994 08:12:31 GMT # 月份不能写全称
15 Nov 1994 08:12:31 GMT # 缺少星期几
六、用其他语言实现签名
如果你不用Python,这里再给你几个其他语言的实现参考。原理都是一样的,只是语法不同。
Go 语言实现
package main
import (
"crypto/hmac"
"crypto/md5"
"crypto/sha256"
"encoding/base64"
"fmt"
hash"
"strings"
"time"
)
func signRequest(method, bucket, key, accessKey, secretKey string, headers map[string]string) string {
// 获取基础信息
contentMD5 := headers["Content-MD5"]
contentType := headers["Content-Type"]
date := time.Now().UTC().Format(time.RFC1123)
// 规范化Header
canonicalHeaders := ""
var qsKeys []string
for k := range headers {
if strings.HasPrefix(strings.ToLower(k), "x-qs-") {
qsKeys = append(qsKeys, k)
}
}
sort.Strings(qsKeys)
for _, k := range qsKeys {
canonicalHeaders += strings.ToLower(k) + ":" + headers[k] + "\n"
}
// 规范化资源路径
canonicalResource := fmt.Sprintf("/%s/%s", bucket, key)
// 构建待签名字符串
stringToSign := fmt.Sprintf("%s\n%s\n%s\n%s\n%s%s",
method, contentMD5, contentType, date, canonicalHeaders, canonicalResource)
// 计算HMAC-SHA256
mac := hmac.New(sha256.New, []byte(secretKey))
mac.Write([]byte(stringToSign))
signature := base64.StdEncoding.EncodeToString(mac.Sum(nil))
return signature
}
func main() {
accessKey := "your-access-key"
secretKey := "your-secret-key"
headers := map[string]string{
"Content-MD5": "",
"Content-Type": "text/plain",
"x-qs-storage-class": "STANDARD",
}
sig := signRequest("PUT", "my-bucket", "hello.txt", accessKey, secretKey, headers)
fmt.Printf("Signature: %s\n", sig)
fmt.Printf("Authorization: QingStor %s:%s\n", accessKey, sig)
}
JavaScript (Node.js) 实现
const crypto = require('crypto');
/**
* 生成青云OSS请求签名
*/
function signRequest(method, bucket, key, options = {}) {
const {
accessKey,
secretKey,
contentType = '',
contentMD5 = '',
customHeaders = {},
queryString = ''
} = options;
// 获取UTC时间
const date = new Date().toUTCString();
// 规范化自定义Header
const qsHeaders = Object.keys(customHeaders)
.filter(k => k.toLowerCase().startsWith('x-qs-'))
.sort()
.map(k => `${k.toLowerCase()}:${customHeaders[k].trim()}`)
.join('\n');
// 构建规范化资源路径
const canonicalResource = `/${bucket}/${key}${queryString}`;
// 构建待签名字符串
const stringToSign = [
method,
contentMD5,
contentType,
date,
qsHeaders + '\n',
canonicalResource
].join('\n');
// 计算HMAC-SHA256签名
const signature = crypto
.createHmac('sha256', secretKey)
.update(stringToSign)
.digest('base64');
return {
signature,
authorization: `QingStor ${accessKey}:${signature}`,
date,
stringToSign
};
}
// 使用示例
const result = signRequest('PUT', 'my-bucket', 'hello.txt', {
accessKey: 'YOUR_ACCESS_KEY',
secretKey: 'YOUR_SECRET_KEY',
contentType: 'text/plain; charset=utf-8',
customHeaders: {
'x-qs-storage-class': 'STANDARD',
'x-qs-acl': 'private'
}
});
console.log('Signature:', result.signature);
console.log('Authorization:', result.authorization);
七、一些实用的技巧和注意事项
聊完基础,再给你几个老司机才懂的Tips:
1. 缓存签名
签名的有效期通常是15分钟(青云服务器会校验Date header和时间差)。所以如果你要在一个短时间内做很多请求,可以缓存签名,不用每次都重新算。但要注意,缓存的Date header和实际发送请求时的Date要一致。
2. 用SKIM(青云官方SDK)
如果你不想自己写签名逻辑,青云官方提供了SDK,Python、Go、Java都有。我之所以写这些签名代码,是为了让你理解原理,实际操作中直接用SDK更省事。
# 安装Python SDK
pip install qingstor
from qingstor.sdk.config import Config
from qingstor.sdk.service.oss import QingStor
config = Config(
access_key_id='YOUR_ACCESS_KEY',
secret_access_key='YOUR_SECRET_KEY',
zone='pek3a'
)
oss = QingStor(config)
# 直接用SDK操作,签名自动生成
3. 生产环境不要硬编码SK
这点我强调一遍——Secret Key千万不要写死在代码里。正确的做法是:
- 放在环境变量里:
export QINGSTOR_SECRET_KEY=your-key - 放在配置中心或密钥管理服务里
- 用
.env文件配合 python-dotenv 加载,但.env文件一定要加到.gitignore里
4. 理解CORS和预签名URL
如果你要做Web前端直接上传文件到OSS,光有签名不够,还需要处理CORS跨域问题。青云OSS支持生成预签名URL,过期时间自定义,前端拿着这个URL就能直接上传,不需要暴露你的AK/SK。
def generate_presigned_url(self, bucket, key, method='PUT', expiration=3600):
"""
生成预签名URL,有效期expiration秒
"""
expires_at = int(time.time()) + expiration
# 青云的预签名URL格式
url = f'{self.base_url}/{bucket}/{key}'
# 实际项目中建议用官方SDK生成,这里只是说明概念
return url
5. 测试环境先用小规模数据
在正式环境跑签名代码之前,建议先用测试Bucket、小文件跑一遍全流程。确认签名正确、上传下载都OK了,再上正式数据。我有一次就是因为签名有个隐藏bug,结果把测试数据搞乱了,得不偿失。
八、总结
好了,这篇文章写到这里,应该够你搞定青云OSS的签名问题了。我再 recap 一下核心要点:
- 签名本质 = HMAC-SHA256(你的SK, 规范化的请求信息)
- 待签名字符串 = 方法 + Content-MD5 + Content-Type + Date + 规范化Header + 规范化资源路径
- 日期格式 必须严格遵循 RFC 1123
- Authorization头 格式是
QingStor <AK>:<Signature> - 生产环境 绝对不要把SK硬编码到代码里
签名这东西,理解原理之后其实不难,难就难在细节上——一个换行符、一个大小写、一个空格,都能让你调半天。希望这篇文章能帮你少走点弯路。
要是还有问题,欢迎在评论区留言,咱们一起讨论。毕竟踩过的坑多了,也就成经验了嘛 😄
