Using GDK to hold Blockstream AMP Assets

The example code on this page is intended to demonstrate the use of common Blockstream Green Development Kit (GDK) calls in Python. It shows how to install GDK and set up a Managed Assets account within the wallet so it is capable of holding assets issued using Blockstream AMP. The example covers:

  1. Login with an existing 24 word mnemonic or generate a new mnemonic.

  2. Set a pin for the wallet and login with the pin.

  3. Check and set the wallet’s Two Factor Authentication (2FA) status.

  4. Handle calls to methods that use 2FA, such as a sending transaction.

  5. Handle events in the Green notification queue, such as new block.

  6. Get a new address and an address pointer.

  7. Get all wallet related transactions from Green, or only ones from a certain block height.

GDK Overview

The Blockstream Green Development Kit (GDK) is a cross-platform library for Blockstream Green wallets that can be used through Python and Java wrappers.

GDK enables you to develop solutions that benefit from the infrastructure that supports the Blockstream Green mobile and desktop wallets.

The example creates a Managed Assets account within the wallet which allows it to accept Blockstream AMP issued assets. It can easily be amended to allow a wallet to receive Bitcoin (BTC), Liquid Bitcoin (L-BTC) and other Liquid issued assets.

The GDK documentation can be used to expand the example further and a good example of a well-featured client application using GDK is the command line interface green-cli.

How to install and run

Create a new directory named gdk-example-python.

Within that directory, create a new file named gdk-example-python.py and copy and paste the example code at the end of the page into it.

Within that directory, also create a new file named requirements.txt and paste the following into it:

https://github.com/Blockstream/gdk/releases/download/release_0.0.33/greenaddress-0.0.33-cp36-cp36m-linux_x86_64.whl; 'linux' in sys_platform and python_version == '3.6' --hash=sha256:c35a15cbbbfedda785cf019bd0691e582219f6a1a491a8d979551dd1a7b0fb6a
https://github.com/Blockstream/gdk/releases/download/release_0.0.33/greenaddress-0.0.33-cp37-cp37m-linux_x86_64.whl; 'linux' in sys_platform and python_version == '3.7' --hash=sha256:cd15e21699cd2ffe4229af4b400ac78fa06e93356e5055d94e83a449ceab1a92
https://github.com/Blockstream/gdk/releases/download/release_0.0.33/greenaddress-0.0.33-cp37-cp37m-macosx_10_14_x86_64.whl; 'darwin' in sys_platform and python_version == '3.7' --hash=sha256:e58be5b848f7a5c71b7872f3cd8588d5e89128473b022a71210da5f095208d8e

From the terminal, move into the new directory:

cd gdk-example-python

Create a virtual environment (if you want to use one):

virtualenv -p python3 venv

If you use a virtual environment, activate it:

source venv/bin/activate

In either environment, install the requirements:

pip install -r requirements.txt

The above should install the GDK library. Alternatively, you can install it by downloading and installing the GDK python wheel for your platform from the GDK release page.

The ‘cp’ in the name refers to the Python version you have installed.

For example for a Python 3.7.* install on a 64 bit Linux platform you would download greenaddress-0.0.33-cp37-cp37m-linux_x86_64.whl and install using pip:

pip install greenaddress-0.0.33-cp37-cp37m-linux_x86_64.whl

Run the example code:

python gdk_example.py

The example python code

Below is the example code that wraps the calls to GDK within a utility class.

import greenaddress as gdk
import json


def main():

    # Create our GDK wallet wrapper.
    # This will handle all our calls to GDK.
    wallet = wallet_utils()


    # Wallet creation and login using Mnemonic
    # ========================================

    # To create a wallet with a Managed Assets account, pass a mnemonic
    # into the following. You can generate a 24 word mnemonic yourself or
    # have GDK generate it for you by leaving mnemonic as None.
    # You can choose to create a wallet that's covered by 2FA or not.
    # 2FA can be activated or deactivated at any point in time.
    """
    mnemonic = wallet.create_new_wallet(create_with_2fa_enabled=True, mnemonic=None)
    print(f'\nMnemonic: {mnemonic}')
    """

    # To login to an existing wallet you can either use the mnemonic or pin.
    # Later we'll see how to use a pin, for now we will use the mnemonic.
    mnemonic = 'your twenty four word mnemonic with spaced words here'
    if not wallet.is_valid_mnemonic(mnemonic):
        raise Exception("Invalid mnemonic.")

    # Login to a GDK wallet session using the mnemonic.
    wallet.login_with_mnemonic(mnemonic)

    # We can now perform calls against the session, such as get balance for
    # the logged in Blockstream AMP Managed Assets account.
    balance = wallet.get_balance()
    print(f'\n{json.dumps(balance, indent=4)}')


    # Using a pin to encrypt the mnemonic and login
    # =============================================

    # You can also login using a pin. Setting the pin for the wallet returns
    # encrypted data that is saved to file. When you login with the pin, the
    # server will give you the key to decrypt the mnemonic which it uses to
    # login. If the pin is entered incorrectly 3 times the server will delete
    # the key and you must use the mnemonic to login.

    """
    # Before setting the pin, login with the wallet's mnemoic.
    wallet.login_with_mnemonic(mnemonic)
    # Then set the pin for the wallet, this saves encrypted data to file.
    pin = 123456
    # You only need to set the pin data once.
    pin_data = wallet.set_pin(mnemonic, pin)
    # After setting the pin you can then login using pin and do not have to
    # enter the mnemoic again. The pin is used to decrypt the local file.
    wallet.login_with_pin(pin)
    """


    # Two factor authorization
    # ========================

    # You can add Two Factor Authentication (2FA) to a wallet when you create
    # it or enable or disable 2FA at a later date.
    # Check the current 2FA status for the wallet.
    twofactor_status = wallet.get_current_2fa_status()
    print(f'\n{json.dumps(twofactor_status, indent=4)}')

    # The example below will enable 2FA on an existing wallet and uses email by
    # default, which you can amend if you want.
    """
    try:
        wallet.twofactor_auth_enabled(False)
    except RuntimeError as e:
        # Will error if 2FA is already enabled
        print(f'\nError: {e}\n')
    """


    # Getting notification data from GDK to obtain the last block height
    # ==================================================================

    # The fetch_block_height example shows how to handle notification events
    # from Green by processing the notifications queue.
    block_height = wallet.fetch_block_height()
    print(f'\nCurrent Liquid block height: {block_height}')


    # Getting a new address and understanding pointers
    # ================================================

    # The new address returned will be confidential, whereas GDK transactions
    # will show the unconfidential address. For this reason, use the address
    # 'pointer' to identify it in transactions.
    address_info = wallet.get_new_address()
    print(f'Address: {address_info["address"]}')
    print(f'Address pointer: {address_info["pointer"]}')

    # Each call creates a new address/pointer pair for the user.
    address_info = wallet.get_new_address()
    print(f'Address: {address_info["address"]}')
    print(f'Address pointer: {address_info["pointer"]}')


    # Getting transaction data from Green using GDK
    # =============================================

    # The example uses block height to restrict the transactions returned.
    # Set this to 0 to see all transactions.
    since_block_height = 0
    txs = wallet.get_wallet_transactions(since_block_height)
    for tx in txs:
        print(f'TRANSACTION ID      : {tx["txhash"]}')
        print(f'CONFIRMATION STATUS : {tx["confirmation_status"]}')
        print(f'BLOCK HEIGHT        : {tx["block_height"]}')
        print(f'TYPE                : {tx["type"]}')
        print(f'INPUT COUNT         : {len(tx["inputs"])}')
        print(f'OUTPUT COUNT        : {len(tx["outputs"])}\n')


    # Sending assets
    # ==============

    # Please be aware that AMP issued assets are issued with a precision
    # that affects how the number of sats sent are converted to the number
    # of units of the asset itself. Please refer to the examples under
    # 'precision' on the following page for more details and examples:
    # https://docs.blockstream.com/blockstream-amp/api-tutorial.html#issuing-an-asset
    # If the asset is registered with the Liquid Assets Registry you can
    # check the precision using the following link, or check with the
    # asset's issuer:
    # https://blockstream.info/liquid/assets
    amount_sat = 1
    asset_id = 'asset id here'
    address = 'destination address here'
    txid = wallet.send_to_address(amount_sat, asset_id, address)
    if txid:
        print(f'\nTransaction sent. Txid: {txid}')
    else:
        print(f'\nTransaction failed. See error logging.')


class wallet_utils:

    # To install GDK, download the GDK python wheel from:
    # https://github.com/Blockstream/gdk/releases
    # The 'cp' number refers to the python version you have.
    # To install GDK, pip install the .whl file:
    # pip install greenaddress-0.0.33-cp37-cp37m-linux_x86_64.whl
    # GDK reference documentation:
    # https://gdk.readthedocs.io/en/latest/


    def __init__(self):
        # 2of2_no_recovery is the account type used by Blockstream AMP.
        # Do not change this value!
        self.AMP_ACCOUNT_TYPE = '2of2_no_recovery'

        # AA stands for Authorized Assets.
        # You can change this if you like, e.g. 'AMP'.
        # Account type and name are used to retrieve the correct account and
        # should be unique per wallet so you can retrieve the right account
        # when you login.
        self.SUB_ACCOUNT_NAME = 'AA'

        # If you use a pin to login, the encrypted data will be saved and read
        # from this file:
        self.PIN_DATA_FILENAME = 'pin_data.json'

        # Initialize GDK.
        gdk.init({})
        self.mnemonic = None
        self.session = None
        self.sub_account_pointer = None
        self.gaid = None
        self.last_block_height = 0


    def create_new_wallet(self, create_with_2fa_enabled, mnemonic=None):
        # Create a new wallet with a Managed Assets account.
        # You can pass in a mnemonic generated outside GDK if you want, or have
        # GDK generate it for you by omitting it. 2FA is enabled if chosen and
        # can be enabled/disabled at any point.
        if not mnemonic:
            mnemonic = gdk.generate_mnemonic()
        self.session = gdk.Session({'name': 'liquid'})
        self.session.register_user({}, mnemonic).resolve()
        self.session.login({}, mnemonic).resolve()
        self.check_signed_in()
        self.session.create_subaccount({'name': self.SUB_ACCOUNT_NAME, 'type': self.AMP_ACCOUNT_TYPE}).resolve()
        if create_with_2fa_enabled:
            self.twofactor_auth_enabled(True)
        return mnemonic


    def login_with_mnemonic(self, mnemonic):
        self.mnemonic = mnemonic
        self.session = gdk.Session({'name': 'liquid'})
        # To see detailed debug info, create the session like this:
        # ({'name': 'liquid', 'log_level': 'debug'})
        self.session.login({}, self.mnemonic).resolve()
        self.check_signed_in()
        self.fetch_sub_account()


    def login_with_pin(self, pin):
        pin_data = open(self.PIN_DATA_FILENAME).read()
        self.session = gdk.Session({'name': 'liquid'})
        self.session.login_with_pin(str(pin), pin_data).resolve()
        self.check_signed_in()
        self.fetch_sub_account()


    def set_pin(self, mnemonic, pin):
        self.check_signed_in()
        pin_data = gdk.set_pin(self.session.session_obj, mnemonic, str(pin), str('device_id_1'))
        open(self.PIN_DATA_FILENAME, 'w').write(pin_data)
        return pin_data


    def get_balance(self):
        self.check_signed_in()
        return self.session.get_balance({'subaccount': self.sub_account_pointer, 'num_confs': 0}).resolve()


    def get_current_2fa_status(self):
        self.check_signed_in()
        return self.session.get_twofactor_config()


    def twofactor_auth_enabled(self, enabled):
        # We will use email but others are available ('sms', 'phone', 'gauth').
        # https://gdk.readthedocs.io/en/latest/gdk-json.html#twofactor-detail
        method = 'email'
        if enabled:
            print('\nRequesting email authentication is enabled for this account')
            email = input('\nPlease enter the email address that you will use to authenticate 2FA requests: ')
            details = {'confirmed':False,'enabled':True,'data':email}
        else:
            print('\nRequesting email authentication is disabled for this account')
            details = {'confirmed': True, 'enabled': False}
        # The following is an example of how to handle the GDK authentication
        # state machine as it progresses to completion.
        self._gdk_resolve(gdk.change_settings_twofactor(self.session.session_obj, method, json.dumps(details)))


    def _gdk_resolve(self, auth_handler):
        # Processes and handles the state of calls that need authentication.
        # The authentication process works as a state machine and may require
        # input to progress. This example only uses email as a authentication
        # method. If you would like to user other methods such as sms, phone,
        # gauth or a hardware device see:
        # https://github.com/Blockstream/green_cli/blob/842697b1c6e382487a2e00606c17d6637fe62e7b/green_cli/green.py#L75

        while True:
            status = gdk.auth_handler_get_status(auth_handler)
            status = json.loads(status)
            state = status['status']
            if state == 'error':
                raise RuntimeError(f'\nAn error occurred authenticating the call: {status}')
            if state == 'done':
                print('\nAuthentication succeeded or not required\n')
                return status['result']
            if state == 'request_code':
                authentication_factor = 'email'
                print(f'\nCode requested via {authentication_factor}.')
                gdk.auth_handler_request_code(auth_handler, authentication_factor)
            elif state == 'resolve_code':
                resolution = input('\nPlease enter the authentication code you received: ')
                gdk.auth_handler_resolve_code(auth_handler, resolution)
            elif state == 'call':
                gdk.auth_handler_call(auth_handler)


    def check_signed_in(self):
        if not self.session:
            raise Exception("You must call sign_in_with_mnemonic first")


    def fetch_sub_account(self):
        sub_accounts = self.session.get_subaccounts().resolve()
        for sub_account in sub_accounts['subaccounts']:
            if self.AMP_ACCOUNT_TYPE == sub_account['type'] and self.SUB_ACCOUNT_NAME == sub_account['name']:
                self.sub_account_pointer = sub_account['pointer']
        if not self.sub_account_pointer:
            raise Exception(f'Cannot find the sub account with name: "{self.SUB_ACCOUNT_NAME}" and type: "{self.AMP_ACCOUNT_TYPE}"')
        self.gaid = self.session.get_subaccount(self.sub_account_pointer).resolve()['receiving_id']
        # Notification queue always has the last block in after session login.
        self.fetch_block_height()


    def fetch_block_height(self):
        # New blocks are added to notifications as they are found so we need to
        # find the latest or, if there hasn't been one since we last checked,
        # use the value set during login in the session_login method.
        # The following provides an example of using GDK's notification queue.
        q = self.session.notifications
        max_block_height = self.last_block_height
        while not q.empty():
            notification = q.get()
            event = notification['event']
            if event == 'block':
                block_height = notification['block']['block_height']
                if block_height > max_block_height:
                    max_block_height = block_height
        self.last_block_height = max_block_height
        return self.last_block_height


    def is_valid_mnemonic(self, mnemonic):
        return gdk.validate_mnemonic(mnemonic)


    def get_new_address(self):
        self.check_signed_in()
        return self.session.get_receive_address({'subaccount': self.sub_account_pointer}).resolve()


    def get_wallet_transactions(self, since_block_height=0):
        # Get the current block height so we can update the returned
        # transaction data with its confirmation status.
        chain_block_height = self.fetch_block_height()
        # We'll use possible statuses of UNCONFIRMED, CONFIRMED, FINAL.
        confirmed_status = None
        depth_from_tip = 0
        all_txs = []
        index = 0
        count = 1
        while(True):
            transactions = self.session.get_transactions({'subaccount': self.sub_account_pointer,'first': index, 'count': count}).resolve()
            for transaction in transactions['transactions']:
                confirmation_status = 'UNCONFIRMED'
                block_height = transaction['block_height']
                if block_height <= since_block_height:
                    break
                # Unconfirmed txs will have a block_height of 0.
                if block_height > 0:
                    depth_from_tip = chain_block_height - block_height
                    # A transaction with 1 confirmation will have a depth of 0.
                    if depth_from_tip == 0:
                        confirmation_status = 'CONFIRMED'
                    if depth_from_tip > 0:
                        confirmation_status = 'FINAL'
                transaction['confirmation_status'] = confirmation_status
                all_txs.append(transaction)
            if len(transactions['transactions']) < count:
                break
            index = index + 1
        return all_txs


    def send_to_address(self, sat_amount, asset_id, destination_address):
        details = {
            'subaccount': self.sub_account_pointer,
            'addressees':[{'satoshi': sat_amount, 'address': destination_address, 'asset_tag': asset_id}]
        }

        try:
            details = self._gdk_resolve(gdk.create_transaction(self.session.session_obj, json.dumps(details)))
            details = self._gdk_resolve(gdk.sign_transaction(self.session.session_obj, json.dumps(details)))
            details = self._gdk_resolve(gdk.send_transaction(self.session.session_obj, json.dumps(details)))
            return details['txhash']
        except RuntimeError as e:
            print(f'\nError: {e}\n')


if __name__ == "__main__":
    main()