Introducción
Desde el 2010, la idea de una computación descentralizada, las alucinantes matemáticas, los problemas únicos y las nuevas formas encontradas para resolverlos, hicieron que fuese ansioso por saber todo sobre esta nueva tecnología y realmente profundizar en las malezas para saber qué es lo que hace que funcione.
Gracias a la facilidad de acceso de cosas de bajo nivel c de python sin sobrecarga, es el punto de entrada perfecto para cualquiera que quiera aprender lo básico y construir su propio blockchain.
Pre-requisitos
Para este tutorial, al menos, vas a necesitar python V3 y pip V3 para que trabaje como es esperado. Puedes encontrar las instrucciones de instalación para ambos aquí.
Una vez instalado, necesitas agarrar estas bibliotecas usando pip antes de continuar:
pip install cryptography
pip install asyncio
pip install threading
pip install socketserver
pip install random
Estructura
Estaremos trazando clases para la blockchain, las transacciones, los bloques, los mineros, los nodos y más importante, la red p2p para que todos puedan hablar entre sí.
Hay mucho trabajo por hacer para lograr que esta sea una cadena totalmente trabajable y segura, pero para el punto de aprendizaje te da mucho para construir ya que está cerca de estar en un sistema de producción de nivel, ¡sólo necesita que uses tu tiempo para aprenderlo, lo cual es la mejor parte!
Construyendo Nuestras Clases
Clase de Blockchain
Las clases de Blockchain rastrearán la recompensa base, el año base, reducir a la mitad la frecuencia en años y empezar la dificultad del número. Inicializaremos esto con su cadena inicial, un array de transacciones, el validador de nodos inicializado, el array de nodos blancos y el nodo minero actual.
Nuestra clase calculará las recompensas para el bloque, creará el bloque génesis en la inicialización, tendrá la habilidad de añadir transacciones, añadir bloques, consulta/actualización de balances, resolver conflictos y revisar la validez de los bloques.
class Blockchain:
BASE_REWARD = 50
BASE_YEAR = 2023
HALVING_FREQUENCY = 4
DIFFICULTY = 1
def __init__(self):
self.chain = [self.create_genesis_block()]
self.transactions = []
self.nodes = []
self.miner_node = Node(self.add_node())
def calculate_reward(self):
current_year = datetime.now().year
elapsed_years = current_year - self.BASE_YEAR
return self.BASE_REWARD / (2 ** (elapsed_years // self.HALVING_FREQUENCY))
def create_genesis_block(self):
return Block(0, time.time(), [], "0", 0)
def add_transaction(self, transaction):
self.transactions.append(transaction)
sender = transaction.sender
recipient = transaction.recipient
amount = transaction.amount
sender_balance = getattr(self.get_node_by_address(sender), 'balance', 0)
recipient_balance = getattr(self.get_node_by_address(recipient), 'balance', 0)
setattr(self.get_node_by_address(sender), 'balance', sender_balance - amount)
setattr(self.get_node_by_address(recipient), 'balance', recipient_balance + amount)
return len(self.chain) + 1
def new_transaction(self, transaction):
self.transactions.append(transaction)
return len(self.chain) - 1
def add_node(self, address=None):
if address is None:
node = Node(str(uuid.uuid4()))
self.miner_node = node
self.nodes.append(node)
self.update_balances() # Actualiza los balances luego de añadir un nuevo nodo
return node.address
else:
node = self.get_node_by_address(address)
if node is None:
node = Node(address)
self.miner_node = node
self.nodes.append(node)
self.update_balances() # Actualiza los balances luego de añadir un nuevo nodo
return node.address
else:
return node.address
def get_balances(self):
balances = {}
for block in self.chain:
for transaction in block.transactions:
sender = transaction.sender
recipient = transaction.recipient
amount = transaction.amount
balances[sender] = balances.get(sender, 0) - amount
balances[recipient] = balances.get(recipient, 0) + amount
# Actualiza los balances basado en las transacciones pendientes
for transaction in self.transactions:
sender = transaction.sender
recipient = transaction.recipient
amount = transaction.amount
balances[sender] = balances.get(sender, 0) - amount
balances[recipient] = balances.get(recipient, 0) + amount
return balances
def update_balances(self):
balances = self.get_balances()
for node in self.nodes:
address = node.address
balance = balances.get(address, 0)
setattr(node, 'balance', balance)
def mine_block(self):
try:
last_block = self.chain[-1]
index = last_block.index + 1
timestamp = time.time()
transactions = self.transactions.copy()
reward = self.calculate_reward()
transactions.append(Transaction(self.miner_node.address, self.miner_node.address, reward, "reward", miner_node=self.miner_node))
previous_hash = last_block.hash_block()
max_attempts = 1000
nonce = 0
while nonce < max_attempts:
block = Block(index, timestamp, transactions, previous_hash, nonce)
block_hash = block.hash_block()
if block_hash[:self.DIFFICULTY] == '0' * self.DIFFICULTY and self.node_validator.validate_block(block, self.nodes):
self.chain.append(block)
self.transactions = []
return block
nonce += 1
return None
except Exception as e:
print("An error occurred while mining a new block:")
print(str(e))
traceback.print_exc()
return None
def get_node_by_address(self, address):
for node in self.nodes:
if node.address == address:
return node
return None
def resolve_conflicts(self):
longest_chain = None
max_length = len(self.chain)
for node in self.nodes:
response = requests.get(f'http://{node.address}/blocks')
if response.status_code == 200:
length = response.json()['length']
chain = response.json()['chain']
if length > max_length and self.is_valid(chain):
max_length = length
longest_chain = chain
if longest_chain:
self.chain = longest_chain
return True
return False
def is_valid(self, chain=None):
if chain is None:
chain = self.chain
for i in range(1, len(chain)):
current_block = chain[i]
previous_block = chain[i - 1]
if current_block.previous_hash != previous_block.hash_block():
return False
return True
def __str__(self):
return json.dumps([block.to_dict() for block in self.chain], indent=2)
El secreto de nuestro POW (prueba de trabajo, proof of work) es la dificultad y la función mine_block. A medida que los bloques son minados, la dificultad incrementa con el tiempo. Resolvemos cualquier conflicto en los estados de bloque entre nodos usando la regla de la cadena más larga y añadiendo una capa de validación extra en el método is_valid() para volver a verificar la integridad del bloque.
Clase del Bloque
Pondremos esto en el archivo blockchain.py ya que los bloques son containers simples y no requieren mucha lógica pesada. Los datos basados en el tiempo, transacciones, los hashes anteriores y siguientes, un nonce y verificación de firma es todo lo que necesitaremos.
Tenemos la función hash_block, la cual genera el hash SHA256 del diccionario de datos y lo imprime al bloque antes de añadirlo a la cadena. Añade esta clase sobre la clase de tu blockchain en el mismo archivo llamado blockchain.py
class Block:
def __init__(self, index, timestamp, transactions, previous_hash, nonce, validations=None, signature=None):
self.index = index
self.timestamp = timestamp
self.transactions = transactions
self.previous_hash = previous_hash
self.nonce = nonce
self.validations = validations if validations else []
self.signature = signature if signature else b''
def hash_block(self):
block_str = json.dumps(self.to_dict(), sort_keys=True)
digest = hashes.Hash(hashes.SHA256(), backend=default_backend())
digest.update(block_str.encode())
return digest.finalize().hex()
def to_dict(self):
return {
"index": self.index,
"timestamp": self.timestamp,
"transactions": [transaction.to_dict() for transaction in self.transactions],
"previous_hash": self.previous_hash,
"nonce": self.nonce,
"validations": self.validations,
"signature": self.signature.hex() if self.signature else None
}
Clase de Nodo
Estas clases manejan nuestra firma y verificación de las transacciones en la red. Crea un nuevo archivo llamado node.py y pégalo en el siguiente código, esto activará la propagación de nuestra transacción. Pero por ahora, necesitamos que las transacciones se transmitan.
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization, hashes
class Node:
def __init__(self, address=None):
self.address = address if address else str(uuid.uuid4())
self.private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
self.public_key = self.private_key.public_key()
def sign(self, message):
digest = hashes.Hash(hashes.SHA256(), backend=default_backend())
digest.update(message)
signature = self.private_key.sign(
digest.finalize(),
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return signature
def verify(self, message, signature):
digest = hashes.Hash(hashes.SHA256(), backend=default_backend())
digest.update(message)
try:
self.public_key.verify(
signature,
digest.finalize(),
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH,
),
hashes.SHA256(),
)
return True
except Exception:
return False
class NodeValidator:
def __init__(self):
self.confirmations_needed = 20
def validate_block(self, block, nodes):
confirmations = 0
for node in nodes:
if self.is_block_approved_by_node(block, node):
confirmations += 1
if confirmations >= self.confirmations_needed:
return True
return False
def is_block_approved_by_node(self, block, node):
try:
node.verify(block.hash_block().encode(), block.signature)
return True
except Exception:
return False
Clase de Transacción
Cada txn en la red será una representación de los datos de esta clase. Crea un nuevo archivo transaction.py y pega lo siguiente. Esta clase es el control funcional que otros métodos llamarán para validar la salud e integridad de la transacción.
class Transaction:
def __init__(self, sender, recipient, amount, data, signature=None, miner_node=None):
if sender is None or recipient is None or amount is None or data is None:
raise ValueError("Invalid transaction parameters")
self.sender = sender
self.recipient = recipient
self.amount = float(amount)
self.data = data
self.signature = signature if signature else b''
self.miner_node = miner_node if miner_node else None
def to_dict(self):
return {
"sender": self.sender,
"recipient": self.recipient,
"amount": self.amount,
"data": self.data,
"signature": self.signature.hex() if self.signature else None,
}
def sign(self):
sender_bytes = self.sender if isinstance(self.sender, bytes) else self.sender.encode()
recipient_bytes = self.recipient if isinstance(self.recipient, bytes) else self.recipient.encode()
amount_bytes = str(self.amount).encode()
data_bytes = self.data if isinstance(self.data, bytes) else self.data.encode()
message = sender_bytes + recipient_bytes + amount_bytes + data_bytes
self.signature = self.miner_node.sign(message)
def verify(self):
sender_bytes = self.sender if isinstance(self.sender, bytes) else self.sender.encode()
recipient_bytes = self.recipient if isinstance(self.recipient, bytes) else self.recipient.encode()
amount_bytes = str(self.amount).encode()
data_bytes = self.data if isinstance(self.data, bytes) else self.data.encode()
message = sender_bytes + recipient_bytes + amount_bytes + data_bytes
return self.miner_node.verify(message, self.signature)
Clase de P2P
La magia de toda esta cosa es la habilidad de cada nodo y minero en la red para comunicarse, siempre y cuando tengan el acceso TCP/IP. Crea el archivo p2p.py y pega el siguiente código para activar la red.
Esto necesitará ser más usable en un escenario de la red cruzada del mundo real, así que revisa el descubrimiento de pares (peer discovery) en la internet y los retos que necesitas sobrepasar para hacer que esto sea una clase usable en el mundo real.
import asyncio
import time
import threading
from socketserver import BaseRequestHandler, ThreadingTCPServer
from transaction import Transaction
class P2PRequestHandler(BaseRequestHandler):
MAX_REQUESTS_PER_MINUTE = 1000
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.requests = 0
self.reset_time = time.time()
def handle(self):
try:
self.requests += 1
current_time = time.time()
if current_time - self.reset_time > 60:
self.requests = 0
self.reset_time = current_time
if self.requests > self.MAX_REQUESTS_PER_MINUTE:
raise Exception("Too many requests")
request = self.request.recv(1024).decode()
response = asyncio.run(self.node.handle_request(request))
return response.encode()
except Exception as e:
error_message = f"An error occurred: {str(e)}"
traceback_str = traceback.format_exc()
print(error_message)
print(traceback_str)
return f"HTTP/1.1 500 Internal Server Error\r\n\r\n{error_message}\n{traceback_str}".encode()
class P2PNode:
def __init__(self, blockchain):
self.peers = []
self.server_address = ('', 5000)
self.running = False
self.blockchain = blockchain
self.server = None
async def handle_request(self, request):
try:
request_str = request.decode()
print("Received request:", request_str)
response = await self.process_request(request_str)
return response.encode()
except Exception as e:
print("An error occurred in the request handler:")
print(str(e))
traceback.print_exc()
return "HTTP/1.1 500 Internal Server Error\r\n\r\n".encode()
async def process_request(self, request):
method, *headers_and_body = request.split('\r\n\r\n')
headers = headers_and_body[0]
body = headers_and_body[1] if len(headers_and_body) > 1 else ''
path = headers.split(' ')[1]
if method == 'POST':
if path == '/transactions/new':
return await self.new_transaction(body)
elif path == '/mine':
return await self.mine()
elif path == '/blocks':
return await self.full_chain()
elif path == '/peers/new':
return await self.add_peer(body)
elif method == 'GET' and path == '/blocks':
return await self.full_chain()
return "HTTP/1.1 404 Not Found\r\n\r\n"
async def start(self):
try:
self.running = True
print("Server started.")
self.server = ThreadingTCPServer(self.server_address, P2PRequestHandler)
self.server.node = self
server_thread = threading.Thread(target=self.server.serve_forever)
server_thread.start()
except Exception as e:
print("An error occurred in the server:", str(e))
Main
Crea un archivo llamado main.py que usará esto para probar la red y sus funciones variadas.
import threading
import asyncio
import time
import random
from blockchain import Blockchain
from transaction import Transaction
from p2p import P2PNode
try:
blockchain = Blockchain()
print("Blockchain created")
# Comienza 25 nodos y financia cada uno con 100 monedas
nodes = []
miner_address = blockchain.add_node() # Add the miner's address
for _ in range(25):
address = blockchain.add_node()
nodes.append(address)
transaction = Transaction(miner_address, address, 100, "Initial funds", miner_node=blockchain.miner_node)
transaction.sign()
blockchain.new_transaction(transaction)
print("25 nodes created and funded.")
# Imprime los balances de todos los nodos
balances = blockchain.get_balances()
print("Balances after funding:")
for address, balance in balances.items():
print(f"{address}: {balance}")
async def start_server():
node = P2PNode(blockchain)
await node.start()
thread = threading.Thread(target=lambda: asyncio.run(start_server()))
thread.start()
time.sleep(1) # Espera a que los nodos sincronicen
# Cada nodo envía una transacción a tres destinatarios aleatorios
for sender in nodes:
recipient = random.choice(nodes)
if recipient != sender:
sender_address = sender
recipient_address = recipient
sender_balance = blockchain.get_balances().get(sender_address, 0)
print("Sender balance:", sender_balance)
if sender_balance >= 1: # Revisa si el emisor tiene suficientes fondos
print("Sender:", sender_address)
print("Recipient:", recipient_address)
transaction = Transaction(sender_address, recipient_address, 1, "Transaction", miner_node=blockchain.miner_node)
transaction.sign()
try:
blockchain.new_transaction(transaction)
except Exception as e:
print("An error occurred:", str(e))
traceback.print_exc()
else:
print("Sender does not have sufficient funds to send the transaction.")
# Comienza a minotaur hasta que tres bloques sean minados
for _ in range(3):
block = blockchain.mine_block()
if block:
print("New block mined successfully.")
print(f"Block hash: {block.hash_block()}")
else:
print("Failed to mine a new block.")
break
print("Three blocks mined successfully.")
# Imprime los balances de todos los nodos
balances = blockchain.get_balances()
print("Balances after all transactions:")
for address, balance in balances.items():
print(f"{address}: {balance}")
print("Printing the blockchain:")
print(blockchain)
# Comienza a minar hasta que la recompensa del minero sean 20 monedas
while blockchain.get_balances()[blockchain.miner_node.address] < 20:
block = blockchain.mine_block()
if block:
print("New block mined successfully.")
print(f"Block hash: {block.hash_block()}")
else:
print("Failed to mine a new block.")
break
print("Miner rewards reached 20 coins.")
# Envía 0.5 monedas a cada nodo desde el minero
for recipient in nodes:
transaction = Transaction(blockchain.miner_node.address, str(recipient), 0.5, "Reward", miner_node=blockchain.miner_node)
transaction.sign()
blockchain.new_transaction(transaction)
print("Reward transactions sent to nodes.")
time.sleep(1) # Espera que las transacciones sean procesadas
# Imprime balances de todos los nodos
balances = blockchain.get_balances()
print("Balances after all transactions:")
for address, balance in balances.items():
print(f"{address}: {balance}")
print("Printing the blockchain:")
print(blockchain)
except Exception as e:
print(f"An error occurred: {e}")
Probando
Carga una ventana terminal como un usuario administrador y escribe, luego pulsa regresar. ¡Esto hará que la cadena gire y te permitirá verlo en acción!
python main.py
Conclusión
Este es un punto inicial básico para una blockchain, aún le faltan capas extras de validación y seguridad y para que esté listo para la producción, ¡pero es un punto genial de comienzo para cualquiera que quiera aprender!
La fuente entera puede encontrarse aquí
Este artículo es una traducción de Robert McMenemy, hecha por Héctor Botero. Puedes encontrar el artículo original aquí.
Sería genial escucharte en nuestro Discord, puedes contarnos tus ideas, comentarios, sugerencias y dejarnos saber lo que necesitas.
Si prefieres puedes escribirnos a @web3dev_es en Twitter.
Discussion (0)