Site icon Arnau Dunjó Workspace

Desenvolupament d’una aplicació blockchain desde 0 amb Python

Imatge Blockchain

Python és un llenguatge de programació fàcil d’entendre i el fan servir milions de programadors, per això el vaig triar per a aquest exemple. El codi d’aquest exemple es pot descarregar de la meva GitHub.

¿Qué es “blockchain”?

Blockchain és una forma d’emmagatzemar dades digitals. Les dades poden ser literalment qualsevol cosa. Per Bitcoin, són les transaccions (registres de transferències de Bitcoin d’un compte a una altra), però fins i tot poden ser arxius. Les dades s’emmagatzemen en forma de blocs, que estan vinculats (o encadenats) entre si mitjançant hashes criptogràfics, d’aquí el nom “blockchain”.

Tota la “màgia” rau en la forma en què aquestes dades s’emmagatzemen i s’agreguen a la cadena de blocs. Una cadena de blocs és essencialment una llista vinculada que conté dades ordenade, amb algunes restriccions:

Aquestes característiques aporten beneficis com:

Anem a implementar- ho utilitzabt un enfocament de baix a dalt. Comencem per definir l’estructura de les dades que emmagatzemarem en la cadena de blocs. Una publicació és un missatge que publica qualsevol usuari a la nostra aplicació. Cada publicació constarà de tres elements essencials:

Emmagatzemar transaccions en blocs

Anem a emmagatzemar dades a la nostra cadena de blocs en un format de sobres conegut: JSON. L’estructura bàsica de transacció que farem servir serà aquesta:

{ 
"author": "some_author_name", 
"content": "Some thoughts that author wants to share", 
"timestamp": "The time at which the content was created" 
}

Les transaccions s’empaqueten en blocs. Un bloc pot contenir una o diverses transaccions de manera que la cadena és una llista de blocs (baules) concatenats. A causa de que pot haver-hi diversos blocs, cada bloc ha de tenir un ID únic:

class Block:
    def __init__(self, index, transactions, timestamp):
        """
        Constructor for the `Block` class.
        :param index: Unique ID of the block.
        :param transactions: List of transactions.
        :param timestamp: Time of generation of the block.
        """
        self.index = index 
        self.transactions = transactions 
        self.timestamp = timestamp

2. Afegint “empremtes digitals” als blocs

Unes de les principals característiques / avantatges dels blocs de la tecnologia “blockchain” és que un bloc no pot ser manipulat. Per aconseguir-ho utilitzarem funcions hash criptogràfiques.

Com que no és l’objectiu d’aquest post explicar el funcionament de les funcions “hash”, el podríem resumir molt ràpidament dient que una opció “hash” és una funció que a partir d’unes dades de qualsevol mida, produeix una cadena alfanumèrica de grandària fixa (un “hash”). Les característiques d’una funció hash ideal són:

En conseqüència, és pràcticament impossible endevinar les dades d’entrada donat el hash. (L’única manera és provar totes les combinacions d’entrada possibles).

Emmagatzemarem el hash del bloc en un camp dins del nostre objecte Block, i actuarà com una empremta digital (o signatura) de les dades contingudes en ell:

from hashlib import sha256
import json

def compute_hash(block):
    """
    Returns the hash of the block instance by first converting it
    into JSON string.
    """
    block_string = json.dumps(self.__dict__, sort_keys=True)
    return sha256(block_string.encode()).hexdigest()

3. Encadenament de blocs

Bé, ara configurarem els blocs. Com dèiem anteriorment, la cadena de blocs és una sèrie de blocs consecutius. Podriem emmagatzemar tots els blocs a la llista de Python (l’equivalent a una matriu). Però això no seria suficient, perquè què passa si algú reemplaça intencionadament un bloc vell amb un bloc nou a la col·lecció? Crea un nou bloc amb transaccions alterades, calcula el hash i el reemplaça per qualsevol bloc més antic.

Necessitem una forma d’assegurar-nos que qualsevol canvi en els blocs anteriors invalidi tota la cadena. La forma que té Bitcoin de fer això és crear dependència entre els blocs, encadenant amb el hash del bloc immediatament anterior a ells. Per encadenar, anem a incloure el hash del bloc anterior en el bloc actual, emmagatzemant-lo en un nou camp anomenat, per exemple previous_hash.

Bé, si cada bloc està vinculat al bloc anterior a través del camp previous_hash, què passa amb el primer bloc? Aquest bloc es diu bloc gènesi i es pot generar manualment o mitjançant alguna lògica. Afegim el camp previous_hash a la classe Block e implementem l’estructura inicial de la nostra classe Blockchain.

from hashlib import sha256
import json
import time


class Block:
    def__init__(self, index, transactions, timestamp, previous_hash):
        """
        Constructor for the `Block` class.
        :param index:         Unique ID of the block.
        :param transactions:  List of transactions
        :param timestamp:     Time of generation of the block.
        :param previous_hash: Hash of the previous block in the chain which this block is part of.                                        
        """
        self.index = index
        self.transactions = transactions
        self.timestamp = timestamp
        self.previous_hash = previous_hash # Adding the previous hash field

    def compute_hash(self):
        """
        Returns the hash of the block instance by first converting it
        into JSON string.
        """
        block_string = json.dumps(self.__dict__, sort_keys=True) # The string equivalent also considers the previous_hash field now
        return sha256(block_string.encode()).hexdigest()

class Blockchain:

    def __init__(self):
        """
        Constructor for the `Blockchain` class.
        """
        self.chain = []
        self.create_genesis_block()

    def create_genesis_block(self):
        """
        A function to generate genesis block and appends it to
        the chain. The block has index 0, previous_hash as 0, and
        a valid hash.
        """
        genesis_block = Block(0, [], time.time(), "0")
        genesis_block.hash = genesis_block.compute_hash()
        self.chain.append(genesis_block)

    @property
    def last_block(self):
        """
        A quick pythonic way to retrieve the most recent block in the chain. Note that
        the chain will always consist of at least one block (i.e., genesis block)
        """
        return self.chain[-1]

Ara, si canvia el contingut de qualsevol dels blocs anteriors:

Al final, s’invalida tota la cadena que segueix a el bloc reemplaçat i l’única manera de solucionar-ho és tornar a calcular tota la cadena.

4. Implementar un algoritme de proba de treball

No obstant això, segueix existint la possibilitat que canviem el bloc anterior i els hashs de tots els blocs següents ja que es poden tornar a calcular amb força facilitat i crear una cadena de blocs vàlida diferent. Per evitar això, podem aprofitar l’asimetria en els esforços de les funcions hash per fer que la tasca de calcular el hash sigui difícil i aleatòria. Així és com fem això: en lloc d’acceptar qualsevol hash per al bloc, li afegim la restricció que el nostre hash ha de començar amb “n zeros a l’esquerra”, on n pot ser qualsevol nombre enter positiu.

Sabem que a menys que canviem les dades del bloc, el hash no canviarà i, per descomptat, no volem canviar les dades existents. ¿O sigui que, que fem? Senzill! Afegirem algunes dades ficticies que podem canviar. Introduïm un nou camp en el nostre bloc anomenat nonce. Un nonce és un nombre que podem seguir canviant fins que obtinguem un hash que satisfaci la nostra restricció. El nonce que satisfà la restricció serveix com a prova que s’ha realitzat algun càlcul. Aquesta tècnica és una versió simplificada de l’algorisme Hashcash utilitzat en Bitcoin. El nombre de zeros especificat en la restricció determina la dificultat del nostre algoritme de prova de treball (com més gran sigui el nombre de zeros, més difícil serà desxifrar el nonce).

A més, a causa de l’asimetria, la prova de treball és difícil de calcular però molt fàcil de verificar una vegada que esbrina el nonce (només ha d’executar la funció hash novament):

class Blockchain:
    # difficulty of PoW algorithm
    difficulty = 2

    """
    Previous code contd..
    """

    def proof_of_work(self, block):
        """
        Function that tries different values of the nonce to get a hash
        that satisfies our difficulty criteria.
        """
        block.nonce = 0

        computed_hash = block.compute_hash()
        while not computed_hash.startswith('0' * Blockchain.difficulty):
            block.nonce += 1
            computed_hash = block.compute_hash()

        return computed_hash

Cal tenir en compte que no existeix una lògica específica per descobrir el nonce ràpidament; només es pot mitjançant força bruta. L’única manera d’accelerar el càlcul és utilitzar xips que estiguin especialment dissenyats per calcular la funció hash en el mínim nombre d’instruccions de CPU.

5. Afegir blocs a la cadena

Per afegir un bloc a la cadena, primer haurem de verificar que:

Vegem el codi per afegir blocs a la cadena:

class Blockchain:
    """
    Previous code contd..
    """

    def add_block(self, block, proof):
        """
        A function that adds the block to the chain after verification.
        Verification includes:
        * Checking if the proof is valid.
        * The previous_hash referred in the block and the hash of a latest block
          in the chain match.
        """
        previous_hash = self.last_block.hash

        if previous_hash != block.previous_hash:
            return False

        if not Blockchain.is_valid_proof(block, proof):
            return False

        block.hash = proof
        self.chain.append(block)
        return True

    def is_valid_proof(self, block, block_hash):
        """
        Check if block_hash is valid hash of block and satisfies
        the difficulty criteria.
        """
        return (block_hash.startswith('0' * Blockchain.difficulty) and
                block_hash == block.compute_hash())

Les transaccions s’emmagatzemaran inicialment com un grup de transaccions no confirmades. El procés de posar les transaccions no confirmades en un bloc i calcular la prova de treball es coneix com “minat de blocs”. Una vegada que es determina el nonce que satisfà les nostres restriccions, podem dir que s’ha minat un bloc i es pot posar en la cadena de blocs.

En la majoria de les moneda digital (inclòs Bitcoin), els miners poden rebre alguna moneda digital com a recompensa per gastar el seu poder de còmput per calcular una prova de treball. Així és com es veu la nostra funció de mineria:

class Blockchain:

    def __init__(self):
        self.unconfirmed_transactions = [] # data yet to get into blockchain
        self.chain = []
        self.create_genesis_block()

    """
    Previous code contd...
    """

    def add_new_transaction(self, transaction):
        self.unconfirmed_transactions.append(transaction)

    def mine(self):
        """
        This function serves as an interface to add the pending
        transactions to the blockchain by adding them to the block
        and figuring out proof of work.
        """
        if not self.unconfirmed_transactions:
            return False

        last_block = self.last_block

        new_block = Block(index=last_block.index + 1,
                          transactions=self.unconfirmed_transactions,
                          timestamp=time.time(),
                          previous_hash=last_block.hash)

        proof = self.proof_of_work(new_block)
        self.add_block(new_block, proof)
        self.unconfirmed_transactions = []
        return new_block.index

Recorda que teniu el codi sencer en el meu GitHub

6. Crear interfícies

Bé, ara és el moment de crear interfícies perquè el nostre node blockchain interactuï amb l’aplicació que construirem. Farem servir un micro-framework de Python popular anomenat Flask per crear una API REST que interactua i invoca diverses operacions en el nostre node blockchain. Si has treballat amb algun framework web anteriorment, el següent codi no t’hauria de ser difícil de seguir.

from flask import Flask, request
import requests

# Initialize flask application
app =  Flask(__name__)

# Initialize a blockchain object.
blockchain = Blockchain()

Necessitem un “endpoint” perquè la nostra aplicació enviï una nova transacció. Això serà utilitzat per la nostra aplicació per afegir noves dades (transaccions) a la cadena de blocs:

# Flask's way of declaring end-points
@app.route('/new_transaction', methods=['POST'])
def new_transaction():
    tx_data = request.get_json()
    required_fields = ["author", "content"]

    for field in required_fields:
        if not tx_data.get(field):
            return "Invalid transaction data", 404

    tx_data["timestamp"] = time.time()

    blockchain.add_new_transaction(tx_data)

    return "Success", 201

Aquí hi ha un “endpoint” per tornar la copia de la cadena.

@app.route('/chain', methods=['GET'])
def get_chain():
    chain_data = []
    for block in blockchain.chain:
        chain_data.append(block.__dict__)
    return json.dumps({"length": len(chain_data),
                       "chain": chain_data})

Aquí es sol·licita a el node que extregui les transaccions no confirmades (si n’hi ha). El farem servir per iniciar una ordre per minar des de la nostra pròpia aplicació:

@app.route('/mine', methods=['GET'])
def mine_unconfirmed_transactions():
    result = blockchain.mine()
    if not result:
        return "No transactions to mine"
    return "Block #{} is mined.".format(result)

@app.route('/pending_tx')
def get_pending_tx():
    return json.dumps(blockchain.unconfirmed_transactions)

7. Establir consens i descentralització

Fins a aquest punt, la cadena de blocs que hem implementat està destinada a executar-se en una sola computadora. Tot i que estem vinculant blocs amb hashs i aplicant la prova de restricció de treball, encara no podem confiar en una sola entitat (en el nostre cas, una sola màquina). Necessitem que les dades es distribueixin, necessitem múltiples nodes que mantinguin la cadena de blocs. Per tant, per realitzar la transició d’un sol node a una xarxa d’igual a igual, primer creem un mecanisme per permetre que un nou node conegui a altres parells de la xarxa:

# Contains the host addresses of other participating members of the network
peers = set()

# Endpoint to add new peers to the network
@app.route('/register_node', methods=['POST'])
def register_new_peers():
    # The host address to the peer node 
    node_address = request.get_json()["node_address"]
    if not node_address:
        return "Invalid data", 400

    # Add the node to the peer list
    peers.add(node_address)

    # Return the blockchain to the newly registered node so that it can sync
    return get_chain()


@app.route('/register_with', methods=['POST'])
def register_with_existing_node():
    """
    Internally calls the `register_node` endpoint to
    register current node with the remote node specified in the
    request, and sync the blockchain as well with the remote node.
    """
    node_address = request.get_json()["node_address"]
    if not node_address:
        return "Invalid data", 400

    data = {"node_address": request.host_url}
    headers = {'Content-Type': "application/json"}

    # Make a request to register with remote node and obtain information
    response = requests.post(node_address + "/register_node",
                             data=json.dumps(data), headers=headers)

    if response.status_code == 200:
        global blockchain
        global peers
        # update chain and the peers
        chain_dump = response.json()['chain']
        blockchain = create_chain_from_dump(chain_dump)
        peers.update(response.json()['peers'])
        return "Registration successful", 200
    else:
        # if something goes wrong, pass it on to the API response
        return response.content, response.status_code


def create_chain_from_dump(chain_dump):
    blockchain = Blockchain()
    for idx, block_data in enumerate(chain_dump):
        block = Block(block_data["index"],
                      block_data["transactions"],
                      block_data["timestamp"],
                      block_data["previous_hash"])
        proof = block_data['hash']
        if idx > 0:
            added = blockchain.add_block(block, proof)
            if not added:
                raise Exception("The chain dump is tampered!!")
        else:  # the block is a genesis block, no verification needed
            blockchain.chain.append(block)
    return blockchain

Un nou node que participa a la xarxa pot invocar el mètode register_with_existing_node per registrar-se amb els nodes existents a la xarxa. Això ajudarà amb el següent:

De totes maneres, hi ha un problema amb diversos nodes atès que la còpia de cadenes d’alguns nodes pot diferir (per exemple per latència de la xarxa). En aquest cas, els nodes han d’acordar quina de les versions de la cadena és la vàlida (hem d’aconseguir un consens).

Un algoritme de consens podria ser establir que la cadena vàlida és la més llarga.

class Blockchain
    """
    previous code continued...
    """
    def check_chain_validity(cls, chain):
        """
        A helper method to check if the entire blockchain is valid.            
        """
        result = True
        previous_hash = "0"

        # Iterate through every block
        for block in chain:
            block_hash = block.hash
            # remove the hash field to recompute the hash again
            # using `compute_hash` method.
            delattr(block, "hash")

            if not cls.is_valid_proof(block, block.hash) or \
                    previous_hash != block.previous_hash:
                result = False
                break

            block.hash, previous_hash = block_hash, block_hash

        return result

def consensus():
    """
    Our simple consensus algorithm. If a longer valid chain is
    found, our chain is replaced with it.
    """
    global blockchain

    longest_chain = None
    current_len = len(blockchain.chain)

    for node in peers:
        response = requests.get('{}/chain'.format(node))
        length = response.json()['length']
        chain = response.json()['chain']
        if length > current_len and blockchain.check_chain_validity(chain):
              # Longer valid chain found!
            current_len = length
            longest_chain = chain

    if longest_chain:
        blockchain = longest_chain
        return True

    return False

A continuació, ens cal desenvolupar un mètode perquè qualsevol node anunciï a la xarxa que ha extret un bloc perquè tothom pugui actualitzar el seu blockchain. Els altres nodes poden verificar la prova de treball i afegir el bloc minat a les seves respectives cadenes:

# endpoint to add a block mined by someone else to
# the node's chain. The node first verifies the block
# and then adds it to the chain.
@app.route('/add_block', methods=['POST'])
def verify_and_add_block():
    block_data = request.get_json()
    block = Block(block_data["index"],
                  block_data["transactions"],
                  block_data["timestamp"],
                  block_data["previous_hash"])

    proof = block_data['hash']
    added = blockchain.add_block(block, proof)

    if not added:
        return "The block was discarded by the node", 400

    return "Block added to the chain", 201


def announce_new_block(block):
    """
    A function to announce to the network once a block has been mined.
    Other blocks can simply verify the proof of work and add it to their
    respective chains.
    """
    for peer in peers:
        url = "{}add_block".format(peer)
        requests.post(url, data=json.dumps(block.__dict__, sort_keys=True))

El mètode announce_new_block s’ha de cridar després que el node extregui cada bloc perquè els parells puguin afegir-lo a les seves cadenes.

8. Crear l’aplicació

Ara és el moment de començar a treballar en la interfície de la nostra aplicació.

L’aplicació necesita connectar-se a un node a la xarxa blockchain per obtenir i enviar les dades (pot haver-hi diversos nodes).

import datetime
import json

import requests
from flask import render_template, redirect, request

from app import app

# Node in the blockchain network that our application will communicate with
# to fetch and add data.
CONNECTED_NODE_ADDRESS = "http://127.0.0.1:8000"

posts = []

La funció fetch_posts obté les dades del punt final / cadena del node, analitza les dades i les emmagatzema en local.

def fetch_posts():
    """
    Function to fetch the chain from a blockchain node, parse the
    data, and store it locally.
    """
    get_chain_address = "{}/chain".format(CONNECTED_NODE_ADDRESS)
    response = requests.get(get_chain_address)
    if response.status_code == 200:
        content = []
        chain = json.loads(response.content)
        for block in chain["chain"]:
            for tx in block["transactions"]:
                tx["index"] = block["index"]
                tx["hash"] = block["previous_hash"]
                content.append(tx)

        global posts
        posts = sorted(content,
                       key=lambda k: k['timestamp'],
                       reverse=True)

L’aplicació té un formulari HTML per a la identificació d’usuari i després realitza una sol·licitud POST a un node connectat per afegir la transacció a el grup de transaccions no confirmades. Després, la xarxa extreu la transacció i finalment es recupera una vegada que actualitzem la nostra pàgina web:

@app.route('/submit', methods=['POST'])
def submit_textarea():
    """
    Endpoint to create a new transaction via our application
    """
    post_content = request.form["content"]
    author = request.form["author"]

    post_object = {
        'author': author,
        'content': post_content,
    }

    # Submit a transaction
    new_tx_address = "{}/new_transaction".format(CONNECTED_NODE_ADDRESS)

    requests.post(new_tx_address,
                  json=post_object,
                  headers={'Content-type': 'application/json'})

    # Return to the homepage
    return redirect('/')

Conclusió

Amb això hauríem de ser capaços de crear una cadena de blocs des de 0 i crear una aplicació simple que permeti als usuaris compartir informació en la cadena de blocs. Òbviament no és tan sofisticada / completa com Bitcoin o Etherum però espero que serveixi per entendre el funcionament bàsic i respondre gran part de les preguntes que poden sorgir a perfils tècnics sobre el funcionament intern d’aquestes.

Font: https://developer.ibm.com/technologies/blockchain/tutorials/develop-a-blockchain-application-from-scratch-in-python/

Exit mobile version