개발 환경 세팅 (3분 완성)
파이썬으로 블록체인을 직접 구현해보면 블록체인의 핵심 원리를 쉽게 이해할 수 있어요. 복잡한 프레임워크 없이 순수 파이썬만으로 시작해볼게요.
# 필요 라이브러리 설치
pip install hashlib datetime json flask
필요 환경:
- Python 3.8 이상
- 코드 에디터 (VS Code 추천)
- 터미널 또는 명령 프롬프트
블록체인 핵심 구조 구현
1. 블록 클래스 만들기
블록체인의 기본 단위인 블록을 먼저 구현해요. 각 블록은 데이터, 타임스탬프, 이전 블록의 해시값을 포함해요.
import hashlib
import json
from datetime import datetime
class Block:
def __init__(self, index, transactions, timestamp, previous_hash, nonce=0):
self.index = index
self.transactions = transactions
self.timestamp = timestamp
self.previous_hash = previous_hash
self.nonce = nonce
self.hash = self.calculate_hash()
def calculate_hash(self):
"""블록의 SHA-256 해시 계산"""
block_string = json.dumps({
"index": self.index,
"transactions": self.transactions,
"timestamp": str(self.timestamp),
"previous_hash": self.previous_hash,
"nonce": self.nonce
}, sort_keys=True)
return hashlib.sha256(block_string.encode()).hexdigest()
2. 블록체인 클래스 구현
이제 블록들을 연결하고 관리하는 블록체인 클래스를 만들어요. 작업 증명(Proof of Work) 알고리즘도 간단하게 구현해볼게요.
class Blockchain:
def __init__(self):
self.chain = []
self.pending_transactions = []
self.mining_reward = 100
self.difficulty = 2 # 난이도 설정
self.create_genesis_block()
def create_genesis_block(self):
"""최초 블록(제네시스 블록) 생성"""
genesis_block = Block(0, [], datetime.now(), "0")
self.chain.append(genesis_block)
def get_latest_block(self):
"""가장 최근 블록 반환"""
return self.chain[-1]
def mine_pending_transactions(self, mining_reward_address):
"""대기 중인 트랜잭션을 마이닝"""
reward_transaction = {
"from": None,
"to": mining_reward_address,
"amount": self.mining_reward
}
self.pending_transactions.append(reward_transaction)
block = Block(
len(self.chain),
self.pending_transactions,
datetime.now(),
self.get_latest_block().hash
)
# 작업 증명 수행
block.nonce = 0
while not block.hash.startswith('0' * self.difficulty):
block.nonce += 1
block.hash = block.calculate_hash()
print(f"블록 마이닝 성공! Hash: {block.hash}")
self.chain.append(block)
self.pending_transactions = []
3. 트랜잭션 처리 기능 추가
실제 거래를 처리하고 잔액을 확인하는 기능을 추가해요.
def create_transaction(self, from_address, to_address, amount):
"""새 트랜잭션 생성"""
transaction = {
"from": from_address,
"to": to_address,
"amount": amount,
"timestamp": str(datetime.now())
}
self.pending_transactions.append(transaction)
def get_balance(self, address):
"""특정 주소의 잔액 계산"""
balance = 0
for block in self.chain:
for transaction in block.transactions:
if transaction['from'] == address:
balance -= transaction['amount']
if transaction['to'] == address:
balance += transaction['amount']
return balance
def is_chain_valid(self):
"""블록체인 무결성 검증"""
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
# 작업 증명 확인
if not current_block.hash.startswith('0' * self.difficulty):
return False
return True
Flask로 REST API 구현
블록체인을 웹 API로 만들어 실제로 사용할 수 있게 해요.
from flask import Flask, jsonify, request
app = Flask(__name__)
blockchain = Blockchain()
@app.route('/mine', methods=['POST'])
def mine():
"""블록 마이닝 엔드포인트"""
data = request.get_json()
miner_address = data.get('miner_address')
if not miner_address:
return jsonify({'error': '마이너 주소가 필요해요'}), 400
blockchain.mine_pending_transactions(miner_address)
return jsonify({
'message': '블록 마이닝 성공!',
'block': {
'index': blockchain.get_latest_block().index,
'hash': blockchain.get_latest_block().hash,
'transactions': blockchain.get_latest_block().transactions
}
})
@app.route('/transaction', methods=['POST'])
def create_transaction():
"""트랜잭션 생성 엔드포인트"""
data = request.get_json()
required = ['from_address', 'to_address', 'amount']
if not all(k in data for k in required):
return jsonify({'error': '필수 정보가 누락되었어요'}), 400
blockchain.create_transaction(
data['from_address'],
data['to_address'],
data['amount']
)
return jsonify({'message': '트랜잭션이 추가되었어요'})
@app.route('/balance/<address>')
def get_balance(address):
"""잔액 조회 엔드포인트"""
balance = blockchain.get_balance(address)
return jsonify({'address': address, 'balance': balance})
@app.route('/chain')
def get_chain():
"""전체 블록체인 조회"""
chain_data = []
for block in blockchain.chain:
chain_data.append({
'index': block.index,
'transactions': block.transactions,
'timestamp': str(block.timestamp),
'previous_hash': block.previous_hash,
'hash': block.hash,
'nonce': block.nonce
})
return jsonify({
'chain': chain_data,
'length': len(chain_data),
'valid': blockchain.is_chain_valid()
})
if __name__ == '__main__':
app.run(debug=True, port=5000)
실전 테스트 시나리오
이제 만든 블록체인을 실제로 테스트해볼게요. 터미널에서 다음 명령어들을 실행해보세요.
# test_blockchain.py
def test_blockchain():
# 블록체인 생성
my_blockchain = Blockchain()
# 트랜잭션 생성
my_blockchain.create_transaction("Alice", "Bob", 50)
my_blockchain.create_transaction("Bob", "Charlie", 25)
# 첫 번째 블록 마이닝
print("마이닝 시작...")
my_blockchain.mine_pending_transactions("Miner1")
# 잔액 확인
print(f"Bob의 잔액: {my_blockchain.get_balance('Bob')}")
print(f"Miner1의 잔액: {my_blockchain.get_balance('Miner1')}")
# 추가 트랜잭션
my_blockchain.create_transaction("Alice", "Dan", 30)
my_blockchain.mine_pending_transactions("Miner2")
# 블록체인 무결성 확인
print(f"블록체인 유효성: {my_blockchain.is_chain_valid()}")
# 블록체인 조작 시도
my_blockchain.chain[1].transactions[0]['amount'] = 1000
print(f"조작 후 유효성: {my_blockchain.is_chain_valid()}")
if __name__ == "__main__":
test_blockchain()
보안 고려사항과 개선점
필수 보안 체크리스트
현재 구현은 교육 목적이에요. 실제 사용하려면 다음 사항들을 반드시 추가해야 해요:
1. 디지털 서명 구현
import ecdsa
import binascii
class Wallet:
def __init__(self):
self.private_key = ecdsa.SigningKey.generate(curve=ecdsa.SECP256k1)
self.public_key = self.private_key.get_verifying_key()
self.address = self.generate_address()
def generate_address(self):
"""공개키로부터 주소 생성"""
public_key_bytes = self.public_key.to_string()
return hashlib.sha256(public_key_bytes).hexdigest()
def sign_transaction(self, transaction):
"""트랜잭션에 서명"""
message = json.dumps(transaction, sort_keys=True).encode()
signature = self.private_key.sign(message)
return binascii.hexlify(signature).decode()
2. 합의 알고리즘 개선
- Proof of Stake 구현 고려
- 난이도 자동 조절 메커니즘
- 포크 처리 로직
3. 네트워크 기능
- P2P 통신 구현
- 노드 간 동기화
- 브로드캐스팅 메커니즘
흔한 에러와 해결법
문제 1: 마이닝이 너무 오래 걸려요
# 난이도를 낮춰보세요
blockchain.difficulty = 1 # 2에서 1로 변경
문제 2: Flask 서버가 시작되지 않아요
# 포트 충돌 확인
lsof -i :5000 # Mac/Linux
netstat -ano | findstr :5000 # Windows
# 다른 포트로 변경
app.run(debug=True, port=5001)
문제 3: 해시 계산이 일치하지 않아요
# JSON 직렬화 순서 고정
json.dumps(data, sort_keys=True) # sort_keys 필수!
중요 경고: 이 코드는 교육 목적이에요. 실제 자금이나 중요한 데이터를 다루는 시스템에는 절대 사용하지 마세요. 프로덕션 환경에서는 검증된 블록체인 플랫폼(이더리움, 하이퍼레저 등)을 사용하세요.