파이썬으로 간단한 블록체인 구현해보기

지금까지 배운 블록체인 개념을 파이썬으로 구현해보고자 하였습니다. 

 

아래 링크를 참고하여 만들었습니다. 

https://hackernoon.com/learn-blockchains-by-building-one-117428612f46

Learn Blockchains by Building One | Hacker Noon

Step 1 : Building a Blockchain

1. 블록체인 기본 구조

블록체인의 기본 뼈대를 만들었습니다. 새로운 블록 그리고 그 안에 새로운 트랜잭션(거래), 위조를 방지하는 해시, 그리고 마지막 블록 함수를 정의했습니다 .

class Blockchain(object):
    def __init__(self):
        self.chain = []
        self.current_transactions = []
        
    def new_block(self): 
        # 새로운 블록 만들어서 블록체인에 넣기 
        pass
    
    def new_transaction(self):
        # 새로운 트랜잭션 만들어서 트랜잭션 리스트에 넣기 
        pass
    
    @staticmethod
    def hash(block):
        # 블록의 해시 
        pass

    @property
    def last_block(self):
        # chian의 마지막 block return
        pass

2. 블록의 기본 구조

다음은 블록 1개의 예시입니다

# 블록 1개 기본 예시
block = {
    'index': 1,
    'timestamp': 1506057125.900785, 
    'transactions': [
        {
            'sender': "8527147fe1f5426f9dd545de4b27ee00",
            'recipient': "a77f5cdfa2934df3954a5c7c7da5df1f",
            'amount': 5,
        } #거래내역
    ],
    'proof': 324984774000,
    'previous_hash': "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" #이전의 해시값 저장 
}

3. 블록에 새로운 트랜잭션 추가

이제 만들어진 블록에 새로운 트랜잭션을 추가해보겠습니다 

• param : 보내는 사람, 받는 사람의 주소, 돈의 양
• return : 마지막 블록의 다음 블록 반환

class Blockchain(object):
    ...
    
    def new_transaction(self, sender, recipient, amount):
        """
        새롭게 채굴된 블록에 들어갈 새로운 트랜잭션 생성
        :param sender: <str> Sender의 Address
        :param recipient: <str> Recipient의 Address
        :param amount: <int> Amount
        :return: <int>  트랜잭션이 받을 블록의 인덱스 
        """
        self.current_transactions.append({
            'sender': sender,
            'recipient': recipient,
            'amount': amount,
        })

        return self.last_block['index'] + 1

4. 새로운 블록 만들기

이제 새로운 블록을 만들어보겠습니다. 

일단, 최초의 블록(genesis block)을 만들고, 새로운 블록을 만들어 그 안에 트랜잭션을 넣습니다. 

import hashlib
import json
from time import time


class Blockchain(object):
    def __init__(self):
        self.current_transactions = []
        self.chain = []

        # 최초의 블록 생성 
        self.new_block(previous_hash=1, proof=100)

    def new_block(self, proof, previous_hash=None):
        """
        블록체인에 새로운 블록을 생성한다
        :param proof: <int> PoW 알고리즘에서 받는 Proof
        :param previous_hash: (Optional) <str> 이전 블록 해시 
        :return: <dict> 새로운 블록
        """

        block = {
            'index': len(self.chain) + 1,
            'timestamp': time(),
            'transactions': self.current_transactions,
            'proof': proof,
            'previous_hash': previous_hash or self.hash(self.chain[-1]),
        }

        # 현재 트랙잭션 list reset 
        self.current_transactions = []

        self.chain.append(block)
        return block

    def new_transaction(self, sender, recipient, amount):
        """
        새롭게 채굴된 블록에 들어갈 새로운 트랜잭션 생성
        :param sender: <str> Sender의 Address
        :param recipient: <str> Recipient의 Address
        :param amount: <int> Amount
        :return: <int>  트랜잭션이 받을 블록의 인덱스 
        """
        self.current_transactions.append({
            'sender': sender,
            'recipient': recipient,
            'amount': amount,
        })

        return self.last_block['index'] + 1

    @property
    def last_block(self):
        return self.chain[-1]

    @staticmethod
    def hash(block):
        """
        블록의 SHA-256 hash 생성
        :param block: <dict> Block
        :return: <str>
        """
        
        # Dictionary가 ordered 되어야함. 안그러면 hash가 inconsistent
        block_string = json.dumps(block, sort_keys=True).encode()
        return hashlib.sha256(block_string).hexdigest()

cf) PoW(Proof of Work)

이제 합의 알고리즘 중 하나인 Proof of Work에 대해서 간단하게 알아보겠습니다. 

• 작업증명의 원리 : 어떤 정수 x와 다른 수 y를 곱해서 "0"으로 끝나야 한다고 결정되면 끝이 0이 되는 수를 찾는 것!

from hashlib import sha256
x = 5
y = 0  # 아직 y가 뭔지 모름 
while sha256(f'{x*y}'.encode()).hexdigest()[-1] != "0":
    y += 1
print(f'The solution is y = {y}')

6. PoW 구현

PoW에서 난이도(difficulty)란 hash를 검증할 때 필요한 조건의 난이도를 의미한다. 

difficulty : hash의 앞에 4자리가 '0000'인가?

import hashlib
import json

from time import time
from uuid import uuid4


class Blockchain(object):
    ...
        
    def proof_of_work(self, last_proof):
        """
        간단한 PoW 알고리즘 
         - hash가 앞에 '0000'을 가지고 있는 p'을 찾아라. 
         - p : previous proof, p' : new proof
        :param last_proof: <int>
        :return: <int>
        """
        proof = 0
        while self.valid_proof(last_proof, proof) is False:
            proof += 1

        return proof

    @staticmethod
    def valid_proof(last_proof, proof):
        """
        Proof 검증 : hash(last_proof, proof)가 앞에 '0000'을 가지고 있나? 
        :param last_proof: <int> Previous Proof
        :param proof: <int> Current Proof
        :return: <bool> True if correct, False if not.
        """
        guess = f'{last_proof}{proof}'.encode()
        guess_hash = hashlib.sha256(guess).hexdigest()
        return guess_hash[:4] == "0000"

Step 2: Our Blockchain as an API

/transactions/new : 블록에 새로운 트랜잭션 생성

2. /mine : 서버에게 새로운 블록 mining 알려줌

3. /chain: Full Blockchain return

 

0. 플라스크에 대한 이해

route : 플라스크 서버로 요청이 들어왔을 때 어떤 함수를 호출할 것인지 조절함.

 - 127.0.0.1 : 인터넷 루프백 주소

 - :5000 -> 웹서버가 실행중인 프로토콜 번호 (플라스크 : 5000)

from flask import Flask
# 플라스크 인스턴스 생성 
app = Flask(__name__)

@app.route('/') 
# @ : 데코레이터 
# flask에서는 @가 url 연결에 활용된다. 

def hello() -> str: # -> :  str로 나온다! 
    return "Hello World from Flask" 

app.run()

1. API 뼈대 만들기

import hashlib
import json
from textwrap import dedent
from time import time
from uuid import uuid4

from flask import Flask


class Blockchain(object):
    ...

# 노드 시작 
app = Flask(__name__)

# 이 노드의 고유 번호 생성 
node_identifier = str(uuid4()).replace('-', '')

# 블록체인 시작 
blockchain = Blockchain()

# Create the /mine endpoint, which is a GET request.
@app.route('/mine', methods=['GET'])
def mine():
    return "We'll mine a new Block"

# Create the /transactions/new endpoint, which is a POST request, since we’ll be sending data to it.
@app.route('/transactions/new', methods=['POST'])
def new_transaction():
    return "We'll add a new transaction"

# Create the /chain endpoint, which returns the full Blockchain.
@app.route('/chain', methods=['GET'])
def full_chain():
    response = {
        'chain': blockchain.chain,
        'length': len(blockchain.chain),
    }
    return jsonify(response), 200

# Runs the server on port 5000.
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

2. The Transactions Endpoint

import hashlib
import json
from textwrap import dedent
from time import time
from uuid import uuid4

from flask import Flask, jsonify, request

...

@app.route('/transactions/new', methods=['POST'])
def new_transaction():
    values = request.get_json()

    # Check that the required fields are in the POST'ed data
    required = ['sender', 'recipient', 'amount']
    if not all(k in values for k in required):
        return 'Missing values', 400

    # 새로운 트랜잭션 생성 
    index = blockchain.new_transaction(values['sender'], values['recipient'], values['amount'])

    response = {'message': f'Transaction will be added to Block {index}'}
    return jsonify(response), 201

3. The Mining Endpoint

• PoW 계산
• 트랜잭션 추가한 miner에게 코인 제공
• chain에 새로운 블록 넣기

import hashlib
import json

from time import time
from uuid import uuid4 #실행될 때마다 다른값 반환하는 모듈 

from flask import Flask, jsonify, request

...

@app.route('/mine', methods=['GET'])
def mine():
    #다음 proof 를 얻기 위해 PoW 알고리즘 돌림 
    last_block = blockchain.last_block
    last_proof = last_block['proof']
    proof = blockchain.proof_of_work(last_proof)

    # Proof 찾은것에 대한 보상이 필요 
    # sender="0" : 이 노드는 새로운 coin을 받았다고 명시.
    blockchain.new_transaction(
        sender="0",
        recipient=node_identifier,
        amount=1,
    )

    # chain에 트랜젹션 넣고 new block forge 
    previous_hash = blockchain.hash(last_block)
    block = blockchain.new_block(proof, previous_hash)

    response = {
        'message': "New Block Forged",
        'index': block['index'],
        'transactions': block['transactions'],
        'proof': block['proof'],
        'previous_hash': block['previous_hash'],
    }
    return jsonify(response), 200

Step 3: Consensus(합의)

이는 새로운 노드를 추가할 때, 이를 결정하는 합의 알고리즘에 관핸 내용이다. 

1. 새로운 노드 추가

일단, 노드 리스트에 새로운 노드를 추가한다. 

...
from urllib.parse import urlparse
...


class Blockchain(object):
    def __init__(self):
        ...
        self.nodes = set() #중복 허용 X
        ...

    def register_node(self, address):
        """
        노드 리스트에 새로운 노드 추가 
        :param address: <str> 노드 주소. Eg. 'http://192.168.0.5:5000'
        :return: None
        """

        parsed_url = urlparse(address)
        self.nodes.add(parsed_url.netloc)

2. 합의 알고리즘 시행

블록체인이 타당한지 아닌지 검증한다. 

...
import requests


class Blockchain(object)
    ...
    
    def valid_chain(self, chain):
        """
        블록체인이 타당한지 결정 
        :param chain: <list> A blockchain
        :return: <bool> T / F
        """

        last_block = chain[0]
        current_index = 1

        while current_index < len(chain):
            block = chain[current_index]
            print(f'{last_block}')
            print(f'{block}')
            print("\n-----------\n")
            # hash block이 맞는지 확인한다. 
            if block['previous_hash'] != self.hash(last_block):
                return False

            # PoW가 잘 작동하는지 확인한다. 
            if not self.valid_proof(last_block['proof'], block['proof']):
                return False

            last_block = block
            current_index += 1

        return True

    def resolve_conflicts(self):
        """
        합의 알고리즘 : 가장 긴것 채택 
        :return: <bool> True if our chain was replaced, False if not
        """

        neighbours = self.nodes
        new_chain = None

        # 우리 체인보다 긴것만 봄 
        max_length = len(self.chain)
        
        # 우리 네트워크의 모든 노드들 다 검증 
        for node in neighbours:
            response = requests.get(f'http://{node}/chain')

            if response.status_code == 200:
                length = response.json()['length']
                chain = response.json()['chain']

                # 길이가 긴지 보고, chain이 타당한지 확인 
                if length > max_length and self.valid_chain(chain):
                    max_length = length
                    new_chain = chain

        # 새로운 타당한 긴 체인을 찾으면 이로 대체함. 
        if new_chain:
            self.chain = new_chain
            return True

        return False

3. As API

@app.route('/nodes/register', methods=['POST'])
def register_nodes():
    values = request.get_json()

    nodes = values.get('nodes')
    if nodes is None:
        return "Error: Please supply a valid list of nodes", 400

    for node in nodes:
        blockchain.register_node(node)

    response = {
        'message': 'New nodes have been added',
        'total_nodes': list(blockchain.nodes),
    }
    return jsonify(response), 201


@app.route('/nodes/resolve', methods=['GET'])
def consensus():
    replaced = blockchain.resolve_conflicts()

    if replaced:
        response = {
            'message': 'Our chain was replaced',
            'new_chain': blockchain.chain
        }
    else:
        response = {
            'message': 'Our chain is authoritative',
            'chain': blockchain.chain
        }

    return jsonify(response), 200