test_user.py 文件源码

def test_ipfs_integration(user_contract, web3):
    # create new private key
    key = encrypt.createKey()
    # get public key from private key
    binPubKey1 = key.publickey().exportKey('DER')
    # set public key for user in blockchain
    user_contract.transact().setPubKey(binPubKey1)
    # retrieve public key for user
    binPubKey2 = user_contract.call().userPubkey(web3.eth.accounts[0])
    # import public key from blockchain
    pubKeyObj = RSA.importKey(binPubKey2)
    # encrypt message with public key of user
    enc_data = encrypt.encrypt(pubKeyObj, 'Last Night...')

    # connect to locally running IPFS daemon
    api = ipfsapi.connect('127.0.0.1', 5001)
    # add encrypted message
    res = api.add_bytes(enc_data)
    print("\n\nIPFS_HASH_CREATE:\t", res)

    # filter for YouHaveMail events on contract
    transfer_filter = user_contract.on("YouHaveMail")
    user_contract.transact().sendMessage(web3.eth.accounts[1], res)
    wait(transfer_filter)
    log_entries = transfer_filter.get()

    # extract ipfsHash from event
    res = log_entries[0]['args']['_ipfsHash']
    print("IPFS_HASH_EVENT:\t", res)
    # get stored encrypted message from IPFS
    ret = api.cat(res)
    print("IPFS_HASH_DATA:\t\t", ret)
    # decrypt message with private key of user
    print("IFPS_DATA_DECR:\t\t", encrypt.decrypt(ret, key))