""" Blockchain-based Ticket Verification System Author: Starlight Agent Version: 1.0 """ import json import hashlib import datetime import base64 from typing import Dict, List, Optional, Any from dataclasses import dataclass, asdict from collections import defaultdict @dataclass class TicketRecord: """Ticket record for blockchain storage""" ticket_id: str event_id: str owner_id: str price: float timestamp: str signature: str status: str metadata: Dict[str, Any] class Block: """Blockchain block structure""" def __init__(self, index: int, timestamp: str, data: List[TicketRecord], previous_hash: str): self.index = index self.timestamp = timestamp self.data = data self.previous_hash = previous_hash self.nonce = 0 self.hash = self.calculate_hash() def calculate_hash(self) -> str: """Calculate block hash""" data_str = json.dumps([asdict(record) for record in self.data], sort_keys=True) block_string = f"{self.index}{self.timestamp}{data_str}{self.previous_hash}{self.nonce}" return hashlib.sha256(block_string.encode()).hexdigest() def mine_block(self, difficulty: int): """Mine block with proof of work""" target = "0" * difficulty while self.hash[:difficulty] != target: self.nonce += 1 self.hash = self.calculate_hash() class BlockchainTicketVerifier: """Blockchain-based ticket verification system""" def __init__(self): self.chain = [self.create_genesis_block()] self.difficulty = 2 self.pending_tickets = [] self.ticket_registry = {} self.revoked_tickets = set() def create_genesis_block(self) -> Block: """Create genesis block""" return Block(0, datetime.datetime.now().isoformat(), [], "0") def get_latest_block(self) -> Block: """Get latest block in chain""" return self.chain[-1] def create_ticket(self, event_id: str, owner_id: str, price: float, metadata: Dict[str, Any] = None) -> str: """Create new ticket and add to blockchain""" ticket_id = self._generate_ticket_id(event_id, owner_id) timestamp = datetime.datetime.now().isoformat() # Create digital signature signature_data = f"{ticket_id}{event_id}{owner_id}{price}{timestamp}" signature = hashlib.sha256(signature_data.encode()).hexdigest() ticket = TicketRecord( ticket_id=ticket_id, event_id=event_id, owner_id=owner_id, price=price, timestamp=timestamp, signature=signature, status="active", metadata=metadata or {} ) self.pending_tickets.append(ticket) self.ticket_registry[ticket_id] = ticket # Mine block if we have enough pending tickets if len(self.pending_tickets) >= 3: self.mine_pending_tickets() return ticket_id def mine_pending_tickets(self): """Mine pending tickets into new block""" if not self.pending_tickets: return new_block = Block( len(self.chain), datetime.datetime.now().isoformat(), self.pending_tickets, self.get_latest_block().hash ) new_block.mine_block(self.difficulty) self.chain.append(new_block) self.pending_tickets.clear() def verify_ticket(self, ticket_id: str, event_id: str, owner_id: str) -> Dict[str, Any]: """Verify ticket authenticity and ownership""" if ticket_id in self.revoked_tickets: return {"valid": False, "reason": "Ticket revoked"} ticket = self.ticket_registry.get(ticket_id) if not ticket: return {"valid": False, "reason": "Ticket not found"} if ticket.event_id != event_id: return {"valid": False, "reason": "Event mismatch"} if ticket.owner_id != owner_id and ticket.status != "transferred": return {"valid": False, "reason": "Ownership mismatch"} if ticket.status != "active": return {"valid": False, "reason": f"Ticket status: {ticket.status}"} # Verify signature signature_data = f"{ticket_id}{event_id}{ticket.owner_id}{ticket.price}{ticket.timestamp}" expected_signature = hashlib.sha256(signature_data.encode()).hexdigest() if ticket.signature != expected_signature: return {"valid": False, "reason": "Invalid signature"} # Verify blockchain integrity if not self.is_chain_valid(): return {"valid": False, "reason": "Blockchain compromised"} return { "valid": True, "ticket": asdict(ticket), "block_index": self._find_ticket_block(ticket_id), "verification_timestamp": datetime.datetime.now().isoformat() } def transfer_ticket(self, ticket_id: str, from_owner: str, to_owner: str) -> Dict[str, Any]: """Transfer ticket ownership""" ticket = self.ticket_registry.get(ticket_id) if not ticket: return {"success": False, "reason": "Ticket not found"} if ticket.owner_id != from_owner: return {"success": False, "reason": "Unauthorized transfer"} if ticket.status != "active": return {"success": False, "reason": f"Cannot transfer ticket in status: {ticket.status}"} # Create transfer record transfer_data = { "from_owner": from_owner, "to_owner": to_owner, "transfer_timestamp": datetime.datetime.now().isoformat() } # Update ticket ticket.owner_id = to_owner ticket.status = "transferred" ticket.metadata["transfer_history"] = ticket.metadata.get("transfer_history", []) ticket.metadata["transfer_history"].append(transfer_data) # Add to blockchain self.pending_tickets.append(ticket) self.mine_pending_tickets() return {"success": True, "new_owner": to_owner, "transfer_id": hashlib.sha256(json.dumps(transfer_data).encode()).hexdigest()} def revoke_ticket(self, ticket_id: str, reason: str) -> Dict[str, Any]: """Revoke ticket due to fraud or cancellation""" ticket = self.ticket_registry.get(ticket_id) if not ticket: return {"success": False, "reason": "Ticket not found"} ticket.status = "revoked" ticket.metadata["revocation_reason"] = reason ticket.metadata["revoked_timestamp"] = datetime.datetime.now().isoformat() self.revoked_tickets.add(ticket_id) # Add to blockchain self.pending_tickets.append(ticket) self.mine_pending_tickets() return {"success": True, "revoked_at": ticket.metadata["revoked_timestamp"]} def is_chain_valid(self) -> bool: """Validate entire blockchain""" for i in range(1, len(self.chain)): current_block = self.chain[i] previous_block = self.chain[i - 1] if current_block.hash != current_block.calculate_hash(): return False if current_block.previous_hash != previous_block.hash: return False return True def _generate_ticket_id(self, event_id: str, owner_id: str) -> str: """Generate unique ticket ID""" data = f"{event_id}{owner_id}{datetime.datetime.now().isoformat()}" return hashlib.sha256(data.encode()).hexdigest()[:16] def _find_ticket_block(self, ticket_id: str) -> Optional[int]: """Find block containing specific ticket""" for block in self.chain: for ticket in block.data: if ticket.ticket_id == ticket_id: return block.index return None def get_ticket_history(self, ticket_id: str) -> List[Dict[str, Any]]: """Get complete ticket transaction history""" history = [] for block in self.chain: for ticket in block.data: if ticket.ticket_id == ticket_id: history.append({ "block_index": block.index, "timestamp": block.timestamp, "ticket_status": ticket.status, "owner_id": ticket.owner_id, "metadata": ticket.metadata }) return history def test_blockchain_verifier(): """Test blockchain ticket verification system""" verifier = BlockchainTicketVerifier() # Create tickets ticket1 = verifier.create_ticket("event123", "user456", 50.0, {"seat": "A1"}) ticket2 = verifier.create_ticket("event123", "user789", 75.0, {"seat": "B2"}) # Mine pending tickets verifier.mine_pending_tickets() # Verify ticket verification = verifier.verify_ticket(ticket1, "event123", "user456") assert verification["valid"] == True # Test transfer transfer = verifier.transfer_ticket(ticket1, "user456", "user999") assert transfer["success"] == True # Verify transfer verification_after = verifier.verify_ticket(ticket1, "event123", "user999") assert verification_after["valid"] == True # Test revocation revoke = verifier.revoke_ticket(ticket2, "fraud_detected") assert revoke["success"] == True # Verify revoked ticket verify_revoked = verifier.verify_ticket(ticket2, "event123", "user789") assert verify_revoked["valid"] == False # Validate blockchain integrity assert verifier.is_chain_valid() == True print("✅ Blockchain ticket verifier working correctly") return verifier if __name__ == "__main__": verifier = test_blockchain_verifier() print(f"Blockchain with {len(verifier.chain)} blocks created successfully")