蚂蚁区块链BaaS登录、存证、取证示例代码

antchain_baas_deposit_demo.py

#!/usr/bin/env python
import base64
import codecs
import json
import logging
import os
import sys
import uuid
from datetime import datetime

import requests
from Cryptodome.Cipher import AES
from Cryptodome.Cipher import PKCS1_OAEP
from Cryptodome.Hash import SHA256
from Cryptodome.PublicKey import RSA
from Cryptodome.Signature import PKCS1_v1_5


class RequestException(Exception):
  pass


class AntChainBaaSApi():
  def __init__(
      self,
      *,
      access_id: str,
      chain_id: str,
      account_name: str,
      kms_id: str,
      tenant_id: str,
      base_url: str = 'https://rest.baas.alipay.com',
      access_key: str = None,
      access_key_path: str = None,
  ):
    if not access_key:
      if not access_key_path:
        raise ValueError('either access_key or access_key_path has to be provided')
      with open(access_key_path, 'rt') as file:
        access_key = file.read()
    self.access_key = access_key
    self.access_id = access_id
    self.account_name = account_name
    self.kms_id = kms_id
    self.tenant_id = tenant_id
    self.chain_id = chain_id
    self.base_url = base_url
    self._curr_token = None

  def sign(self, content):
    private_key = RSA.import_key(self.access_key)
    signer = PKCS1_v1_5.new(private_key)
    signature = signer.sign(SHA256.new(content.encode("utf-8")))
    return signature.hex()

  @staticmethod
  def _handle_response(resp):
    if resp.status_code != 200:
      raise RequestException(f'API request fails. url={resp.url}, '
                             f'status_code={resp.status_code}, body={resp.text}')
    ret = resp.json()
    if not ret['success']:
      raise RequestException(f'API returns unsuccessful result. url={resp.url}, '
                             f'status_code={resp.status_code}, body={resp.text}')
    return ret['data']

  def get_token(self):
    # reference: https://antchain.antgroup.com/docs/11/146925
    ts = int(datetime.now().timestamp() * 1000)
    secret = self.sign(f'{self.access_id}{ts}')

    full_url = f'{self.base_url}/api/contract/shakeHand'
    resp = requests.request('post', full_url, json={
      'accessId': self.access_id,
      'time': ts,
      'secret': secret,
    })
    return self._handle_response(resp)

  @property
  def curr_token(self):
    if self._curr_token is None:
      self._curr_token = self.get_token()
    return self._curr_token

  def request(self, method, url, json):
    full_url = f'{self.base_url}/api/{url}'
    json = json or {}
    json['accessId'] = self.access_id
    json['token'] = self.curr_token
    resp = requests.request(method, full_url, json=json)
    return self._handle_response(resp)

  def query_transaction(self, hash: str, decode_data: bool = True):
    ret = self.request('post', 'contract/chainCall', json={
      'method': 'QUERYTRANSACTION',
      'bizid': self.chain_id,
      'hash': hash,
      'account': self.account_name,
      'mykmsKeyId': self.kms_id,
      'tenantid': self.tenant_id,
    })
    data = json.loads(ret)
    if decode_data:
      data['transactionDO']['data'] = base64.b64decode(data['transactionDO']['data']).decode()
    return data

  def deposit(self, content, order_id):
    """Return the hash"""
    return self.request('post', 'contract/chainCallForBiz', json={
      'orderId': order_id,
      'content': content,
      'method': 'DEPOSIT',
      'bizid': self.chain_id,
      'account': self.account_name,
      'mykmsKeyId': self.kms_id,
      'tenantid': self.tenant_id,
    })


if __name__ == '__main__':
  kwargs = {
    'base_url': 'https://rest.baas.alipay.com',
    'chain_id': '<please_input_the_chain_id>',
    'account_name': '<please_input_the_account_name>',
    'kms_id': '<please_input_the_kms_id>‘,
    'access_id': '<please_input_the_access_id>',
    'access_key_path': '<please_input_the_access_key_path>',
    'tenant_id': '<please_input_the_tenant_id>',
  }
  api = AntChainBaaSApi(**kwargs)
  data = 'Hello data!'
  hash = api.deposit(data, order_id=str(uuid.uuid4()))
  print(f'{hash=}')
  trans = api.query_transaction(hash)
  print(f'{trans=}')
  assert trans['transactionDO']['data'] == data