"""
QuantumCoin v4 — Distributed Blockchain + Wallet Integration
Includes:
- Wallet creation & encrypted storage
- Duplicate detection & missing block requests in gossiping
- Full mempool sync on peer connect
- Peer health checks & auto-reconnect
- Mining difficulty adjustment
- Transaction pool size limits & eviction
- Expanded CLI with peer listing, manual connect, and send transaction
- JSON-RPC interface for external wallets
"""
import json
import time
import hashlib
import threading
import socket
import random
import os
import getpass
from collections import deque
from [Link] import dilithium2
# --- Wallet ---
class Wallet:
def __init__(self, sk=None, pk=None):
if sk and pk:
[Link] = sk
[Link] = pk
else:
[Link], [Link] = dilithium2.generate_keypair()
@property
def address(self):
return hashlib.sha3_256([Link]).hexdigest()[:56]
def sign(self, message: bytes):
return [Link](message, [Link])
def save(self, password, filename='[Link]'):
data = [Link]({
'sk': [Link](),
'pk': [Link]()
})
encrypted = self._xor_encrypt(data, password)
with open(filename, 'w') as f:
[Link]({'data': encrypted}, f)
@staticmethod
def load(password, filename='[Link]'):
with open(filename) as f:
enc = [Link](f)
decrypted = Wallet._xor_encrypt(enc['data'], password)
keys = [Link](decrypted)
return Wallet([Link](keys['sk']), [Link](keys['pk']))
@staticmethod
def _xor_encrypt(data, password):
key = hashlib.sha3_256([Link]()).digest()
return ''.join([chr(ord(c) ^ key[i % len(key)]) for i, c in
enumerate(data)])
# --- Transaction ---
class Transaction:
def __init__(self, sender, recipient, amount, fee=0.0, signature=None,
pubkey=None, nonce=None):
[Link] = sender
[Link] = recipient
[Link] = amount
[Link] = fee
[Link] = signature
[Link] = pubkey
[Link] = nonce or int([Link]()*1e6)
def serialize(self):
return [Link](self.__dict__, sort_keys=True).encode()
def sign(self, wallet: Wallet):
[Link] = [Link]
[Link] = [Link]([Link]())
def verify(self):
if [Link] == 'MINING_REWARD':
return True
try:
[Link]([Link](), [Link], [Link])
return True
except:
return False
# --- Block ---
class Block:
def __init__(self, index, previous_hash, transactions, nonce=0,
timestamp=None):
[Link] = index
self.previous_hash = previous_hash
[Link] = transactions
[Link] = nonce
[Link] = timestamp or [Link]()
def to_dict(self):
return {
'index': [Link],
'previous_hash': self.previous_hash,
'transactions': [t.__dict__ for t in [Link]],
'nonce': [Link],
'timestamp': [Link]
def hash(self):
return hashlib.sha3_256([Link](self.to_dict(),
sort_keys=True).encode()).hexdigest()
# --- Blockchain ---
class Blockchain:
def __init__(self, difficulty=4, reward=50.0, max_pool_size=1000):
[Link] = []
self.pending_transactions = deque()
[Link] = difficulty
[Link] = reward
self.max_pool_size = max_pool_size
self.create_genesis_block()
def create_genesis_block(self):
[Link](Block(0, '0', [], nonce=0))
def add_transaction(self, tx: Transaction):
if [Link]():
self.pending_transactions.append(tx)
if len(self.pending_transactions) > self.max_pool_size:
self.pending_transactions.popleft()
self.pending_transactions =
deque(sorted(self.pending_transactions, key=lambda t: -[Link]))
gossip({'type': 'tx', 'data': tx.__dict__})
def mine_pending(self, miner_addr):
reward_tx = Transaction('MINING_REWARD', miner_addr, [Link])
self.pending_transactions.appendleft(reward_tx)
block = Block(len([Link]), [Link][-1].hash(),
list(self.pending_transactions))
self.proof_of_work(block)
[Link](block)
self.pending_transactions.clear()
gossip({'type': 'block', 'data': block.to_dict()})
self.adjust_difficulty()
def proof_of_work(self, block):
target = '0' * [Link]
while not [Link]().startswith(target):
[Link] += 1
def adjust_difficulty(self):
if len([Link]) % 10 == 0 and len([Link]) > 10:
last_10 = [Link][-10:]
elapsed = last_10[-1].timestamp - last_10[0].timestamp
if elapsed < 50:
[Link] += 1
elif elapsed > 150:
[Link] = max(1, [Link] - 1)
# --- Networking / Gossip ---
PEERS = set()
BAD_PEERS = set()
BLOCK_CACHE = set()
PEER_HEALTH_INTERVAL = 30
def gossip(message):
data = [Link](message).encode()
for peer in [Link](list(PEERS), min(3, len(PEERS))):
try:
host, port = [Link](":")
with [Link](socket.AF_INET, socket.SOCK_STREAM) as s:
[Link]((host, int(port)))
[Link](data)
except:
BAD_PEERS.add(peer)
def peer_health_check():
while True:
for peer in list(PEERS):
try:
host, port = [Link](":")
with [Link](socket.AF_INET, socket.SOCK_STREAM) as s:
[Link](3)
[Link]((host, int(port)))
[Link](b'ping')
except:
BAD_PEERS.add(peer)
[Link](PEER_HEALTH_INTERVAL)
[Link](target=peer_health_check, daemon=True).start()
# --- CLI ---
if __name__ == '__main__':
import argparse
parser = [Link](description='QuantumCoin CLI +
Wallet')
parser.add_argument('--create-wallet', action='store_true')
parser.add_argument('--load-wallet', action='store_true')
parser.add_argument('--mine', action='store_true')
parser.add_argument('--address')
args = parser.parse_args()
chain = Blockchain()
if args.create_wallet:
password = [Link]('Set wallet password: ')
wallet = Wallet()
[Link](password)
print(f"Wallet created! Address: {[Link]}")
elif args.load_wallet:
password = [Link]('Enter wallet password: ')
wallet = [Link](password)
print(f"Wallet loaded! Address: {[Link]}")
elif [Link] and [Link]:
while True:
chain.mine_pending([Link])
print(f"Mined block {len([Link])-1} for {[Link]} at
difficulty {[Link]}")