云服务商的api鉴权

2026-04-04 2146 words 5 minutes

Contents

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302


import base64
import datetime
import hashlib
import hmac
import json
import os
import sys
import uuid
from typing import Optional, Union
from urllib.parse import quote, urlencode
import requests

METHOD_GET = "GET"
METHOD_POST = "POST"
file = False


def hmac_sha256(secret, data):  # type: (Union[bytes, str], str) -> bytes
    return hmac.new(
        key=(
            bytearray(secret)
            if type(secret) == bytes
            else bytearray(secret, encoding="utf8")
        ),
        msg=bytearray(data, encoding="utf8"),
        digestmod=hashlib.sha256,
    ).digest()


def base64_of_hmac(data):  # type: (bytes) -> bytes
    return base64.b64encode(s=data)


def get_request_uuid():  # type: () -> str
    return str(uuid.uuid1())


def get_sorted_str(data):  # type: (dict) -> str
    return "&".join(
        map(
            lambda x_y: "%s=%s" % (x_y[0], x_y[1]),
            sorted(data.items(), key=lambda item: item[0]),
        )
    )


def encode_query_str(query):  # type: (str) -> str
    after_query = ""

    if len(query) != 0:
        param = query.split("&")
        param.sort()

        for str in param:
            if len(after_query) < 1:
                s = str.split("=")

                if len(s) <= 2:
                    encode_str = quote(s[1])
                    str = s[0] + "=" + encode_str
                    after_query = after_query + str
                else:
                    encode_str = ""
                    str = s[0] + "=" + encode_str
                    after_query = after_query + str
            else:
                s = str.split("=")

                if len(s) >= 2:
                    encode_str = quote(s[1])
                    str = s[0] + "=" + encode_str
                    after_query = after_query + "&" + str
                else:
                    encode_str = ""
                    str = s[0] + "=" + encode_str
                    after_query = after_query + "&" + str

    return after_query


def print_log(log_info):  # type: (str) -> None
    print("[%s]: %s" % (str(object=datetime.datetime.now()), log_info))


def build_sign(
    query_params, body_params, eop_date, request_uuid, method
):  # type: (dict, Union[bytearray, dict], str, str, str) -> str
    body_str = json.dumps(body_params) if not file and body_params else ""
    body = (
        (
            body_params
            if file
            else (
                json.dumps(body_params).encode(encoding="utf-8")
                if isinstance(body_params, dict)
                else body_params.encode(encoding="utf-8")
            )
        )
        if method == METHOD_POST
        else body_str.encode(encoding="utf-8")
    )
    header_str = "ctyun-eop-request-id:%s\neop-date:%s\n" % (
        request_uuid,
        eop_date,  # 格式为：%Y%m%dT%H%M%SZ。
    )  # 请求头中必要的两个参数。
    query_str = encode_query_str(query=get_sorted_str(data=query_params))
    sign_date = eop_date.split("T")[0]

    # 计算鉴权密钥。
    k_time = hmac_sha256(data=eop_date, secret=sk)
    k_ak = hmac_sha256(data=ak, secret=k_time)
    k_date = hmac_sha256(data=sign_date, secret=k_ak)

    # 构建请求头的鉴权字段值。
    signature_str = "%s\n%s\n%s" % (
        header_str,
        query_str,
        hashlib.sha256(body).hexdigest(),
    )
    sign_header = "%s Headers=ctyun-eop-request-id;eop-date Signature=%s" % (
        ak,
        base64_of_hmac(data=hmac_sha256(data=signature_str, secret=k_date)).decode(
            encoding="utf8"
        ),
    )

    return sign_header.encode(encoding="utf8")


def get_sign_headers(
    query_params, body, method, content_type
):  # type: (dict, dict, str, str) -> dict
    eop_date = datetime.datetime.strftime(datetime.datetime.now(), "%Y%m%dT%H%M%SZ")
    request_uuid = get_request_uuid()
    return {
        "Content-type": content_type,
        "ctyun-eop-request-id": request_uuid,
        "Eop-Authorization": build_sign(
            body_params=body,
            eop_date=eop_date,
            method=method,
            query_params=query_params,
            request_uuid=request_uuid,
        ),
        "Eop-date": eop_date,
    }


def execute(
    url,
    query,
    method,
    params=None,
    header_params=None,
    content_type="application/json;charset=UTF-8",
):  # type: (str, str, str, Optional[dict], Optional[dict], str) -> requests.Response
    if "application/x-www-form-urlencoded" in content_type:
        params = urlencode(params)

    header_params = header_params or {}
    params = params or {}

    if method == METHOD_GET:
        body, query_params = ({}, params)
    else:
        body, query_params = (params, {})

        if len(query) > 0:
            for q in query.split("&"):
                query_params[q.split("=")[0]] = q.split("=")[1]

    print("\n" + "-" * 10)
    headers = get_sign_headers(query_params, body, method, content_type)
    headers.update(header_params)
    headers.update(
        {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:109.0) Gecko/20100101 Firefox/110.0"
        }
    )

    if query:
        url = url + "?" + encode_query_str(query)

    # fmt: off
    print_log(u"URL: %s" % url)
    print_log(u"请求方式: %s" % method)
    print_log(u"请求头: %s" % headers)
    print_log(u"请求参数: %s" % params)
    print_log(u"请求参数类型: %s" % type(params))
    # fmt: on

    requests.packages.urllib3.disable_warnings()

    if method == METHOD_GET:
        res = requests.get(url, params=params, headers=headers, verify=False)
    else:
        if "application/x-www-form-urlencoded" in content_type:
            res = requests.post(data=params, headers=headers, url=url, verify=False)
        elif "multipart/form-data" in content_type:
            res = requests.post(data=params, headers=headers, url=url, verify=False)
        else:
            res = requests.post(headers=headers, json=params, url=url, verify=False)

    # fmt: off
    print_log(u"返回状态码: %s" % res.status_code)
    print_log(u"返回: %s" % res.text)
    # fmt: on

    return res


def get(
    url, query="", params=None, header_params=None
):  # type: (str, str, Optional[dict], Optional[dict]) -> requests.Response
    """发送 GET 请求。

    Parameters
    ----------
    url : str
        请求的 URL。

    query : str, optional
        请求的查询字符串（默认值：空字符串）。

    params : dict, optional
        请求的参数（默认值：空）。

    header_params : dict, optional
        请求的头部参数（默认值：空）。

    Returns
    -------
    requests.Response
        请求的响应。
    """
    return execute(
        header_params=header_params,
        method=METHOD_GET,
        params=params,
        query=query,
        url=url,
    )


def post(
    url,
    query="",
    params=None,
    header_params=None,
    content_type="application/json;charset=UTF-8",
):
    """发送 POST 请求。

    Parameters
    ----------
    url : str
        请求的 URL。

    query : str, optional
        请求的查询字符串（默认值：空字符串）。

    params : dict, optional
        请求的参数（默认值：空）。

    header_params : dict, optional
        请求的头部参数（默认值，空）。

    content_type : str, optional
        请求的内容类型（默认值：application/json;charset=UTF-8）。

    Returns
    -------
    requests.Response
        请求的响应。
    """
    return execute(
        content_type=content_type,
        header_params=header_params,
        method=METHOD_POST,
        params=params,
        query=query,
        url=url,
    )


if __name__ == "__main__":
    # 福州池的API地址为ces-fjfz-b.ctapi.ctyun.cn
    # 鉴权的AK/SK
    ak = "xxx"
    sk = "xxx"

    # NOTE: 以下为调用示例，请按实际情况做出修改。
    get(
        # params={"regionID": "d7d93102848711ea9ff10242ac110002"},
        url='https://ces-fjfz-b.ctapi.ctyun.cn/v3/projects',
    )
    # post(
    #     params={"regionID": "d7d93102848711ea9ff10242ac110002", "number": 1},
    #     url="https://monitor-global.ctapi.ctyun.cn/v4/monitor/query-mem-top",
    # )

# https://ctecs-global.ctapi.ctyun.cn/v4/region/list-regions

核心摘要

上述的签名鉴权流程核心是一个基于 HMAC-SHA256 算法的链式密钥派生和签名计算过程，确保 API 请求的完整性与安全性
整个流程可概括为以下四个关键步骤

准备 AK/SK 与元数据
构造待签名字符串

格式化请求头
排序并编码查询参数
计算请求体的 SHA256 哈希
链式计算动态密钥

k_time、k_ak、k_date
生成最终签名与鉴权头

signature
组装并发送请求

拼接 Eop-Authorization 字段
格式: AK Headers=… Signature=…

详细代码解析与原理说明

1. 密钥与参数准备

代码首先定义了访问密钥 ak 和 sk，并生成了每个请求必须的唯一标识 ctyun-eop-request-id (通过 UUID 生成) 和精确到秒的请求时间 eop-date (格式为 YYYYMMDDTHHMMSSZ)

1
2
3
4


ak = "xxx"
sk = "xxx"
eop_date = datetime.datetime.strftime(datetime.datetime.now(), "%Y%m%dT%H%M%SZ")
request_uuid = get_request_uuid()

2. 构造待签名字符串 (`build_sign`函数核心)

这是签名计算的第一步，将请求的三个关键部分按规则拼接成一个字符串

请求头部分

仅包含 ctyun-eop-request-id 和 eop-date 两个必填头，并按名称排序后格式化
1

header_str = "ctyun-eop-request-id:%s\neop-date:%s\n" % (request_uuid, eop_date)
查询参数部分

对 query_params 字典按键名排序，然后拼接成 key=value&key=value 的形式。代码中的 encode_query_str 函数对值进行了 URL 编码（quote），符合 RFC 3986 规范
1

query_str = encode_query_str(query=get_sorted_str(data=query_params))

请求体部分

对原始请求体（Body）进行 SHA-256 哈希计算，并转换为 16 进制字符串。对于 GET 请求，请求体为空

1
2
3


body_str = json.dumps(body_params) if not file and body_params else ""
body = ... # 编码为bytes
hashlib.sha256(body).hexdigest()

最终拼接

将上述三部分用换行符 \n 连接

1

signature_str = "%s\n%s\n%s" % (header_str, query_str, hashlib.sha256(body).hexdigest())

3. 链式计算动态密钥

使用 HMAC-SHA256 算法，通过一系列链式计算，从固定的 sk 派生出一个具有时效性的签名密钥

1
2
3


k_time = hmac_sha256(data=eop_date, secret=sk)
k_ak = hmac_sha256(data=ak, secret=k_time)
k_date = hmac_sha256(data=sign_date, secret=k_ak) # sign_date是eop_date的前8位

这种链式结构使得最终签名密钥 k_date 不仅依赖于永久密钥 sk，还依赖于请求时间 eop_date，从而保证了签名的时效性和安全性

4. 生成最终签名与鉴权头

用上一步得到的 k_date 作为密钥，对步骤 2 构造的 signature_str 进行 HMAC-SHA256 签名，并进行 Base64 编码
最后，将 ak、参与签名的头列表和签名值按固定格式拼接到 Eop-Authorization 头中

1
2
3
4


sign_header = "%s Headers=ctyun-eop-request-id;eop-date Signature=%s" % (
    ak,
    base64_of_hmac(data=hmac_sha256(data=signature_str, secret=k_date)).decode(encoding="utf8"),
)