BLOCKCHAIN TECHNOLOGIES LAB
WEEK-3
NAME: V.KAVERI
ROLL NO: 23R25A6209
TOPIC: Simulating Bitcoin Transactions and Double-Spending
PROBLEM SATEMENT-1:
Design a Python script to create a transaction with sender, receiver, amount, and a digital signature.
How would you verify the transaction’s authenticity?
SOURCE CODE:
import json
import hashlib
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa
from cryptography.hazmat.backends import default_backend
# Transaction class
class Transaction:
def __init__(self, sender_public_key, receiver, amount, signature):
self.sender = sender_public_key
self.receiver = receiver
self.amount = amount
self.signature = signature
def to_dict(self):
return {
"sender": self.sender,
"receiver": self.receiver,
"amount": self.amount,
"signature": self.signature
}
def to_json(self):
return json.dumps(self.to_dict(), sort_keys=True)
# Block class
class Block:
def __init__(self, transactions, previous_hash):
self.transactions = transactions
self.previous_hash = previous_hash
self.hash = self.compute_hash()
def compute_hash(self):
tx_data = [tx.to_json() for tx in self.transactions]
block_string = json.dumps(tx_data, sort_keys=True) + self.previous_hash
return hashlib.sha256(block_string.encode()).hexdigest()
# Blockchain class
class Blockchain:
def __init__(self):
self.chain = []
genesis_block = Block([], "0" * 64)
self.chain.append(genesis_block)
def add_block(self, transactions):
previous_hash = self.chain[-1].hash
block = Block(transactions, previous_hash)
self.chain.append(block)
def is_chain_valid(self):
for i in range(1, len(self.chain)):
current = self.chain[i]
prev = self.chain[i - 1]
if current.previous_hash != prev.hash:
return False
if current.hash != current.compute_hash():
return False
return True
# Digital signature functions
def generate_keys():
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
public_key = private_key.public_key()
return private_key, public_key
def sign_transaction(private_key, message):
signature = private_key.sign(
message.encode(),
padding.PKCS1v15(),
hashes.SHA256()
)
return signature.hex()
def verify_signature(public_key_pem, message, signature_hex):
public_key = serialization.load_pem_public_key(
public_key_pem.encode(),
backend=default_backend()
)
try:
public_key.verify(
bytes.fromhex(signature_hex),
message.encode(),
padding.PKCS1v15(),
hashes.SHA256()
)
return True
except Exception:
return False
# Pretty print blockchain
def print_blockchain(blockchain):
for i, block in enumerate(blockchain.chain):
print(f"\nBlock {i}:" + (" (Genesis Block)" if i == 0 else ""))
print(f"Hash: {block.hash}")
print(f"Previous Hash: {block.previous_hash}")
tx_list = [tx.to_dict() for tx in block.transactions]
print("Transactions:")
print(json.dumps(tx_list, indent=2))
# Main test
def main():
# Generate keys for sender
sender_private_key, sender_public_key = generate_keys()
# Export public key to PEM format
sender_pem = sender_public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
).decode()
# Create blockchain
blockchain = Blockchain()
# Create 3 transactions
tx1_msg = f"{sender_pem}Bob100"
tx1_sig = sign_transaction(sender_private_key, tx1_msg)
tx1 = Transaction(sender_pem, "Bob", 100, tx1_sig)
tx2_msg = f"{sender_pem}Charlie50"
tx2_sig = sign_transaction(sender_private_key, tx2_msg)
tx2 = Transaction(sender_pem, "Charlie", 50, tx2_sig)
tx3_msg = f"{sender_pem}Dave200"
tx3_sig = sign_transaction(sender_private_key, tx3_msg)
tx3 = Transaction(sender_pem, "Dave", 200, tx3_sig)
# Verify one transaction
print("Is transaction valid?", verify_signature(sender_pem, tx1_msg, tx1_sig))
# Add transactions to blockchain
blockchain.add_block([tx1])
blockchain.add_block([tx2])
blockchain.add_block([tx3])
print("\n[+] Blockchain created with 3 transaction blocks.")
print_blockchain(blockchain)
print("\n[!] Blockchain valid?", blockchain.is_chain_valid())
# Tampering test
print("\n[!] Tampering Block 2 (Changing amount to 9999)...\n")
blockchain.chain[2].transactions[0].amount = 9999 # Tamper with Block 2
blockchain.chain[2].hash = blockchain.chain[2].compute_hash() # Recompute hash
print_blockchain(blockchain)
print("\n[!] Blockchain valid after tampering?", blockchain.is_chain_valid())
if __name__ == "__main__":
main()
OUTPUT:
Case1
Case2
Case3:
PROBLEM STATEMENT-2:
Implement a simple blockchain with a proof-of-work mechanism. What happens to the blockchain’s
integrity if two miners simultaneously attempt to add conflicting transactions? Ascertain the
transaction that remains valid.
SOURCE CODE:
import hashlib
import time
import copy
class Block:
def __init__(self, index, timestamp, data, previous_hash, difficulty=4):
self.index = index
self.timestamp = timestamp
self.data = data
self.previous_hash = previous_hash
self.nonce = 0
self.difficulty = difficulty
self.hash = self.mine_block()
def calculate_hash(self):
value = f"{self.index}{self.timestamp}{self.data}{self.previous_hash}{self.nonce}"
return hashlib.sha256(value.encode()).hexdigest()
def mine_block(self):
prefix = "0" * self.difficulty
while True:
hash_value = self.calculate_hash()
if hash_value.startswith(prefix):
return hash_value
self.nonce += 1
class Blockchain:
def __init__(self):
self.chain = [self.create_genesis_block()]
self.difficulty = 4
def create_genesis_block(self):
return Block(0, time.time(), "Genesis Block", "0")
def get_latest_block(self):
return self.chain[-1]
def add_block(self, data):
last_block = self.get_latest_block()
new_block = Block(len(self.chain), time.time(), data, last_block.hash, self.difficulty)
self.chain.append(new_block)
print(f"Block {new_block.index} mined: Nonce={new_block.nonce},
Hash={new_block.hash}")
def is_chain_valid(self, chain=None):
if chain is None:
chain = self.chain
for i in range(1, len(chain)):
current = chain[i]
previous = chain[i - 1]
if current.hash != current.calculate_hash():
return False
if current.previous_hash != previous.hash:
return False
return True
def print_chain(self):
for block in self.chain:
print(f"\nBlock {block.index}")
print(f"Timestamp: {block.timestamp}")
print(f"Previous Hash: {block.previous_hash}")
print(f"Hash: {block.hash}")
print(f"Nonce: {block.nonce}")
print(f"Data: {block.data}")
def simulate_fork_and_consensus(original_chain):
# Simulate Miner 1 extending the chain by 2 blocks
miner1_chain = copy.deepcopy(original_chain.chain)
miner1 = Blockchain()
miner1.chain = miner1_chain
print("\n--- Miner 1 mining blocks ---")
for i in range(4, 6):
block_data = f"Tx {i}: Miner 1"
last_block = miner1.get_latest_block()
new_block = Block(len(miner1.chain), time.time(), block_data, last_block.hash)
miner1.chain.append(new_block)
print(f"Miner 1 mined Block {new_block.index}: Nonce={new_block.nonce},
Hash={new_block.hash}")
# Simulate Miner 2 extending the chain by 1 block
miner2_chain = copy.deepcopy(original_chain.chain)
miner2 = Blockchain()
miner2.chain = miner2_chain
print("\n--- Miner 2 mining blocks ---")
block_data = "Tx 4: Miner 2"
last_block = miner2.get_latest_block()
new_block = Block(len(miner2.chain), time.time(), block_data, last_block.hash)
miner2.chain.append(new_block)
print(f"Miner 2 mined Block {new_block.index}: Nonce={new_block.nonce},
Hash={new_block.hash}")
# Consensus: choose the longest valid chain
if len(miner1.chain) > len(miner2.chain) and
Blockchain().is_chain_valid(miner1.chain):
print("\nMiner 1 has the longer chain. This is the longest valid chain now.")
return miner1.chain
else:
print("\nMiner 2 has the longer chain. This is the longest valid chain now.")
return miner2.chain
def tamper_chain(blockchain):
print("\n--- Tampering Block 2 ---")
blockchain.chain[2].data = "Tampered Tx: Bob -> Eve"
blockchain.chain[2].hash = blockchain.chain[2].calculate_hash()
def main():
blockchain = Blockchain()
print("Mining block 1...")
blockchain.add_block("Tx 1: Alice -> Bob")
print("Mining block 2...")
blockchain.add_block("Tx 2: Bob -> Charlie")
print("Mining block 3...")
blockchain.add_block("Tx 3: Charlie -> Dave")
print("\nOriginal Blockchain:")
blockchain.print_chain()
print("\nIs blockchain valid?", blockchain.is_chain_valid())
print("\n--- Simulating Fork ---")
new_chain = simulate_fork_and_consensus(blockchain)
blockchain.chain = new_chain
print("\nFinal Blockchain After Consensus:")
blockchain.print_chain()
print("\nIs final blockchain valid?", blockchain.is_chain_valid())
# Tampering case at end
tamper_chain(blockchain)
print("\nBlockchain After Tampering:")
blockchain.print_chain()
print("\nIs blockchain valid after tampering?", blockchain.is_chain_valid())
if __name__ == "__main__":
main()
OUTPUT:
Case1
Case2
Case3
\
PROBLEM STATEMENT-3:
Simulate a double-spend attack by attempting to spend the same Bitcoin twice in two different
transactions. Propose a solution to detect and prevent this attack.
SOURCE CODE:
import hashlib
import time
class Transaction:
def __init__(self, tx_id, sender, recipient, amount, timestamp=None):
self.tx_id = tx_id
self.sender = sender
self.recipient = recipient
self.amount = amount
self.timestamp = timestamp if timestamp else time.time()
def __str__(self):
return f"Tx ID: {self.tx_id} | Sender: {self.sender} | Recipient: {self.recipient} |
Amount: {self.amount} | Timestamp: {self.timestamp}"
class Block:
def __init__(self, index, transactions, previous_hash):
self.index = index
self.timestamp = time.time()
self.transactions = transactions
self.previous_hash = previous_hash
self.nonce, self.hash = self.mine_block()
def compute_hash(self):
tx_data = ''.join([str(tx.__dict__) for tx in self.transactions])
block_string = str(self.index) + str(self.timestamp) + tx_data + self.previous_hash +
str(self.nonce)
return hashlib.sha256(block_string.encode()).hexdigest()
def mine_block(self):
self.nonce = 0
computed_hash = self.compute_hash()
while not computed_hash.startswith('0000'):
self.nonce += 1
computed_hash = self.compute_hash()
return self.nonce, computed_hash
def __str__(self):
block_info = f"Block {self.index}\n"
block_info += f"Timestamp: {self.timestamp}\n"
block_info += f"Previous Hash: {self.previous_hash}\n"
block_info += f"Hash: {self.hash}\n"
block_info += f"Nonce: {self.nonce}\n"
block_info += "Transactions:\n"
for tx in self.transactions:
block_info += f" {tx}\n"
return block_info
class Blockchain:
def __init__(self):
self.chain = []
self.used_tx_signatures = set()
self.create_genesis_block()
def create_genesis_block(self):
print("STEP 1: Create Blockchain and Genesis Block\nCreating Blockchain and
Genesis Block...\n")
genesis_tx = Transaction("genesis_tx", "SYSTEM", "Alice", 100)
self.add_block_to_chain([genesis_tx])
def add_block_to_chain(self, transactions):
valid_txs = []
for tx in transactions:
tx_signature = (tx.tx_id, tx.timestamp)
if tx_signature in self.used_tx_signatures:
print(f"[ERROR] Double Spend Detected! Transaction '{tx.tx_id}' with
timestamp '{tx.timestamp}' already used.")
return False
valid_txs.append(tx)
if not valid_txs:
print("No valid transactions to add. Block rejected.\n")
return False
previous_hash = self.chain[-1].hash if self.chain else "0"
block_index = len(self.chain)
new_block = Block(block_index, valid_txs, previous_hash)
self.chain.append(new_block)
for tx in valid_txs:
self.used_tx_signatures.add((tx.tx_id, tx.timestamp))
print(new_block)
return True
def is_chain_valid(self):
for i in range(1, len(self.chain)):
current = self.chain[i]
previous = self.chain[i-1]
if current.previous_hash != previous.hash:
return False
if current.hash != current.compute_hash():
return False
return True
def simulate():
blockchain = Blockchain()
print("\nSTEP 2: Add Block 1\n")
tx1 = Transaction("tx1", "Alice", "Bob", 50)
blockchain.add_block_to_chain([tx1])
print("\nSTEP 3: Add Block 2\n")
tx2 = Transaction("tx2", "Bob", "Charlie", 25)
blockchain.add_block_to_chain([tx2])
print("\nSTEP 4: Simulate Double Spend Attempt\n")
print("Block 3")
print("Attempting transaction:")
print(f" Tx ID: {tx1.tx_id} | Sender: Alice | Recipient: Eve | Amount: 50 | Timestamp:
{tx1.timestamp}\n")
# Reusing same tx_id and timestamp intentionally to simulate double spending
tx3 = Transaction("tx1", "Alice", "Eve", 50, timestamp=tx1.timestamp)
blockchain.add_block_to_chain([tx3])
print("\nSTEP 5: Validate Blockchain")
print("Blockchain valid:", blockchain.is_chain_valid())
simulate()
OUTPUT:
PROBLEM STATEMENT-4:
Modify the proof-of-work difficulty in your simulation. How does increasing or decreasing the
difficulty affect the time taken to mine a block?
SOURCE CODE:
import hashlib
import time
def is_even_hex_char(char):
return char.lower() in ['0', '2', '4', '6', '8', 'a', 'c', 'e']
def proof_of_work(difficulty):
prefix = '0' * difficulty
nonce = 0
start_time = time.time()
while True:
text = f"block-data-{nonce}"
hash_result = hashlib.sha256(text.encode()).hexdigest()
if hash_result.startswith(prefix) and is_even_hex_char(hash_result[-1]):
end_time = time.time()
return nonce, hash_result, end_time - start_time
nonce += 1
def simulate_modified_pow():
print("Proof of Work Simulation")
print("------------------------")
print("Rule: Hash must start with N zeros AND end with an even hex digit (0, 2, 4, 6, 8, a, c, e).\
n")
for difficulty in range(2, 6):
nonce, hash_result, elapsed = proof_of_work(difficulty)
print(f"Difficulty: {difficulty} (Target hash starts with {difficulty} zeros)")
print(f" Nonce found: {nonce}")
print(f" Hash: {hash_result}")
print(f" Time taken: {elapsed:.4f} seconds\n")
print("Conclusion: Increasing difficulty -> Exponentially more time to mine.")
simulate_modified_pow()
OUTPUT:
