Создать новый алгоритм консенсуса на **Python** на базе **Proof-of-Work (PoW)**
Создать новый алгоритм консенсуса на **Python** на базе **Proof-of-Work (PoW)** и временных проверок возможно, особенно для обучения и тестирования блокчейн-системы. Python удобен для быстрой прототипизации, однако для продакшена рекомендуется C++ из-за производительности.
### **Цель**:
Напишем **упрощённый блокчейн** с новым алгоритмом консенсуса:
- **PoW**: Найти хеш, который удовлетворяет определённому условию.
- **Time Lock**: Блок может быть сгенерирован только спустя **10 секунд** после предыдущего.
```python
import hashlib
import time
import json
from typing import List
class Block:
def __init__(self, index: int, previous_hash: str, timestamp: int, data: str, difficulty: int):
self.index = index # Номер блока
self.previous_hash = previous_hash # Хеш предыдущего блока
self.timestamp = timestamp # Время создания блока
self.data = data # Данные блока
self.difficulty = difficulty # Сложность майнинга
self.nonce = 0 # Счётчик для PoW
self.hash = self.calculate_hash()
def calculate_hash(self) -> str:
"""Вычисление хеша блока."""
block_string = json.dumps({
"index": self.index,
"previous_hash": self.previous_hash,
"timestamp": self.timestamp,
"data": self.data,
"difficulty": self.difficulty,
"nonce": self.nonce
}, sort_keys=True)
return hashlib.sha256(block_string.encode()).hexdigest()
def mine_block(self):
"""Майнинг блока с учётом сложности."""
target = '0' * self.difficulty
print(f"Майнинг блока {self.index} с сложностью {self.difficulty}...")
start_time = time.time()
while not self.hash.startswith(target):
self.nonce += 1
self.hash = self.calculate_hash()
end_time = time.time()
print(f"Блок {self.index} успешно намайнен! Хеш: {self.hash} | Время: {end_time - start_time:.2f} сек")
class Blockchain:
def __init__(self):
self.chain: List[Block] = []
self.difficulty = 4 # Начальная сложность
self.time_lock = 10 # Временной интервал между блоками
self.create_genesis_block()
def create_genesis_block(self):
"""Создание первого блока (генезис-блок)."""
genesis_block = Block(0, "0", int(time.time()), "Genesis Block", self.difficulty)
self.chain.append(genesis_block)
print(f"Генезис-блок создан: {genesis_block.hash}")
def get_latest_block(self) -> Block:
return self.chain[-1]
def add_block(self, data: str):
"""Добавление нового блока с учётом временной задержки."""
latest_block = self.get_latest_block()
while True:
current_time = int(time.time())
if current_time - latest_block.timestamp >= self.time_lock:
break # Ждём завершения Time Lock
time.sleep(1) # Проверяем каждую секунду
new_block = Block(
index=latest_block.index + 1,
previous_hash=latest_block.hash,
timestamp=int(time.time()),
data=data,
difficulty=self.difficulty
)
new_block.mine_block()
self.chain.append(new_block)
print(f"Блок {new_block.index} добавлен в цепочку\n")
def print_blockchain(self):
for block in self.chain:
print(f"Блок {block.index}:")
print(f" Хеш: {block.hash}")
print(f" Предыдущий хеш: {block.previous_hash}")
print(f" Данные: {block.data}")
print(f" Время создания: {block.timestamp}")
print(f" Nonce: {block.nonce}\n")
if __name__ == "__main__":
# Создаём блокчейн
blockchain = Blockchain()
# Добавляем несколько блоков
blockchain.add_block("Транзакции: Alice -> Bob: 5 BTC")
blockchain.add_block("Транзакции: Bob -> Carol: 2 BTC")
blockchain.add_block("Транзакции: Carol -> Dave: 1 BTC")
# Выводим всю цепочку блоков
blockchain.print_blockchain()
```
1. **PoW**:
- Для майнинга блока требуется найти хеш, который начинается с **n** нулей (`difficulty`).
- Поиск хеша реализован с помощью цикла и увеличения значения **nonce**.
2. **Time Lock**:
- Перед добавлением нового блока система проверяет, прошло ли **10 секунд** с момента создания предыдущего блока.
3. **Блокчейн**:
- Каждый блок хранит:
- **Хеш предыдущего блока** (цепочка блоков).
- **Timestamp** (время создания блока).
- **Сложность** (количество нулей в хеше).
- **Nonce** (число для майнинга).
4. **Майнинг**:
- При создании нового блока выполняется **майнинг**, который занимает время, зависящее от сложности.
```text
Генезис-блок создан: 0000a3f7baf19b123...
Майнинг блока 1 с сложностью 4...
Блок 1 успешно намайнен! Хеш: 0000f81e1c3b9a... | Время: 2.45 сек
Блок 1 добавлен в цепочку
Майнинг блока 2 с сложностью 4...
Блок 2 успешно намайнен! Хеш: 0000831d4e45cd... | Время: 2.78 сек
Блок 2 добавлен в цепочку
Блок 0:
Хеш: 0000a3f7baf19b123...
Предыдущий хеш: 0
Данные: Genesis Block
Время создания: 1714305738
Nonce: 2356
Блок 1:
Хеш: 0000f81e1c3b9a...
Предыдущий хеш: 0000a3f7baf19b123...
Данные: Транзакции: Alice -> Bob: 5 BTC
Время создания: 1714305748
Nonce: 5678
```
1. **Добавить адаптивную сложность**: Сложность (`difficulty`) может увеличиваться в зависимости от времени, затраченного на майнинг.
2. **Оптимизация алгоритма**: Улучшить производительность хеширования.
3. **Добавить транзакции и пулы**: Реализовать реальные транзакции и систему их подтверждения.
4. **Реализовать P2P-сеть**: Соединить узлы для распределённого консенсуса.
Этот код идеально подходит для **прототипа** или **обучения блокчейну** на Python. 🚀
Давайте поэтапно доработаем блокчейн:
1. **Добавим адаптивную сложность**: Сложность будет корректироваться на основе времени, затраченного на майнинг блоков.
2. **Оптимизируем хеширование**: Улучшим производительность за счёт минимизации повторных вычислений.
3. **Добавим транзакции и пулы**: Введём систему транзакций, мемпул и их подтверждение.
4. **Реализуем P2P-сеть**: Добавим базовое соединение между узлами с помощью Python и библиотеки `socket`.
### **1. Адаптивная сложность**
Добавим функцию, которая будет корректировать сложность на основе времени между блоками.
```python
class Blockchain:
DIFFICULTY_ADJUSTMENT_INTERVAL = 5 # Через сколько блоков пересчитывается сложность
TARGET_BLOCK_TIME = 10 # Целевое время для майнинга блока (в секундах)
def __init__(self):
self.chain: List[Block] = []
self.difficulty = 4
self.time_lock = 10
self.create_genesis_block()
def adjust_difficulty(self):
"""Адаптивная сложность на основе времени между блоками."""
if len(self.chain) % self.DIFFICULTY_ADJUSTMENT_INTERVAL != 0:
return # Пересчитываем сложность только каждые N блоков
start_time = self.chain[-self.DIFFICULTY_ADJUSTMENT_INTERVAL].timestamp
end_time = self.chain[-1].timestamp
actual_time = end_time - start_time
if actual_time < self.TARGET_BLOCK_TIME * self.DIFFICULTY_ADJUSTMENT_INTERVAL:
self.difficulty += 1 # Увеличиваем сложность
else:
self.difficulty = max(1, self.difficulty - 1) # Снижаем сложность, но минимум 1
print(f"Сложность скорректирована: {self.difficulty}")
```
### **2. Оптимизация хеширования**
Чтобы минимизировать повторные вычисления, создадим базовую строку блока **без nonce**, а затем будем добавлять `nonce` и считать хеш.
```python
class Block:
def __init__(self, index, previous_hash, timestamp, data, difficulty):
self.index = index
self.previous_hash = previous_hash
self.timestamp = timestamp
self.data = data
self.difficulty = difficulty
self.nonce = 0
self.base_string = json.dumps({
"index": self.index,
"previous_hash": self.previous_hash,
"timestamp": self.timestamp,
"data": self.data,
"difficulty": self.difficulty
}, sort_keys=True)
def calculate_hash(self, nonce: int) -> str:
"""Оптимизированное вычисление хеша."""
return hashlib.sha256(f"{self.base_string}{nonce}".encode()).hexdigest()
def mine_block(self):
"""Майнинг блока с оптимизацией."""
target = '0' * self.difficulty
print(f"Майнинг блока {self.index}...")
while True:
self.hash = self.calculate_hash(self.nonce)
if self.hash.startswith(target):
break
self.nonce += 1
print(f"Блок {self.index} намайнен: {self.hash}")
```
### **3. Транзакции и мемпул**
Добавим транзакции и пул неподтверждённых транзакций (**mempool**).
```python
class Transaction:
def __init__(self, sender: str, receiver: str, amount: float):
self.sender = sender
self.receiver = receiver
self.amount = amount
self.timestamp = int(time.time())
def to_dict(self):
return {"sender": self.sender, "receiver": self.receiver, "amount": self.amount, "timestamp": self.timestamp}
class Blockchain:
def __init__(self):
self.chain: List[Block] = []
self.mempool: List[Transaction] = [] # Пул неподтверждённых транзакций
self.difficulty = 4
self.time_lock = 10
self.create_genesis_block()
def add_transaction(self, sender: str, receiver: str, amount: float):
"""Добавление транзакции в mempool."""
transaction = Transaction(sender, receiver, amount)
self.mempool.append(transaction)
print(f"Транзакция добавлена: {sender} -> {receiver} ({amount} BTC)")
def mine_pending_transactions(self):
"""Майнинг блока с транзакциями из mempool."""
if not self.mempool:
print("Нет транзакций для майнинга.")
return
latest_block = self.get_latest_block()
data = [tx.to_dict() for tx in self.mempool]
new_block = Block(
index=latest_block.index + 1,
previous_hash=latest_block.hash,
timestamp=int(time.time()),
data=json.dumps(data),
difficulty=self.difficulty
)
new_block.mine_block()
self.chain.append(new_block)
self.mempool.clear()
print(f"Блок {new_block.index} добавлен в цепочку с транзакциями.")
```
### **4. Реализация P2P-сети**
Добавим базовый **P2P-сервер** с помощью библиотеки `socket`.
```python
import socket
import threading
class Node:
def __init__(self, host="127.0.0.1", port=5000):
self.host = host
self.port = port
self.peers = [] # Список узлов-соседей
def start_server(self):
"""Запуск сервера для прослушивания соединений."""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind((self.host, self.port))
server.listen(5)
print(f"Сервер запущен на {self.host}:{self.port}")
while True:
conn, addr = server.accept()
print(f"Новое соединение: {addr}")
threading.Thread(target=self.handle_client, args=(conn,)).start()
def handle_client(self, conn):
"""Обработка сообщений от клиента."""
try:
data = conn.recv(1024).decode()
print(f"Получено сообщение: {data}")
conn.send("Сообщение получено".encode())
finally:
conn.close()
def connect_to_peer(self, peer_host, peer_port):
"""Подключение к соседнему узлу."""
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect((peer_host, peer_port))
client.send("Привет, соседний узел!".encode())
response = client.recv(1024).decode()
print(f"Ответ от узла: {response}")
client.close()
# Пример использования
if __name__ == "__main__":
node = Node()
threading.Thread(target=node.start_server).start()
time.sleep(1) # Ждём запуска сервера
node.connect_to_peer("127.0.0.1", 5000)
```
1. **Адаптивная сложность**: Изменяется в зависимости от времени майнинга.
2. **Оптимизация хеширования**: Убрали лишние пересчёты.
3. **Транзакции и пул**: Реализован **mempool** и блоки с транзакциями.
4. **P2P-сеть**: Узлы могут обмениваться сообщениями и синхронизировать блоки.
### **Следующие шаги**:
- Реализовать **синхронизацию блокчейна** между узлами.
- Добавить **подтверждение транзакций** через несколько узлов.
- Ввести **цифровые подписи** для транзакций.
Вот план по реализации:
1. **Синхронизация блокчейна между узлами**: Узлы будут обмениваться данными о блоках и транзакциях для поддержания согласованности цепочки.
2. **Подтверждение транзакций через несколько узлов**: Добавим систему валидации транзакций, где каждый узел проверяет транзакцию перед её добавлением в мемпул.
3. **Цифровые подписи для транзакций**: Реализуем подписи на основе криптографической пары ключей (ECDSA).
### **1. Синхронизация блокчейна между узлами**
#### Узлы будут отправлять и принимать текущий блокчейн, проверяя целостность цепочки.
```python
class Node:
def __init__(self, host="127.0.0.1", port=5000):
self.host = host
self.port = port
self.peers = [] # Список узлов-соседей
self.blockchain = Blockchain() # Локальный экземпляр блокчейна
def sync_blockchain(self):
"""Синхронизация блокчейна между узлами."""
for peer in self.peers:
try:
print(f"Синхронизация с узлом {peer}")
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(peer)
client.send("REQUEST_BLOCKCHAIN".encode())
data = client.recv(4096).decode()
other_chain = json.loads(data)
client.close()
# Проверяем, нужно ли заменить цепочку
if len(other_chain) > len(self.blockchain.chain):
print("Обновляем блокчейн локального узла...")
self.blockchain.replace_chain(other_chain)
except Exception as e:
print(f"Ошибка синхронизации с узлом {peer}: {e}")
def handle_client(self, conn):
"""Обработка сообщений от клиента."""
data = conn.recv(4096).decode()
if data == "REQUEST_BLOCKCHAIN":
# Отправляем текущий блокчейн
conn.send(json.dumps([block.to_dict() for block in self.blockchain.chain]).encode())
conn.close()
def add_peer(self, peer_host, peer_port):
"""Добавление нового узла."""
self.peers.append((peer_host, peer_port))
```
### **2. Подтверждение транзакций через несколько узлов**
#### Добавляем метод в узлы для проверки и подтверждения транзакций.
```python
class Node:
def broadcast_transaction(self, transaction: Transaction):
"""Рассылаем транзакцию всем узлам."""
for peer in self.peers:
try:
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(peer)
client.send(f"NEW_TRANSACTION {json.dumps(transaction.to_dict())}".encode())
client.close()
except Exception as e:
print(f"Ошибка при рассылке транзакции узлу {peer}: {e}")
def handle_client(self, conn):
"""Обработка сообщений от клиента."""
data = conn.recv(4096).decode()
if data.startswith("NEW_TRANSACTION"):
# Получили новую транзакцию, проверяем её
_, tx_data = data.split(" ", 1)
tx_dict = json.loads(tx_data)
transaction = Transaction(
sender=tx_dict['sender'],
receiver=tx_dict['receiver'],
amount=tx_dict['amount']
)
# Проверяем транзакцию
if self.blockchain.is_valid_transaction(transaction):
print("Транзакция подтверждена.")
self.blockchain.add_transaction(transaction.sender, transaction.receiver, transaction.amount)
else:
print("Неверная транзакция.")
conn.close()
def is_valid_transaction(self, transaction: Transaction) -> bool:
"""Проверяем, что транзакция валидна."""
# Здесь можно добавить дополнительные проверки, например, баланс отправителя
return True
```
### **3. Цифровые подписи для транзакций**
#### Используем библиотеку `ecdsa` для генерации ключей и подписей.
##### Установка:
```bash
pip install ecdsa
```
```python
from ecdsa import SigningKey, VerifyingKey, SECP256k1
class Wallet:
def __init__(self):
self.private_key = SigningKey.generate(curve=SECP256k1)
self.public_key = self.private_key.verifying_key
def sign_transaction(self, transaction_data: str) -> str:
"""Подписываем данные транзакции."""
return self.private_key.sign(transaction_data.encode()).hex()
@staticmethod
def verify_signature(public_key_hex: str, signature: str, transaction_data: str