В этом уроке мы рассмотрим этапы создания блокчейна с нуля с использованием Python. Мы начнем с реализации основных структур данных и методов для хранения и управления блокчейном. Далее мы добавим в блокчейн дополнительные функции, такие как система проверки работоспособности и одноранговая сеть для распределенной связи. Наконец, мы создадим простой HTTP API для взаимодействия с блокчейном из внешних приложений. К концу этого руководства вы будете хорошо понимать, как работает блокчейн и как его создать на Python.
Этот урок будет разбит на три части:
- Настройка проекта и создание базовой структуры блокчейна
- Реализация основных функций блокчейна, включая добавление новых блоков, проверку целостности блокчейна и обработку конфликтов.
- Добавление дополнительных функций в блокчейн, таких как система проверки работоспособности и одноранговая сеть для распределенной связи.
Прежде чем мы начнем, важно иметь общее представление о том, как работает блокчейн. Блокчейн — это распределенная, децентрализованная, общедоступная книга, которая записывает транзакции на нескольких компьютерах. Каждый блок в цепочке содержит запись нескольких транзакций, и после добавления блока в цепочку его нельзя изменить.
Теперь давайте начнем!
1. Настройка проекта и создание базовой структуры блокчейна
Во-первых, давайте создадим новый файл Python и импортируем необходимые библиотеки:
import hashlib import json from time import time
Библиотека hashlib будет использоваться для создания хэшей для блоков, json будет использоваться для кодирования и декодирования блоков при их сохранении и загрузке из файла, а time будет использоваться для записи метки времени для каждого блока.
Далее давайте определим класс для блокчейна:
class Blockchain:
def __init__(self):
self.current_transactions = []
self.chain = []
self.nodes = set()
# Create the genesis block
self.new_block(previous_hash=1, proof=100)
def new_block(self, proof, previous_hash=None):
"""
Create a new Block in the Blockchain
:param proof: The proof given by the Proof of Work algorithm
:param previous_hash: Hash of previous Block
:return: New Block
"""
pass
def new_transaction(self, sender, recipient, amount):
"""
Creates a new transaction to go into the next mined Block
:param sender: Address of the Sender
:param recipient: Address of the Recipient
:param amount: Amount
:return: The index of the Block that will hold this transaction
"""
pass
@staticmethod
def hash(block):
"""
Creates a SHA-256 hash of a Block
:param block: Block
"""
pass
@property
def last_block(self):
"""
Returns the last Block in the chain
"""
pass
Класс Blockchain имеет несколько методов:
__init__(): инициализирует блок-цепочку генезис-блоком (первым блоком в цепочке). Генезисный блок имеет доказательство 100 и предыдущий хэш 1.new_block(): Создает новый блок и добавляет его в цепочку.new_transaction(): Добавляет новую транзакцию в список транзакций, которые будут включены в следующий блок.hash(): хеширует блок с помощью SHA-256.last_block(): возвращает последний блок в цепочке.
В списке current_transactions будут храниться транзакции, которые еще не были добавлены в блок, в списке chain будут храниться блоки в блокчейне, а в списке nodes будут храниться узлы в сети.
Теперь заполним метод new_block(), чтобы создать новый блок и добавить его в цепочку:
def new_block(self, proof, previous_hash=None):
"""
Create a new Block in the Blockchain
:param proof: The proof given by the Proof of Work algorithm
:param previous_hash: Hash of previous Block
:return: New Block
"""
block = {
'index': len(self.chain) + 1,
'timestamp': time(),
'transactions': self.current_transactions,
'proof': proof,
'previous_hash': previous_hash or self.hash(self.chain[-1]),
}
# Reset the current list of transactions
self.current_transactions = []
self.chain.append(block)
return block
Этот метод создает новый блок с заданным доказательством и предыдущим хешем и добавляет его в список chain. Он также сбрасывает список current_transactions для подготовки к следующему блоку.
Далее реализуем метод new_transaction() для добавления новой транзакции в список транзакций, которые будут включены в следующий блок:
def new_transaction(self, sender, recipient, amount):
"""
Creates a new transaction to go into the next mined Block
:param sender: Address of the Sender
:param recipient: Address of the Recipient
:param amount: Amount
:return: The index of the Block that will hold this transaction
"""
self.current_transactions.append({
'sender': sender,
'recipient': recipient,
'amount': amount,
})
return self.last_block['index'] + 1
Этот метод добавляет новую транзакцию в список current_transactions и возвращает индекс блока, который будет содержать эту транзакцию (следующий блок, который будет добыт).
Наконец, давайте реализуем методы hash() и last_block():
@staticmethod
def hash(block):
"""
Creates a SHA-256 hash of a Block
:param block: Block
"""
block_string = json.dumps(block, sort_keys=True).encode()
return hashlib.sha256(block_string).hexdigest()
@property
def last_block(self):
"""
Returns the last Block in the chain
"""
return self.chain[-1]
Метод hash() принимает блок в качестве входных данных и возвращает его хеш, используя SHA-256. Метод last_block() возвращает последний блок в цепочке.
После реализации этих методов у нас теперь есть базовая структура для нашей цепочки блоков. В следующей части мы реализуем основные функции блокчейна, включая добавление новых блоков, проверку целостности блокчейна и обработку конфликтов.
2. Реализация основных функций блокчейна
В этой части мы реализуем следующие методы:
proof_of_work(): простой алгоритм Proof of Work (PoW), который будет использоваться для создания новых блоков.valid_proof(): метод проверки доказательства блока.valid_chain(): Метод, позволяющий определить, является ли данная цепочка блоков допустимой.resolve_conflicts(): метод разрешения конфликтов между разными версиями блокчейна.
Начнем с реализации метода proof_of_work():
def proof_of_work(self, last_proof):
"""
Simple Proof of Work Algorithm:
- Find a number p' such that hash(pp') contains leading 4 zeroes, where p is the previous p'
- p is the previous proof, and p' is the new proof
:param last_proof: <int>
:return: <int>
"""
proof = 0
while self.valid_proof(last_proof, proof) is False:
proof += 1
return proof
Этот метод принимает последнее доказательство в качестве входных данных и пытается найти новое доказательство, увеличивая значение счетчика до тех пор, пока не будет найдено действительное доказательство (доказательство с четырьмя ведущими нулями в хэше).
Далее реализуем метод valid_proof():
@staticmethod
def valid_proof(last_proof, proof):
"""
Validates the Proof
:param last_proof: <int> Previous Proof
:param proof: <int> Current Proof
:return: <bool> True if correct, False if not.
"""
guess = f'{last_proof}{proof}'.encode()
guess_hash = hashlib.sha256(guess).hexdigest()
return guess_hash[:4] == "0000"
Этот метод принимает последнее доказательство и текущее доказательство в качестве входных данных и проверяет, имеет ли хэш конкатенации двух четыре начальных нуля.
Далее реализуем метод valid_chain():
def valid_chain(self, chain):
"""
Determine if a given blockchain is valid
:param chain: <list> A blockchain
:return: <bool> True if valid, False if not
"""
last_block = chain[0]
current_index = 1
while current_index < len(chain):
block = chain[current_index]
print(f'{last_block}')
print(f'{block}')
print("\n-----------\n")
# Check that the hash of the block is correct
if block['previous_hash'] != self.hash(last_block):
return False
# Check that the Proof of Work is correct
if not self.valid_proof(last_block['proof'], block['proof']):
return False
last_block = block
current_index += 1
return True
Этот метод выполняет итерацию по цепочке блоков и проверяет правильность хэша каждого блока и действительность Proof of Work. Если какая-либо из этих проверок не пройдена, возвращается False. Если проверки проходят для всех блоков в цепочке, возвращается True.
Наконец, давайте реализуем метод resolve_conflicts():
def resolve_conflicts(self):
"""
This is our consensus algorithm, it resolves conflicts
by replacing our chain with the longest one in the network.
:return: <bool> True if our chain was replaced, False if not
"""
neighbours = self.nodes
new_chain = None
# We're only looking for chains longer than ours
max_length = len(self.chain)
# Grab and verify the chains from all the nodes in our network
for node in neighbours:
response = requests.get(f'http://{node}/chain')
if response.status_code == 200:
length = response.json()['length']
chain = response.json()['chain']
# Check if the length is longer and the chain is valid
if length > max_length and self.valid_chain(chain):
max_length = length
new_chain = chain
# Replace our chain if we discovered a new, valid chain longer than ours
if new_chain:
self.chain = new_chain
return True
return False
Этот метод проверяет все узлы в сети на наличие более длинной допустимой цепочки и, если находит, заменяет текущую цепочку новой.
Благодаря внедрению этих методов у нас теперь есть полностью функциональный блокчейн. В следующей части мы добавим в блокчейн дополнительные функции, такие как система проверки работоспособности и одноранговая сеть для распределенной связи.
3. Добавление дополнительных функций в блокчейн
В этой заключительной части мы добавим в блокчейн следующие функции:
- Система проверки работоспособности для защиты блокчейна и предотвращения ее несанкционированного доступа.
- Одноранговая сеть для распределенной связи между узлами в сети.
- API для взаимодействия с блокчейном из внешних приложений.
Во-первых, давайте немного изменим систему проверки работоспособности, которую мы создали ранее. В настоящее время метод proof_of_work() просто увеличивает счетчик до тех пор, пока не будет найдено действительное доказательство, но это не очень безопасно. Вместо этого мы можем использовать более безопасный алгоритм проверки работоспособности, такой как алгоритм майнинга биткойнов SHA-256.
Чтобы реализовать это, мы можем изменить метод proof_of_work() следующим образом:
def proof_of_work(self, last_block):
"""
Bitcoin's SHA-256 mining algorithm:
- Find a number p' such that hash(pp') contains leading 4 zeroes, where p is the previous p'
- p is the previous proof, and p' is the new proof
:param last_block: <dict> last Block
:return: <int>
"""
last_proof = last_block['proof']
last_hash = self.hash(last_block)
proof = 0
while self.valid_proof(last_proof, proof, last_hash) is False:
proof += 1
return proof
Этот метод теперь принимает последний блок в качестве входных данных и пытается найти новое доказательство, увеличивая счетчик до тех пор, пока не будет найдено действительное доказательство (доказательство, имеющее четыре начальных нуля в хэше). Также мы передаем хеш последнего блока в метод valid_proof() в качестве дополнительного параметра.
Далее давайте изменим метод valid_proof(), чтобы использовать новый алгоритм проверки работоспособности:
@staticmethod
def valid_proof(last_proof, proof, last_hash):
"""
Validates the Proof
:param last_proof: <int> Previous Proof
:param proof: <int> Current Proof
:param last_hash: <str> The hash of the Previous Block
:return: <bool> True if correct, False if not.
"""
guess = f'{last_proof}{proof}{last_hash}'.encode()
guess_hash = hashlib.sha256(guess).hexdigest()
return guess_hash[:4] == "0000"
Этот метод теперь принимает хэш последнего блока в качестве дополнительного параметра и использует его для проверки доказательства.
Теперь, когда у нас есть безопасная система проверки работоспособности, давайте реализуем одноранговую сеть для распределенной связи между узлами в сети. Для этого мы будем использовать библиотеку Flask для создания простого HTTP API для блокчейна.
Во-первых, давайте импортируем библиотеку Flask и создадим новый объект Blockchain:
from flask import Flask
app = Flask(__name__)
# Generate a globally unique address for this node
node_identifier = str(uuid4()).replace('-', '')
# Instantiate the Blockchain
blockchain = Blockchain()
Далее давайте определим следующие маршруты для API:
/mine: Маршрут для добычи нового блока./transactions/new: Маршрут для создания новой транзакции./chain: Маршрут для возврата полной цепочки блоков.
Вот код этих маршрутов:
@app.route('/mine', methods=['GET'])
def mine():
# We run the proof of work algorithm to get the next proof...
last_block = blockchain.last_block
proof = blockchain.proof_of_work(last_block)
# We must receive a reward for finding the proof.
# The sender is "0" to signify that this node has mined a new coin.
blockchain.new_transaction(
sender="0",
recipient=node_identifier,
amount=1,
)
# Forge the new Block by adding it to the chain
previous_hash = blockchain.hash(last_block)
block = blockchain.new_block(proof, previous_hash)
response = {
'message': "New Block Forged",
'index': block['index'],
'transactions': block['transactions'],
'proof': block['proof'],
'previous_hash': block['previous_hash'],
}
return jsonify(response), 200
@app.route('/transactions/new', methods=['POST'])
def new_transaction():
values = request.get_json()
# Check that the required fields are in the POST'ed data
required = ['sender', 'recipient', 'amount']
if not all(k in values for k in required):
return 'Missing values', 400
# Create a new Transaction
index = blockchain.new_transaction(values['sender'], values['recipient'], values['amount'])
response = {'message': f'Transaction will be added to Block {index}'}
return jsonify(response), 201
@app.route('/chain', methods=['GET'])
def full_chain():
response = {
'chain': blockchain.chain,
'length': len(blockchain.chain),
}
return jsonify(response), 200
Маршрут mine() добывает новый блок, запуская алгоритм проверки работоспособности и добавляя новую транзакцию для майнера (с вознаграждением в 1 монету). Маршрут new_transaction() создает новую транзакцию и добавляет ее в список транзакций, которые будут включены в следующий блок. Маршрут full_chain() возвращает полную цепочку блоков.
После определения этих маршрутов мы можем запустить сервер и взаимодействовать с блокчейном через API. Чтобы запустить сервер, добавьте в конец файла следующий код:
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
Это запустит сервер на локальном компьютере, прослушивая порт 5000. Теперь вы можете взаимодействовать с блокчейном, отправляя HTTP-запросы на сервер.
Наконец, давайте добавим поддержку одноранговой сети, разрешив узлам регистрироваться на сервере и транслировать транзакции и блоки другим узлам в сети.
Для этого нам потребуется изменить маршруты /mine и /transactions/new, чтобы транслировать новый блок или транзакцию в сеть. Также нам потребуется добавить новый маршрут для регистрации новых узлов в сети.
Вот код этих дополнительных маршрутов:
@app.route('/nodes/register', methods=['POST'])
def register_nodes():
values = request.get_json()
nodes = values.get('nodes')
if nodes is None:
return "Error: Please supply a valid list of nodes", 400
for node in nodes:
blockchain.register_node(node)
response = {
'message': 'New nodes have been added',
'total_nodes': list(blockchain.nodes),
}
return jsonify(response), 201
@app.route('/nodes/resolve', methods=['GET'])
def consensus():
replaced = blockchain.resolve_conflicts()
if replaced:
response = {
'message': 'Our chain was replaced',
'new_chain': blockchain.chain
}
else:
response = {
'message': 'Our chain is authoritative',
'chain': blockchain.chain
}
return jsonify(response), 200
Маршрут register_nodes() позволяет узлам регистрироваться на сервере и добавлять их в список узлов в сети. Маршрут consensus() реализует алгоритм консенсуса и разрешает любые конфликты, заменяя текущую цепочку самой длинной действительной цепочкой в сети.
С добавлением этих дополнительных функций у нас теперь есть полнофункциональная система блокчейна на Python. Теперь вы можете использовать API для взаимодействия с блокчейном, добычи новых блоков, создания новых транзакций и присоединения к одноранговой сети.
Вы можете использовать следующие методы HTTP для взаимодействия с API:
GET /mine: Создайте новый блок и добавьте его в блокчейн.POST /transactions/new: Создайте новую транзакцию и добавьте ее в список транзакций, которые будут включены в следующий блок.GET /chain: Вернуть полную цепочку блоков.POST /nodes/register: Зарегистрируйте новые узлы на сервере.GET /nodes/resolve: Запустите алгоритм консенсуса, чтобы разрешить конфликты и убедиться, что текущая цепочка является самой длинной допустимой цепочкой в сети.
Вот пример того, как вы можете использовать API для создания новой транзакции:
import requests
# Create a new transaction
new_transaction = {
'sender': 'Alice',
'recipient': 'Bob',
'amount': 5,
}
# Send the transaction to the server
response = requests.post('http://localhost:5000/transactions/new', json=new_transaction)
print(response.json())
Это отправит запрос POST на маршрут /transactions/new с новой транзакцией в качестве тела запроса и распечатает ответ с сервера.
Я надеюсь, что это руководство было полезно для понимания того, как создать полнофункциональную систему блокчейна на Python. Если у вас есть какие-либо вопросы или вам нужны дополнительные разъяснения, пожалуйста, не стесняйтесь спрашивать.
