파이썬으로 간단한 블록체인 만들기
블록체인은 블록이라고 불리는 것들이 암호기법을 통해 체인처럼 연결된 것 [1]을 말합니다.
흔히 아래와 같은 형태로 표현되는데요:
위의 정의에서 '암호기법을 통해 연결된다'와 같은 부분들은 실제로 간단한 블록체인을 만들어보면 복잡한 설명이 필요없이 쉽게 이해가 됩니다. 이번 글에서는 파이썬으로 구현된 간단한(100줄 이내 ^^) 블록체인을 살펴보면서, 블록체인의 일반적인 구조를 알아보겠습니다:
- 블록체인 클래스 구조
- flask를 이용해 api로 제공하기
블록체인 클래스 구조
먼저, Blockchain이라는 클래스는 아래와 같은 생성자로 만들어 볼 수 있습니다:
import datetime
import hashlib
import json
class Blockchain:
def __init__(self):
self.chain = []
self.create_block(proof = 1, previous_hash = '0')
def create_block(self, proof, previous_hash):
block = {
'index': len(self.chain) + 1,
'timestamp': str(datetime.datetime.now()),
'proof': proof,
'previous_hash': previous_hash
}
self.chain.append(block)
return block
chain이라는 인스턴스 변수에 리스트가 할당되고, create_block 메소드를 호출합니다. create_block은 json으로 구조화된 block을 생성하고 chain이라는 인스턴스 변수에 append 해주고 block을 리턴합니다. 생성자 __init__ 안에서 호출되는 self.create_block은 가장 처음 블록인 제네시스(genesis) 블록을 생성해주기 위해 proof = 1, previous_hash = '0'이라는 아규먼트를 사용합니다.
추가적으로 Block 클래스에는 아래와 같은 get_previous_block, proof_of_work, hash라는 메소드를 가집니다:
def get_previous_block(self):
return self.chain[-1]
def proof_of_work(self, previous_proof):
new_proof = 1
check_proof = False
while check_proof is False:
hash_operation = hashlib.sha256(str(new_proof**2 - previous_proof**2).encode()).hexdigest()
if hash_operation.startswith('0000'):
check_proof = True
else:
new_proof += 1
return new_proof
def hash(self, block):
encoded_block = json.dumps(block, sort_keys = True).encode()
return hashlib.sha256(encoded_block).hexdigest()
각각 아래와 같은 기능을 수행하는데요:
- get_previous_block: chain 리스트에 있는 가장 마지막 블록을 리턴
- proof_of_work: 이전 proof 값을 받고 previous_proof와의 연산 해시 값이 특정 조건을 만족하는 new_proof를 찾아 리턴함
- hash: 딕셔너리 형태의 block을 받아서 json으로 dump하고 인코딩하여 해시값을 얻어 리턴함
위와 같은 클래스를 통해서 아래와 같이 블록을 채굴(mine)할 수 있습니다:
blockchain = Blockchain()
previous_block = blockchain.get_previous_block()
previous_proof = previous_block['proof']
proof = blockchain.proof_of_work(previous_proof)
previous_hash = blockchain.hash(previous_block)
block = blockchain.create_block(proof, previous_hash)
print(blockchain.chain)
#output
[{'index': 1,
'timestamp': '2021-12-14 17:21:13.498696',
'proof': 1,
'previous_hash': '0'},
{'index': 2,
'timestamp': '2021-12-14 17:21:13.500686',
'proof': 533,
'previous_hash': '4fb3563bc622a0e4b3fea0f8c5b3eedd975e3d2f76825eedcb2bc7fbbf7163a8'}]
위의 상태를 이미지로 표현하면, 2개의 블록이 아래와 같은 형태로 연결되어 있다는 사실을 알 수 있습니다:
이러한 블록체인이 규칙에 맞게 생성되어 있는지 확인하기 위해서는 아래의 is_valid_chain과 같은 메소드를 추가할 수 있습니다:
def is_valid_chain(self, chain):
previous_block = chain[0]
block_index = 1
while block_index < len(chain):
block = chain[block_index]
if block['previous_hash'] != self.hash(previous_block):
return False
previous_proof = previous_block['proof']
proof = block['proof']
hash_operation = hashlib.sha256(str(proof**2 - previous_proof**2).encode()).hexdigest()
if not hash_operation.startswith('0000'):
return False
previous_block = block
block_index += 1
return True
검증 방식은 genesis block의 다음 블록부터 마지막 블록까지
- previous_hash 값이 이전 블록의 hash값과 동일한지
- proof값이 previous_proof 값과 연산 후 해시 값 계산 시 특정 조건('0000'으로 시작)을 만족하는지
체크합니다.
flask를 이용해 api로 제공하기
flask를 이용해 위에서 만든 간단한 블록체인을 API로 제공할 수 있습니다.
from flask import Flask, jsonify
app = Flask(__name__)
app.config['JSONIFY_PRETTYPRINT_REGULAR'] = False
blockchain = Blockchain()
@app.route('/mine_block', methods = ['GET'])
def mine_block():
previous_block = blockchain.get_previoous_block()
previous_proof = previous_block['proof']
proof = blockchain.proof_of_work(previous_proof)
previous_hash = blockchain.hash(previous_block)
block = blockchain.create_block(proof, previous_hash)
responses = {
'message': 'Congratulations, you just mined a block!',
**block
}
return jsonify(responses), 200
@app.route('/get_chain', methods = ['GET'])
def get_chain():
response = {
'chain': blockchain.chain,
'length': len(blockchain.chain)
}
return jsonify(response), 200
app.run(host = '0.0.0.0', port = 6000)
이후 Block 클래스와 flask API로 구성된 파일을 실행하고 Postman으로 호출하여 확인해 볼 수 있습니다.
최종 코드 [2]
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import datetime
import hashlib
import json
from flask import Flask, jsonify
class Blockchain:
def __init__(self):
self.chain = []
self.create_block(proof = 1, previous_hash = '0')
def create_block(self, proof, previous_hash):
block = {
'index': len(self.chain) + 1,
'timestamp': str(datetime.datetime.now()),
'proof': proof,
'previous_hash': previous_hash
}
self.chain.append(block)
return block
def get_previous_block(self):
return self.chain[-1]
def proof_of_work(self, previous_proof):
new_proof = 1
check_proof = False
while check_proof is False:
hash_operation = hashlib.sha256(str(new_proof**2 - previous_proof**2).encode()).hexdigest()
if hash_operation.startswith('0000'):
check_proof = True
else:
new_proof += 1
return new_proof
def hash(self, block):
encoded_block = json.dumps(block, sort_keys = True).encode()
return hashlib.sha256(encoded_block).hexdigest()
def is_valid_chain(self, chain):
previous_block = chain[0]
block_index = 1
while block_index < len(chain):
block = chain[block_index]
if block['previous_hash'] != self.hash(previous_block):
return False
previous_proof = previous_block['proof']
proof = block['proof']
hash_operation = hashlib.sha256(str(proof**2 - previous_proof**2).encode()).hexdigest()
if not hash_operation.startswith('0000'):
return False
previous_block = block
block_index += 1
return True
app = Flask(__name__)
app.config['JSONIFY_PRETTYPRINT_REGULAR'] = False
blockchain = Blockchain()
@app.route('/mine_block', methods = ['GET'])
def mine_block():
previous_block = blockchain.get_previous_block()
previous_proof = previous_block['proof']
proof = blockchain.proof_of_work(previous_proof)
previous_hash = blockchain.hash(previous_block)
block = blockchain.create_block(proof, previous_hash)
responses = {
'message': 'Congratulations, you just mined a block!',
**block
}
return jsonify(responses), 200
@app.route('/get_chain', methods = ['GET'])
def get_chain():
response = {
'chain': blockchain.chain,
'length': len(blockchain.chain)
}
return jsonify(response), 200
app.run(host = '0.0.0.0', port = 5000)
Reference