캡스톤 프로젝트: PyChain 완성 — REST API 노드와 포트폴리오 배포
학습 목표
- ✓Flask를 사용하여 블록체인 노드의 REST API를 구현하고 curl 또는 Postman으로 테스트할 수 있다
- ✓전체 프로젝트에 대한 통합 테스트를 작성하여 핵심 기능(채굴, 검증, 이중 지불 방지)을 자동 검증할 수 있다
- ✓해시 함수부터 체인 검증까지 전체 블록체인 동작 원리를 처음부터 끝까지 자신의 코드를 기반으로 설명할 수 있다
캡스톤 프로젝트: PyChain 완성 — REST API 노드와 포트폴리오 배포
여기까지 온 당신에게
블록체인을 만들 줄 아는 사람은 많다. 그걸 서비스로 띄울 줄 아는 사람은 드물다.
레슨 9에서 우리는 공격자의 눈으로 블록체인을 해부했다. 51% 공격, 이기적 채굴, 시빌 공격. 시스템의 균열을 직접 들여다본 사람만이 진짜 견고한 시스템을 설계한다. resolve_conflicts()로 가장 긴 유효 체인을 선택하는 합의 로직까지 구현하면서, PyChain은 "장난감"에서 "작동하는 프로토타입"으로 한 단계 올라섰다.
오늘은 졸업 시험이다. 9개 레슨에 걸쳐 쌓아온 모든 조각 — 해시 함수, 디지털 서명, 트랜잭션, 블록, PoW, 체인 검증, UTXO, 머클 트리, 보안 — 을 하나의 Flask REST API 서버로 통합한다.
내가 처음 이더리움 노드 클라이언트 코드를 읽었을 때, 가장 놀란 건 블록체인 로직이 아니었다. 그걸 감싸는 네트워크 레이어의 양이었다. 체인 로직은 전체의 30%도 안 됐다. 나머지는 전부 API, 피어 통신, 상태 관리. 엔진보다 차체가 더 큰 셈이다. 오늘 그 경험을 바탕으로 "코어 로직을 HTTP API로 감싸는 기술"을 가르친다.
Today's Mission: 브라우저에서 작동하는 블록체인 노드
오늘 완성할 산출물:
| 파일 | 역할 | 상태 |
|---|---|---|
blockchain.py | 코어 블록체인 엔진 | ✅ 레슨 1~9에서 완성 |
app.py | Flask REST API 서버 | 🆕 오늘 작성 |
test_pychain.py | pytest 통합 테스트 | 🆕 오늘 작성 |
README.md | 포트폴리오 문서 | 🆕 오늘 작성 |
최종 결과물: curl 한 줄로 트랜잭션을 보내고, 블록을 채굴하고, 체인 전체를 조회할 수 있는 작동하는 블록체인 노드.
🤔 생각해보세요: 왜 블록체인 노드에 REST API가 필요할까? 비트코인 코어도 JSON-RPC API를 제공한다. 지갑 앱, 블록 탐색기, 마이닝 풀 — 전부 이 API를 통해 노드와 대화한다.
💡 REST vs JSON-RPC
비트코인은 JSON-RPC를, 이더리움도 JSON-RPC를 쓴다. 하지만 학습 프로젝트에서는 REST가 더 직관적이다. GET /chain이 {"jsonrpc":"2.0","method":"getblockchain"} 보다 읽기 쉽다. 실무에서는 gRPC를 쓰는 추세(go-ethereum의 경우)지만, 원리는 같다 — 코어 로직을 네트워크 인터페이스로 노출하는 것.
Step 1: 코어 엔진 정리 — blockchain.py 최종본
9개 레슨에 걸쳐 만든 코드를 하나의 파일로 통합한다. 레슨 6에서 설계한 Blockchain 클래스의 chain 리스트와 pending_transactions, 레슨 7의 UTXO 잔고 추적, 레슨 8의 머클 루트 계산 — 흩어져 있던 퍼즐 조각들이 드디어 한 그림으로 맞춰진다.
# blockchain.py — PyChain 코어 엔진 (레슨 1~9 통합)
import hashlib
import json
import time
from typing import List, Dict, Any
class Block:
"""블록 하나를 표현하는 클래스 (레슨 4에서 설계한 구조)"""
def __init__(self, index: int, transactions: List[Dict],
previous_hash: str, nonce: int = 0):
self.index = index
self.timestamp = time.time()
self.transactions = transactions
self.previous_hash = previous_hash
self.nonce = nonce
self.merkle_root = self.compute_merkle_root()
self.hash = self.compute_hash()
def compute_hash(self) -> str:
"""블록 데이터를 SHA-256으로 해싱 (레슨 1의 핵심)"""
block_data = json.dumps({
"index": self.index,
"timestamp": self.timestamp,
"transactions": self.transactions,
"previous_hash": self.previous_hash,
"nonce": self.nonce,
"merkle_root": self.merkle_root
}, sort_keys=True)
return hashlib.sha256(block_data.encode()).hexdigest()
def compute_merkle_root(self) -> str:
"""트랜잭션들의 머클 루트 계산 (레슨 8의 '토너먼트 대진표')"""
if not self.transactions:
return hashlib.sha256(b"empty").hexdigest()
tx_hashes = [
hashlib.sha256(json.dumps(tx, sort_keys=True).encode()).hexdigest()
for tx in self.transactions
]
while len(tx_hashes) > 1:
if len(tx_hashes) % 2 == 1:
tx_hashes.append(tx_hashes[-1]) # 홀수면 마지막 복제
tx_hashes = [
hashlib.sha256((tx_hashes[i] + tx_hashes[i+1]).encode()).hexdigest()
for i in range(0, len(tx_hashes), 2)
]
return tx_hashes[0]
def to_dict(self) -> Dict[str, Any]:
"""블록을 JSON 직렬화 가능한 딕셔너리로 변환"""
return {
"index": self.index,
"timestamp": self.timestamp,
"transactions": self.transactions,
"previous_hash": self.previous_hash,
"nonce": self.nonce,
"merkle_root": self.merkle_root,
"hash": self.hash
}
class Blockchain:
"""
PyChain 블록체인 — 레슨 6의 체인 구조 + 레슨 5의 PoW
"""
DIFFICULTY = 4 # 해시 앞에 '0' 4개 필요
MINING_REWARD = 50 # 채굴 보상
def __init__(self):
self.chain: List[Block] = []
self.pending_transactions: List[Dict] = []
self.utxo_pool: Dict[str, List[Dict]] = {} # 레슨 7: UTXO 풀
self._create_genesis_block()
def _create_genesis_block(self):
"""제네시스 블록 생성 — 모든 체인의 시작"""
genesis = Block(0, [], "0" * 64)
genesis.nonce = 0
genesis.hash = genesis.compute_hash()
self.chain.append(genesis)
def get_last_block(self) -> Block:
return self.chain[-1]
def add_transaction(self, sender: str, recipient: str, amount: float) -> int:
"""
대기 트랜잭션 추가 — 레슨 3의 트랜잭션 구조
반환값: 이 트랜잭션이 포함될 블록 인덱스
"""
if sender != "MINING_REWARD":
balance = self.get_balance(sender)
if balance < amount:
raise ValueError(
f"잔액 부족: {sender}의 잔고 {balance}, 전송 시도 {amount}"
)
self.pending_transactions.append({
"sender": sender,
"recipient": recipient,
"amount": amount
})
return self.get_last_block().index + 1
def get_balance(self, address: str) -> float:
"""
체인 전체를 순회하여 잔고 계산
레슨 7의 UTXO '지폐 다발' 개념 — 단순화 버전
"""
balance = 0.0
for block in self.chain:
for tx in block.transactions:
if tx["sender"] == address:
balance -= tx["amount"]
if tx["recipient"] == address:
balance += tx["amount"]
# 대기 트랜잭션도 반영
for tx in self.pending_transactions:
if tx["sender"] == address:
balance -= tx["amount"]
return balance
def proof_of_work(self, block: Block) -> Block:
"""
PoW 채굴 — 레슨 5에서 구현한 넌스 돌리기
해시 앞자리가 '0' * DIFFICULTY로 시작할 때까지 반복
"""
target = "0" * self.DIFFICULTY
while not block.hash.startswith(target):
block.nonce += 1
block.hash = block.compute_hash()
return block
def mine_pending_transactions(self, miner_address: str) -> Block:
"""대기 트랜잭션을 블록으로 묶고 채굴"""
# 채굴 보상 트랜잭션 추가
self.pending_transactions.append({
"sender": "MINING_REWARD",
"recipient": miner_address,
"amount": self.MINING_REWARD
})
new_block = Block(
index=len(self.chain),
transactions=self.pending_transactions,
previous_hash=self.get_last_block().hash
)
mined_block = self.proof_of_work(new_block)
self.chain.append(mined_block)
self.pending_transactions = []
return mined_block
def is_chain_valid(self) -> bool:
"""
체인 무결성 검증 — 레슨 6의 핵심
레슨 9에서 배운 것: 이 검증이 있어야 공격 체인을 거부할 수 있다
"""
target = "0" * self.DIFFICULTY
for i in range(1, len(self.chain)):
current = self.chain[i]
previous = self.chain[i - 1]
# 해시 재계산 검증
if current.hash != current.compute_hash():
return False
# 이전 블록 해시 연결 검증
if current.previous_hash != previous.hash:
return False
# PoW 검증
if not current.hash.startswith(target):
return False
return True
def to_dict(self) -> List[Dict]:
"""체인 전체를 딕셔너리 리스트로 변환"""
return [block.to_dict() for block in self.chain]
200줄이 안 되는 코드다. 하지만 이 안에 해시 함수, 머클 트리, PoW, 잔고 추적, 체인 검증까지 — 비트코인의 핵심 메커니즘이 빠짐없이 들어있다. 엔진은 완성됐다. 이제 이 엔진에 핸들과 페달을 달아서 누구나 운전할 수 있게 만들 차례다.
🤔 생각해보세요:
mine_pending_transactions에서 채굴 보상을pending_transactions에 넣고 바로 블록에 포함시킨다. 이 순서가 중요한 이유는?
답변 보기
채굴 보상 트랜잭션(coinbase)은 반드시 해당 블록의 트랜잭션 목록 안에 있어야 머클 루트에 포함된다. 머클 루트가 바뀌면 블록 해시가 바뀌고, PoW를 다시 해야 한다. 비트코인에서도 coinbase 트랜잭션은 항상 블록의 첫 번째 트랜잭션이다.
Step 2: Flask API 설계 — 엔드포인트 맵
왜 Flask인가? Django는 이 규모에선 대포로 참새를 잡는 격이고, FastAPI의 async는 단일 노드 프로토타입에선 오버스펙이다. 블록체인 노드처럼 가볍고 빠르게 돌아야 하는 서버에는 Flask의 미니멀함이 딱 맞는다. 이더리움의 초기 Python 클라이언트(pyethereum)도 비슷한 철학으로 만들어졌다.
먼저 설계부터 하자. 클라이언트가 어떤 요청을 보내면, Flask가 받아서 블록체인 엔진에 전달하고, 결과를 JSON으로 돌려준다. 레스토랑의 웨이터와 같다 — 손님(클라이언트)과 주방(엔진) 사이에서 주문을 전달하고 요리를 가져다주는 역할.
| 엔드포인트 | 메서드 | 설명 | 요청 바디 |
|---|---|---|---|
/transactions/new | POST | 새 트랜잭션 추가 | {sender, recipient, amount} |
/mine | POST | 대기 트랜잭션 채굴 | {miner_address} |
/chain | GET | 전체 체인 조회 | — |
/chain/valid | GET | 체인 무결성 검증 | — |
/balance/<address> | GET | 주소별 잔고 조회 | — |
설계도가 완성됐다. 이걸 코드로 바꿔보자.
Step 3: app.py 구현 — 블록체인을 HTTP로 감싸기
# app.py — PyChain Flask REST API 노드
from flask import Flask, jsonify, request
from blockchain import Blockchain
app = Flask(__name__)
node = Blockchain() # 블록체인 인스턴스 생성
@app.route("/transactions/new", methods=["POST"])
def new_transaction():
"""새 트랜잭션을 대기 풀에 추가"""
data = request.get_json()
required = ["sender", "recipient", "amount"]
if not data or not all(k in data for k in required):
return jsonify({"error": "sender, recipient, amount 필드가 필요합니다"}), 400
try:
block_index = node.add_transaction(
sender=data["sender"],
recipient=data["recipient"],
amount=float(data["amount"])
)
return jsonify({
"message": f"트랜잭션이 블록 {block_index}에 추가 예정",
"transaction": data
}), 201
except ValueError as e:
return jsonify({"error": str(e)}), 400
@app.route("/mine", methods=["POST"])
def mine():
"""대기 트랜잭션을 채굴하여 새 블록 생성"""
data = request.get_json()
if not data or "miner_address" not in data:
return jsonify({"error": "miner_address 필드가 필요합니다"}), 400
if len(node.pending_transactions) == 0:
return jsonify({"error": "채굴할 트랜잭션이 없습니다"}), 400
new_block = node.mine_pending_transactions(data["miner_address"])
return jsonify({
"message": "새 블록이 채굴되었습니다! ⛏️",
"block": new_block.to_dict()
}), 200
@app.route("/chain", methods=["GET"])
def get_chain():
"""전체 블록체인 조회"""
return jsonify({
"chain": node.to_dict(),
"length": len(node.chain)
}), 200
@app.route("/chain/valid", methods=["GET"])
def validate_chain():
"""체인 무결성 검증 — 레슨 9의 보안 검증"""
is_valid = node.is_chain_valid()
return jsonify({
"valid": is_valid,
"length": len(node.chain),
"message": "체인이 유효합니다 ✅" if is_valid
else "체인이 손상되었습니다 ❌"
}), 200
@app.route("/balance/<address>", methods=["GET"])
def get_balance(address):
"""특정 주소의 잔고 조회"""
balance = node.get_balance(address)
return jsonify({
"address": address,
"balance": balance
}), 200
@app.route("/pending", methods=["GET"])
def get_pending():
"""대기 중인 트랜잭션 조회"""
return jsonify({
"pending_transactions": node.pending_transactions,
"count": len(node.pending_transactions)
}), 200
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=True)
50줄짜리 파일 하나로 블록체인 노드가 인터넷에 열린다. 짧지만 허술하지 않다. 내가 스마트 컨트랙트 감사를 하면서 뼈저리게 배운 교훈이 있다 — API 레이어에서 입력 검증을 빠뜨리는 순간 보안 구멍이 생긴다. amount를 float()으로 변환하는 것, 필수 필드를 체크하는 것, 이런 사소해 보이는 검증 한 줄 한 줄이 방어벽이다.
잔고 부족 시 ValueError를 던지는 것도 마찬가지다. 레슨 7에서 UTXO를 '지폐 다발'에 비유했던 걸 떠올려보자 — 주머니에 있는 것보다 많이 꺼낼 수는 없다. 이중 지불 방지의 첫 번째 방어선이 바로 여기다.
❌ → 🤔 → ✅ API 에러 처리의 3단계 진화
API에서 입력 검증과 에러 처리를 어떻게 하느냐에 따라 "학생 프로젝트"와 "포트폴리오에 올릴 만한 코드"가 갈린다. 트랜잭션 엔드포인트를 예로 들어 그 차이를 직접 비교해보자.
❌ WRONG WAY — 검증 없이 그냥 넘기기
@app.route("/transactions/new", methods=["POST"])
def new_transaction():
data = request.get_json()
# 아무 검증 없이 바로 호출 — 뭐가 문제일까?
block_index = node.add_transaction(
data["sender"], data["recipient"], data["amount"]
)
return jsonify({"message": "추가됨"}), 200
문제점이 산더미다:
data가None이면? →TypeError: 'NoneType' object is not subscriptable— 500 에러와 함께 Flask 내부 스택 트레이스가 클라이언트에 노출된다. 공격자에게 서버 구조를 알려주는 셈."amount"에"abc"가 오면? → 블록체인 엔진 깊숙이 들어가서야 터진다. 어디서 문제가 생겼는지 추적이 어렵다.- 잔액 부족 시? →
ValueError예외가 처리 안 되어 역시 500 에러. 클라이언트는 자기가 뭘 잘못한 건지 알 수 없다. - 상태 코드가 항상 200 — 성공이든 실패든 같은 코드를 반환하면 클라이언트 프로그램이 결과를 구분할 수 없다.
🤔 BETTER — try/except로 감싸기
@app.route("/transactions/new", methods=["POST"])
def new_transaction():
try:
data = request.get_json()
block_index = node.add_transaction(
data["sender"], data["recipient"], data["amount"]
)
return jsonify({"message": f"블록 {block_index}에 추가 예정"}), 201
except Exception as e:
return jsonify({"error": str(e)}), 400
나아졌다 — 최소한 500 에러 대신 400을 돌려준다. 하지만:
- 모든 에러를 한 바구니에 담는다. 필드 누락(클라이언트 잘못)이든, 잔액 부족(비즈니스 규칙)이든, 서버 내부 오류든 전부 같은 400. 클라이언트가 "내 요청 형식이 틀린 건가, 잔고가 부족한 건가?"를 구분할 수 없다.
amount에 문자열이 와도 엔진까지 전달된다. 경계에서 걸러야 할 것을 안쪽으로 흘려보내는 실수다.Exception을 포괄적으로 잡으면 진짜 서버 버그도 400으로 숨겨버린다. 디버깅할 때 지옥이 열린다.
✅ BEST — 경계에서 검증하고, 에러를 구분하기
@app.route("/transactions/new", methods=["POST"])
def new_transaction():
data = request.get_json()
# 1단계: 입력 형식 검증 — API 경계에서 걸러낸다
required = ["sender", "recipient", "amount"]
if not data or not all(k in data for k in required):
return jsonify({"error": "sender, recipient, amount 필드가 필요합니다"}), 400
# 2단계: 타입 변환 — 엔진에 넘기기 전에 정제한다
try:
amount = float(data["amount"])
except (ValueError, TypeError):
return jsonify({"error": "amount는 숫자여야 합니다"}), 400
# 3단계: 비즈니스 로직 에러만 따로 처리
try:
block_index = node.add_transaction(
sender=data["sender"],
recipient=data["recipient"],
amount=amount
)
return jsonify({
"message": f"트랜잭션이 블록 {block_index}에 추가 예정",
"transaction": data
}), 201
except ValueError as e:
# 잔액 부족 등 비즈니스 규칙 위반
return jsonify({"error": str(e)}), 400
왜 이게 BEST인가:
- 3겹의 방어벽: 형식 검증 → 타입 변환 → 비즈니스 로직. 각 단계에서 자기 책임의 에러만 잡는다.
- 클라이언트가 받는 에러 메시지가 행동 가능하다(actionable). "필드가 없다" vs "숫자가 아니다" vs "잔액 부족" — 각각 다른 대응이 필요하고, 메시지가 그걸 안내한다.
- 예상하지 못한 에러(
RuntimeError등)는 잡지 않으므로 500으로 올라간다. 이건 의도적인 설계다 — 서버 버그는 숨기지 말고 드러내야 고친다. - 레슨 9에서 배운 보안 원칙의 실전 적용이다: 시스템의 경계(API 레이어)에서 모든 외부 입력을 의심하고 검증하라.
💡 이 패턴은 블록체인에만 해당되는 게 아니다. 모든 REST API에 적용된다. 면접에서 "API 에러 처리를 어떻게 하겠습니까?"라는 질문에 이 3단계를 설명할 수 있으면, 주니어와 중급 개발자의 차이를 보여주는 답이 된다.
엔진을 만들었고, 핸들도 달았다. 이제 직접 운전해볼 시간이다.
Step 4: 실전 테스트 — curl로 노드와 대화하기
서버를 실행하고 순서대로 따라하자.
터미널 1 — 서버 시작:
# 필요한 패키지 설치
pip install flask pytest
# 서버 실행
python app.py
# Output:
# * Running on http://0.0.0.0:5000
# * Debug mode: on
터미널 2 — API 호출:
# 1) 트랜잭션 생성 — Alice가 Bob에게 보내기
curl -X POST http://localhost:5000/transactions/new \
-H "Content-Type: application/json" \
-d '{"sender":"alice","recipient":"bob","amount":30}'
# Output:
# {"message":"트랜잭션이 블록 1에 추가 예정","transaction":{...}}
# 2) 대기 트랜잭션 확인
curl http://localhost:5000/pending
# Output:
# {"count":1,"pending_transactions":[{"amount":30,"recipient":"bob","sender":"alice"}]}
# 3) 채굴 — alice가 먼저 자기에게 보상을 줌 (채굴자)
curl -X POST http://localhost:5000/mine \
-H "Content-Type: application/json" \
-d '{"miner_address":"alice"}'
# Output:
# {"block":{"hash":"0000...","index":1,...},"message":"새 블록이 채굴되었습니다! ⛏️"}
# 4) 잔고 확인 — alice는 채굴 보상(50) - 전송(30) = 20
curl http://localhost:5000/balance/alice
# Output:
# {"address":"alice","balance":20.0}
# bob은 30을 받았다
curl http://localhost:5000/balance/bob
# Output:
# {"address":"bob","balance":30.0}
# 5) 전체 체인 조회
curl http://localhost:5000/chain
# Output:
# {"chain":[{"index":0,...},{"index":1,...}],"length":2}
# 6) 체인 검증
curl http://localhost:5000/chain/valid
# Output:
# {"length":2,"message":"체인이 유효합니다 ✅","valid":true}
🤔 생각해보세요: alice의 잔고가 20인 상태에서 bob에게 50을 보내려고 하면 어떻게 될까?
답변 보기
add_transaction에서 잔고를 확인하고 ValueError를 던진다. API는 400 에러와 함께 "잔액 부족" 메시지를 반환한다. 레슨 9에서 배운 이중 지불 방지의 실전 적용이다. 직접 테스트해보자:
curl -X POST http://localhost:5000/transactions/new \
-H "Content-Type: application/json" \
-d '{"sender":"alice","recipient":"bob","amount":50}'
# Output:
# {"error":"잔액 부족: alice의 잔고 20.0, 전송 시도 50.0"}
Step 5: pytest 통합 테스트 — 코드의 안전망
"잘 돌아가는데요?"는 근거가 아니다. 수동 테스트로 확인한 건 지금 이 순간 돌아간다는 것뿐이다. 내가 솔리디티 감사 현장에서 목격한 원칙이 하나 있다: 테스트 없는 코드는 작동하는 게 아니라 운이 좋은 것이다. DeFi 프로토콜에서 테스트 커버리지가 낮아서 수백만 달러가 해킹된 사례를 직접 여러 번 봤다.
# test_pychain.py — PyChain 통합 테스트
import pytest
import json
from app import app
from blockchain import Blockchain, Block
@pytest.fixture
def client():
"""Flask 테스트 클라이언트 생성"""
app.config["TESTING"] = True
with app.test_client() as client:
yield client
@pytest.fixture
def fresh_chain():
"""깨끗한 블록체인 인스턴스"""
return Blockchain()
# === 코어 엔진 테스트 ===
class TestBlockchain:
def test_genesis_block_exists(self, fresh_chain):
"""제네시스 블록이 올바르게 생성되는가"""
assert len(fresh_chain.chain) == 1
assert fresh_chain.chain[0].index == 0
assert fresh_chain.chain[0].previous_hash == "0" * 64
def test_mining_creates_new_block(self, fresh_chain):
"""채굴이 새 블록을 생성하는가"""
fresh_chain.add_transaction("alice", "bob", 10)
block = fresh_chain.mine_pending_transactions("miner1")
assert len(fresh_chain.chain) == 2
assert block.index == 1
# PoW 검증 — 해시가 '0000'으로 시작해야 함 (레슨 5)
assert block.hash.startswith("0" * Blockchain.DIFFICULTY)
def test_mining_reward(self, fresh_chain):
"""채굴자가 보상을 받는가"""
fresh_chain.add_transaction("alice", "bob", 10)
fresh_chain.mine_pending_transactions("miner1")
balance = fresh_chain.get_balance("miner1")
assert balance == 50 # MINING_REWARD
def test_insufficient_balance(self, fresh_chain):
"""잔액 부족 시 에러가 발생하는가 (이중 지불 방지)"""
with pytest.raises(ValueError, match="잔액 부족"):
fresh_chain.add_transaction("alice", "bob", 100)
def test_chain_validity(self, fresh_chain):
"""체인 무결성 검증이 작동하는가 (레슨 6)"""
fresh_chain.add_transaction("alice", "bob", 10)
fresh_chain.mine_pending_transactions("miner1")
assert fresh_chain.is_chain_valid() is True
def test_tampered_chain_detected(self, fresh_chain):
"""변조된 체인을 감지하는가 (레슨 9 보안)"""
fresh_chain.add_transaction("alice", "bob", 10)
fresh_chain.mine_pending_transactions("miner1")
# 블록 데이터 변조!
fresh_chain.chain[1].transactions[0]["amount"] = 99999
assert fresh_chain.is_chain_valid() is False
# === API 엔드포인트 테스트 ===
class TestAPI:
def test_get_chain(self, client):
"""GET /chain이 제네시스 블록을 반환하는가"""
resp = client.get("/chain")
data = json.loads(resp.data)
assert resp.status_code == 200
assert data["length"] >= 1
def test_create_transaction(self, client):
"""POST /transactions/new 가 트랜잭션을 추가하는가"""
# 먼저 채굴로 잔고 확보
client.post("/mine", json={"miner_address": "alice"})
resp = client.post("/transactions/new", json={
"sender": "alice",
"recipient": "bob",
"amount": 10
})
assert resp.status_code == 201
def test_missing_fields_returns_400(self, client):
"""필수 필드 누락 시 400 에러"""
resp = client.post("/transactions/new", json={
"sender": "alice"
# recipient, amount 누락
})
assert resp.status_code == 400
def test_mine_and_validate(self, client):
"""채굴 후 체인이 유효한가"""
client.post("/mine", json={"miner_address": "tester"})
resp = client.get("/chain/valid")
data = json.loads(resp.data)
assert data["valid"] is True
def test_balance_after_transaction(self, client):
"""트랜잭션 후 잔고가 올바른가"""
# alice에게 채굴 보상 부여
client.post("/mine", json={"miner_address": "alice"})
# alice → bob 전송
client.post("/transactions/new", json={
"sender": "alice", "recipient": "bob", "amount": 20
})
client.post("/mine", json={"miner_address": "alice"})
resp = client.get("/balance/bob")
data = json.loads(resp.data)
assert data["balance"] == 20.0
테스트 실행:
pytest test_pychain.py -v
# Output:
# test_pychain.py::TestBlockchain::test_genesis_block_exists PASSED
# test_pychain.py::TestBlockchain::test_mining_creates_new_block PASSED
# test_pychain.py::TestBlockchain::test_mining_reward PASSED
# test_pychain.py::TestBlockchain::test_insufficient_balance PASSED
# test_pychain.py::TestBlockchain::test_chain_validity PASSED
# test_pychain.py::TestBlockchain::test_tampered_chain_detected PASSED
# test_pychain.py::TestAPI::test_get_chain PASSED
# test_pychain.py::TestAPI::test_create_transaction PASSED
# test_pychain.py::TestAPI::test_missing_fields_returns_400 PASSED
# test_pychain.py::TestAPI::test_mine_and_validate PASSED
# test_pychain.py::TestAPI::test_balance_after_transaction PASSED
# ======================== 11 passed ========================
11개 전부 초록불. 이 순간부터 당신의 블록체인은 **"돌아가는 것 같다"가 아니라 "검증됐다"**고 말할 수 있다.
🔍 심화: 왜 `fresh_chain` fixture를 쓰는가?
Flask 앱의 node 객체는 전역 변수라서 테스트 간 상태가 공유된다. API 테스트는 이 전역 상태를 써서 통합 테스트 성격을 가지지만, 코어 엔진 테스트에서는 fresh_chain fixture로 매 테스트마다 깨끗한 체인을 쓴다. 실무에서도 단위 테스트는 격리, 통합 테스트는 공유 상태 원칙을 따른다.
Step 6: README.md — 포트폴리오의 얼굴
코드를 잘 짜는 것과 그걸 잘 보여주는 것은 전혀 다른 기술이다. GitHub 채용 담당자는 README를 15초 본다. 코드를 열어보기도 전에 README에서 판단이 끝난다. 내가 수백 개의 Web3 프로젝트 레포를 리뷰하면서 깨달은 것: 사용 예시가 없는 README는 무시당한다.
# ⛓️ PyChain — 파이썬으로 만드는 미니 블록체인
SHA-256 해시, 작업 증명(PoW), UTXO 잔고 추적, 머클 트리를
직접 구현한 교육용 블록체인 + REST API 노드.
## 아키텍처
PyChain/ ├── blockchain.py # 코어 엔진 (Block, Blockchain 클래스) ├── app.py # Flask REST API 서버 ├── test_pychain.py # pytest 통합 테스트 (11개) └── README.md
## 핵심 기능
- ⛏️ SHA-256 기반 Proof-of-Work 채굴
- 💰 트랜잭션 생성 및 잔고 검증 (이중 지불 방지)
- 🌳 머클 트리로 트랜잭션 무결성 보장
- 🔗 체인 무결성 검증 (해시 체인 + PoW 검증)
- 🌐 REST API로 브라우저/curl에서 상호작용
## 빠른 시작
```bash
pip install flask pytest
python app.py
API 사용 예시
# 채굴
curl -X POST http://localhost:5000/mine \
-H "Content-Type: application/json" \
-d '{"miner_address":"alice"}'
# 트랜잭션 전송
curl -X POST http://localhost:5000/transactions/new \
-H "Content-Type: application/json" \
-d '{"sender":"alice","recipient":"bob","amount":10}'
# 체인 조회
curl http://localhost:5000/chain
테스트
pytest test_pychain.py -v
학습 자료
이 프로젝트는 '비트코인과 블록체인 완전 입문' 과정의 캡스톤 프로젝트입니다. 구현 원리:
- SHA-256 해시 함수와 데이터 무결성
- 공개키 암호화와 디지털 서명
- UTXO 모델 기반 잔고 추적
- 머클 트리를 활용한 트랜잭션 요약
- 51% 공격과 블록체인 보안 모델
> ⚠️ 위 README 자체가 마크다운 코드 블록 안에 있어서 복잡하게 보이지만, 실제 `README.md` 파일로 저장하면 깔끔하게 렌더링된다.
```mermaid
mindmap
root((PyChain<br/>프로젝트))
코어 엔진
Block 클래스
Blockchain 클래스
PoW 채굴
머클 루트
API 레이어
/transactions/new
/mine
/chain
/chain/valid
/balance
테스트
코어 테스트 6개
API 테스트 5개
문서화
README.md
아키텍처 다이어그램
사용 예시
Review: 자기 점검 체크리스트
프로젝트가 포트폴리오에 올릴 수준인지, 하나씩 짚어보자.
| # | 항목 | 확인 |
|---|---|---|
| 1 | python app.py로 서버가 정상 시작되는가? | ☐ |
| 2 | curl로 트랜잭션 생성 → 채굴 → 잔고 확인 흐름이 작동하는가? | ☐ |
| 3 | 잔액 부족 시 400 에러가 반환되는가? | ☐ |
| 4 | pytest test_pychain.py -v에서 11개 전부 PASSED인가? | ☐ |
| 5 | /chain/valid가 True를 반환하는가? | ☐ |
| 6 | README.md에 설치 방법, 사용 예시, 아키텍처가 있는가? | ☐ |
| 7 | 변조 감지 테스트가 작동하는가? (체인 위조 시 False) | ☐ |
❌ 흔한 실수 TOP 3:
blockchain.py와app.py를 같은 디렉토리에 놓지 않음 —from blockchain import Blockchain이 실패한다.- 채굴 없이 트랜잭션 전송 시도 — 잔고가 0이라 무조건 실패한다. 반드시 먼저 채굴해서 보상을 받아야 보낼 돈이 생긴다.
- Flask 앱의 전역
node객체를 모르고 서버 재시작 — 체인이 통째로 초기화된다. 메모리에만 존재하기 때문이다. 실무에서는 디스크에 저장한다.
💡 체크리스트 7번에서 막혔다면
test_tampered_chain_detected 테스트를 보자. 블록의 transactions 리스트를 직접 수정하면, compute_hash() 결과가 저장된 hash와 달라진다. is_chain_valid()에서 이 불일치를 잡아낸다. 레슨 6에서 "이전 블록의 해시를 다음 블록이 저장한다"는 원리가 여기서 빛을 발한다 — 하나의 블록을 고치면 그 뒤의 모든 블록 해시가 깨진다.
Next Level: 시니어가 여기에 추가하는 것들
여기까지 완성했다면 이미 견고한 프로토타입이다. 하지만 포트폴리오에서 눈에 띄고 싶다면, 실제 프로덕션 블록체인 프로젝트와의 차이를 좁혀야 한다.
1. 피어 네트워크 (P2P)
지금은 단일 노드다. 혼자 있는 블록체인은 그냥 데이터베이스다. 진짜 블록체인은 여러 노드가 서로 체인을 공유하고 검증한다. Flask 서버를 포트 다르게 2개 띄우고, /nodes/register와 /nodes/resolve 엔드포인트를 추가해보자. 레슨 9의 resolve_conflicts() — 가장 긴 유효 체인을 선택하는 로직이 드디어 제 자리를 찾는다.
2. 디스크 영속성 서버를 껐다 켜면 체인이 통째로 날아간다. SQLite나 JSON 파일로 체인 상태를 저장하는 기능을 추가하면 실무 감각이 확 살아난다.
3. 웹 프론트엔드
/chain 응답을 JavaScript로 파싱해서 블록 탐색기 UI를 만들어보자. 블록을 카드로 보여주고, 클릭하면 트랜잭션 목록이 펼쳐지는 간단한 SPA면 충분하다.
4. Docker 컨테이너화
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY . .
RUN pip install flask pytest
EXPOSE 5000
CMD ["python", "app.py"]
이 4줄이면 어디서든 docker run으로 노드를 띄울 수 있다. 면접관이 "이걸 어떻게 배포하겠습니까?"라고 물었을 때, 답이 이미 준비된 거다.
🔨 프로젝트 업데이트
지금까지의 전체 프로젝트 구조
pychain/
├── blockchain.py ← 레슨 1~9 통합 코어 엔진
├── app.py ← 🆕 레슨 10: Flask REST API
├── test_pychain.py ← 🆕 레슨 10: pytest 통합 테스트
└── README.md ← 🆕 레슨 10: 포트폴리오 문서
이번 레슨에서 추가된 것
app.py— 6개의 REST 엔드포인트로 블록체인 노드를 HTTP 서비스로 변환test_pychain.py— 11개의 자동화 테스트 (코어 6개 + API 5개)README.md— GitHub 포트폴리오용 프로젝트 문서
실행 순서
# 1. 프로젝트 디렉토리 생성
mkdir pychain && cd pychain
# 2. blockchain.py, app.py, test_pychain.py 파일 생성
# (위 코드를 각 파일에 복사-붙여넣기)
# 3. 의존성 설치
pip install flask pytest
# 4. 테스트 먼저 실행 — 모든 것이 작동하는지 확인
pytest test_pychain.py -v
# 예상 출력: 11 passed
# 5. 서버 실행
python app.py
# 예상 출력: * Running on http://0.0.0.0:5000
# 6. 새 터미널에서 API 테스트
curl -X POST http://localhost:5000/mine \
-H "Content-Type: application/json" \
-d '{"miner_address":"alice"}'
curl http://localhost:5000/chain
curl http://localhost:5000/chain/valid
지금 바로 실행해보자. 11개 테스트가 전부 통과하고, curl로 채굴과 트랜잭션이 되는 걸 직접 눈으로 확인했다면 — 축하한다. 당신은 해시 함수부터 REST API 노드까지, 블록체인을 처음부터 끝까지 직접 만든 것이다.
전체 여정 정리 다이어그램
점선 화살표를 보자. 각 레슨에서 만든 개별 함수가 최종 API 노드의 정확히 어느 지점에 꽂히는지 보여준다. 라이브러리를 가져다 쓴 게 아니다. 한 줄 한 줄 직접 작성한 코드가 하나의 시스템으로 맞물려 돌아간다. 그게 "밑바닥부터 만든다"의 진짜 의미다.
난이도 포크
🟢 쉬웠다면
핵심 정리:
- 블록체인 코어 로직은
blockchain.py하나에 200줄 미만 - Flask API로 감싸면 HTTP를 통해 누구나 접근 가능
- pytest로 코어와 API를 동시에 테스트
- README는 포트폴리오의 첫인상 — 사용 예시를 반드시 포함
다음 단계: 이 프로젝트를 GitHub에 올리고, 피어 네트워크(P2P) 기능을 추가해보세요. 두 노드가 체인을 동기화하는 것 — 그게 진짜 "분산" 블록체인이다.
🟡 어려웠다면
Flask가 처음이라 어려웠을 수 있다. 핵심만 다시 잡자:
@app.route("/경로", methods=["GET"])는 "이 URL로 요청이 오면 아래 함수를 실행해" 라는 뜻이다request.get_json()은 클라이언트가 보낸 JSON 데이터를 파이썬 딕셔너리로 변환한다jsonify()는 파이썬 딕셔너리를 JSON 응답으로 변환한다- 결국 Flask는 딕셔너리를 받아서 딕셔너리를 돌려주는 중개자다
추가 연습: 서버를 켜고 curl 대신 브라우저에서 http://localhost:5000/chain에 접속해보자. JSON이 바로 보인다. GET 요청은 브라우저 주소창에 URL을 치는 것과 같다.
🔴 도전 과제
면접 수준 확장 과제:
-
난이도 자동 조절: 블록이 10개 생길 때마다
DIFFICULTY를 재계산하여 채굴 시간이 평균 10초가 되도록 조정하라. 비트코인의 2016블록 난이도 조절을 단순화한 것이다. -
멀티 노드 합의:
app.py를 포트 5000, 5001에서 각각 실행하고,/nodes/register로 서로를 등록한 뒤/nodes/resolve로 가장 긴 체인을 동기화하는 기능을 구현하라.requests라이브러리를 사용해 다른 노드에 HTTP 요청을 보낸다. -
디지털 서명 통합: 레슨 2에서 배운 ECDSA 서명을 트랜잭션에 적용하라.
add_transaction에서 서명을 검증하고, 서명이 없거나 유효하지 않으면 거부하는 로직을 추가하라.
코스 마무리: 당신이 만든 것
10개의 레슨. 파이썬 200줄. 밑바닥부터 쌓아올린 블록체인.
SHA-256으로 데이터의 지문을 찍었다. 디지털 서명으로 신원을 증명했다. 트랜잭션을 구조화하고, 블록에 담아 PoW로 채굴하고, 체인으로 엮고, 머클 트리로 요약하고, 공격을 분석하고, 마지막으로 REST API로 세상에 열었다.
이제 누군가 "블록체인이 뭐야?"라고 물으면, 추상적인 비유를 늘어놓을 필요가 없다. 자기가 작성한 코드를 보여주면 된다. 그게 진짜 이해다.