Question

简介

所有向Xiaomi Cloud-ML服务请求都需要带上请求和签名，客户端签名发送请求后，服务端会重新签名以认证用户身份。

签名算法

签名方法与AWS类似，需要注意签名的参数和顺序。 实现签名后需要通过下面的单元测试。

def test_sign(self):
  url = 'https://api.github.com/user?a=b'
  timestamp = '1474203860'
  content_md5 = 'd41d8cd98f00b204e9800998ecf8427e'
  app_secret = "sk"

  self.assertEquals(
      self.signer._sign(url, timestamp, content_md5, app_secret),
      '\x10\xe1pv\x96\x1c\x96\xfb\xc7\xe2\x16\x9d\xf4Ma5\x1dO\x86f')

def test_sign_to_base64(self):
  url = 'https://api.github.com/user?a=b'
  timestamp = '1474203860'
  content_md5 = 'd41d8cd98f00b204e9800998ecf8427e'
  app_secret = "sk"

  self.assertEquals(
      self.signer._sign_to_base64(url, timestamp, content_md5, app_secret),
      'EOFwdpYclvvH4had9E1hNR1PhmY=')

Python实现

服务端签名在cloud_ml_common目录中，代码如下。

import base64
import time
import hmac
import hashlib
from hashlib import sha1
from requests.auth import AuthBase
from urllib import unquote
from urlparse import urlparse
from constant import Constant

class Signer(AuthBase):
  ''' The signer class used to sign the request. '''
  def __init__(self, app_key, app_secret):
    self._app_key = str(app_key)
    self._app_secret = str(app_secret)
  def __call__(self, request):
    url = request.url
    if not request.body:
      request.body = ""
    timestamp = request.headers.get(Constant.TIMESTAMP, str(int(time.time())))
    content_md5 = request.headers.get(Constant.CONTENT_MD5,
                                      hashlib.md5(request.body).hexdigest())

    request.headers[Constant.TIMESTAMP] = timestamp
    request.headers[Constant.CONTENT_MD5] = content_md5
    request.headers[Constant.AUTHORIZATION] = self._sign_to_base64(
        url, timestamp, content_md5, self._app_secret)
    request.headers[Constant.SECRET_KEY_ID] = self._app_key
    return request

  def _sign(self, url, timestamp, content_md5, app_secret):
    ''' Sign the specified http request. '''

    string_to_sign = "{}\n{}\n{}\n".format(url, timestamp, content_md5)
    digest = hmac.new(app_secret, string_to_sign, digestmod=sha1)
    return digest.digest()

  def _sign_to_base64(self, url, timestamp, content_md5, app_secret):
    ''' Sign the specified request to base64 encoded result. '''

    signature = self._sign(url, timestamp, content_md5, app_secret)
    return base64.encodestring(signature).strip()

  def _get_header_value(self, http_headers, name):
    if http_headers is not None and name in http_headers:
      value = http_headers[name]
      if type(value) is list:
        return http_headers[name][0]
      else:
        return value
    return ""