Upload - Python

使用 Python 编程语言编写的请求示例。

---

hash_util

import hashlib


def sha256(file_path: str):
    sha256_hash = hashlib.sha256()
    with open(file_path, "rb") as f:
        # Read and update hash in chunks of 4K
        for byte_block in iter(lambda: f.read(4096), b""):
            sha256_hash.update(byte_block)
    return sha256_hash.hexdigest()

---

fire_upload

import datetime
import json
import uuid
from unittest import TestCase

import requests

import hash_util
import rsa_util


class Test(TestCase):
    # 服务商（agency）的编号
    # 此处使用 agency_no = 100000000000000000 作为模拟
    # 真实请求场景，请咨询 ice-run 的技术支持获取正式的 服务商编号 agency_no
    AGENCY_NO = "100000000000000000"

    # 安全算法 RSA
    SECURITY = "RSA"

    # ice-run 的 公钥
    PUBLIC_KEY = "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQC78QqLKTkVcwvB2Scm4IoImOXmIthVmx4HHr5Im09aol2r2UlMK3jl4lamdxktvjFswWx41hQBD4tyYFHTY+mQJsRK8Zwm7LZmLJDHcgibIuLxEddln+MI+rWwT+SOdOt7lBvd9PcO4nYr/3Nqkyk8L4PrT+GF7dpVk8iE/VpTwQIDAQAB"

    # agency 的 私钥
    PRIVATE_KEY = "MIICdwIBADANBgkqhkiG9w0BAQEFAASCAmEwggJdAgEAAoGBAOyC76Y+/2NqkkIVUtAHkbZkBYI2XXkuqE9o4oIDkbyAY04jC5MLa3oVc3uGG7T/Elodo9dIbIxNh7/ir7GQLR/hY+ookHAewurJ1z4Jn5TQeRk3+nYQcX6a6orfUOMffwcn0Jtat8D+cNAu3uCi+N7Ez/yNpItNORYqTa8AkgzrAgMBAAECgYAS/rsfl3ysb+E6RHsnsQvvYZ4dpJ8iPfCPnCVg+sdoI8mV+3ORBkBGCFYDjDRKd5fyO+IuRqdNJ2bpLtwcfy9YchcV/x40GIuAlre6A6qt4Raq+0l4FJufaqgdmSiLxd5zPX79gdMvBqTbdYzCSDGP5Pf3pTqIS3iKvSixFTQBzQJBAO6k6rE6JE/EHrbRjVtZNitDZ8rouCL0s9nGsgIDBSfgOCClO+HlZKcHq6W9ry8j1gOFeGH2rF1yMslA/ZymRL0CQQD9tk/gk9Xs1Eg1if7SLwvNxUP+z96oEqQTF26tSudPRwOO88fnfE4ZZ4heBt1QI+s7IhG39qecsFW269nvKfbHAkEArXmEgUBalQFjslGyB+1Zyyk8keuJrx9ifbRKQdwgK1R6eICkfxlZiXGx/NFeP041jGnBkLTXpzYUZOexc+YJoQJBANR3r87vnxAU+l+zr62O7oCk+XtT0y/HZJYEYpBHEQyn+MfnSXqG89R8iovLjd0GJ4E+173KlrU2SqHEQ57w8pMCQAEoDjFUXpYzBl8e0M9XsSpGY531mbOl8ASnVp4OSyavaCsn56eEkL9TI5q69KrYUEA4PN3BazMWfll2s/wu6G0="

    # ice-run 的 test 环境的 API 网关 的域名地址
    # 正式生产环境的域名地址请查阅 ice-run 的技术文档
    DOMAIN = "test-fire-api.ice.run"

    # 文件上传接口地址
    UPLOAD = "/file-service/api/upload"

    def test_file(self):
        # 对方公钥
        public_key = rsa_util.public_key(self.PUBLIC_KEY)

        # 己方私钥
        private_key = rsa_util.private_key(self.PRIVATE_KEY)

        # 此处使用 `${URL}` 模拟请求地址
        url = "https://" + self.DOMAIN + self.UPLOAD
        print("request url :", url)

        # filename = "/tmp/0.txt"
        filename = "/tmp/1.png"

        # 文件 hash
        hash_plains = hash_util.sha256(filename)

        # hash 密文：agency 使用 ice-run 公钥 对 hash 明文加密后得到的密文
        hash_cipher = rsa_util.encrypt(public_key, hash_plains)
        print("hash cipher :", hash_cipher)

        # 请求签名：agency 使用 私钥 对 hash 密文进行签名
        # 当 ice-run 收到请求时，将会使用 agency 公钥 进行验签
        sign = rsa_util.sign(private_key, hash_cipher)
        print("request sign :", sign)

        # 请求时间，当前世界标准时间的字符串，格式为 RFC-1123 标准，服务器会校验时间差，不允许超过设置的时间间隔（默认为 2 分钟）
        time = datetime.datetime.now(datetime.UTC).strftime('%a, %d %b %Y %H:%M:%S GMT')

        # 链路追踪号（防止重放攻击的请求编号）
        # 单位时间内（默认为 1 分钟）不重复的字符串，建议使用 uuid 或者时间戳+随机数。限制长度 = 32。正则表达式 = `^[0-9a-f]{32}$`
        trace = uuid.uuid4().hex

        # 请求头
        headers = {
            "X-Client": self.AGENCY_NO,
            "X-Security": self.SECURITY,
            "X-Hash": hash_cipher,
            "X-Sign": sign,
            "X-Time": time,
            "X-Trace": trace
        }
        print("request header :", headers)

        # 请求数据 request body
        # 此处使用手动拼装标准的 multipart/form-data 的 body 的数据格式，仅用于模拟。
        # 实际场景可以使用其它三方库或者自行封装。
        data = {
            "hash": hash_plains
        }
        with open(filename, "rb") as f:
            files = {
                "file": (filename, f.read())
            }

        # http 客户端。此处使用 requests 库模拟，实际场景建议自行选择 HTTP 客户端组件包
        # 发起 HTTP 请求之前，请确认 `${DOMAIN}` 可以正常访问，部分客户端的服务器环境可能需要设置请求代理
        response = requests.post(url, data=data, files=files, headers=headers)

        # 响应状态码
        status_code = response.status_code
        print("response status :", status_code)
        assert status_code == 200

        # 响应 header
        print("response header :", response.headers)

        sign = response.headers.get("X-Sign")
        print("response sign :", sign)

        # 响应 body json 格式的字符串
        response_body = response.text
        print("response body :", response_body)

        # 响应数据 response body
        # 此处使用 json 对象 模拟，实际场景建议定义数据模型类
        # 数据的 键 有 code 和 message 和 data
        # 其中 data 为 json 序列化后的字符串 RSA 加密后的密文
        response_object = response.json()

        # 响应码，默认 000000 表示请求成功
        code = response_object["code"]
        print("response code :", code)
        assert code == "000000"

        # 响应 数据密文
        data_cipher = response_object["data"]
        print("data cipher :", data_cipher)

        # 响应验签：ice-run 处理响应时，会使用 ice-run 私钥对响应密文进行签名
        # agency 收到响应后，需要使用 ice-run 公钥进行验签
        verify = rsa_util.verify(public_key, data_cipher, sign)
        if not verify:
            raise Exception("verification failed")

        # 响应 数据明文
        data_plains = rsa_util.decrypt(private_key, data_cipher)
        print("data plains :", data_plains)

        data = json.loads(data_plains)