如何用Python 脚本实现日志服务的 API 签名

容器与中间件中间件技术服务知识库
问题描述

如何使用 Python 脚本通过 openapi 完成对 TLS 服务的调用操作?

问题分析

因为 TLS 服务目前没有通过 TOP 网关,所以一些签名逻辑与 RDS 等产品 有些差异,本文会提供示例代码和差异(代码中 tips 注释),演示如何正确完成 TLS 服务的 openapi 的调用。

解决方案
import sys, os, base64, datetime, hashlib, hmac 
import requests 
import json

# ************* REQUEST VALUES *************
method = 'GET'
host = 'tls-cn-beijing.volces.com'
region = 'cn-beijing'

# tips1:  这里的 endpoint 地址需要添加 Action的名字例如 DescribeProjects
endpoint = 'https://tls-cn-beijing.volces.com/DescribeProjects'

def sign(key, msg):
  return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()

def getSignatureKey(key, dateStamp, regionName, serviceName):
  kDate = sign(key.encode('utf-8'), dateStamp)
  kRegion = sign(kDate, regionName)
  kService = sign(kRegion, serviceName)
  kSigning = sign(kService, 'request')
  return kSigning

def formatParameters(parameters):
  request_parameters_init = ''
  for key in sorted(parameters):
    request_parameters_init += key + '=' + parameters[key] + '&'
  request_parameters = request_parameters_init[:-1]
  return request_parameters

def sigv4(access_key, secret_key, service, request_parameters):
  if access_key is None or secret_key is None:
    print('No access key is available.')
    sys.exit()

  t = datetime.datetime.utcnow()
  current_date = t.strftime('%Y%m%dT%H%M%SZ')
  # current_date = '20210818T095729Z'
  datestamp = t.strftime('%Y%m%d') 

# tips2  canonical_uri  也需要设置对应 Action 的名字
  canonical_uri = '/DescribeProjects' 

  canonical_querystring = request_parameters

  signed_headers = 'content-type;host;x-content-sha256;x-date'

  payload_hash = hashlib.sha256(('').encode('utf-8')).hexdigest()

  content_type = 'application/x-www-form-urlencoded; charset=utf-8'

  canonical_headers = 'content-type:' + content_type + '\n' + 'host:' + host + '\n' + 'x-content-sha256:' + payload_hash + '\n' + 'x-date:' + current_date + '\n'
  canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash
  print(canonical_request)
  algorithm = 'HMAC-SHA256'
  credential_scope = datestamp + '/' + region + '/' + service + '/' + 'request'
  string_to_sign = algorithm + '\n' + current_date + '\n' + credential_scope + '\n' + hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()
  print(string_to_sign)
  signing_key = getSignatureKey(secret_key, datestamp, region, service)
  print(signing_key)
  signature = hmac.new(signing_key, (string_to_sign).encode('utf-8'), hashlib.sha256).hexdigest()
  print(signature)

  authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + credential_scope + ', ' + 'SignedHeaders=' + signed_headers + ', ' + 'Signature=' + signature
  print(authorization_header)
  headers = {'X-Date': current_date, 
      'Authorization':authorization_header,
      'X-Content-Sha256': payload_hash,
      'Content-Type': content_type
      }
  print(headers)

  # ************* SEND THE REQUEST *************
  request_url = endpoint + '?' + canonical_querystring

  print('\nBEGIN REQUEST++++++++++++++++++++++++++++++++++++')
  print('Request URL = ' + request_url)
  r = requests.get(request_url, headers=headers)

  print('\nRESPONSE++++++++++++++++++++++++++++++++++++')
  print('Response code: %d\n' % r.status_code)
  print(r.text)
  parsed = json.loads(r.text)
  print(json.dumps(parsed, indent=4, sort_keys=True))


if __name__=="__main__":
  access_key = '$ak'
  secret_key = '$sk'
  service = 'TLS'
# tip3 Action 和 Version 不是必选参数,可以传空
  create_instance_parameters = {
  }

  formatted_parameters = formatParameters(create_instance_parameters)
  sigv4(access_key, secret_key, service, formatted_parameters)
参考文档

https://www.volcengine.com/docs/6470/112165 如果您有其他问题,欢迎您联系火山引擎技术支持服务

44
0
0
0
相关产品
评论
未登录
看完啦,登录分享一下感受吧~
暂无评论