ABOUT ME

-

Today
-
Yesterday
-
Total
-
  • 파이썬으로 간단한 블록체인 만들기
    Cryptography 2021. 12. 13. 23:20
    반응형

    블록체인은 블록이라고 불리는 것들이 암호기법을 통해 체인처럼 연결된 것 [1]을 말합니다.

     

    흔히 아래와 같은 형태로 표현되는데요:

     

    Abstract blockchain - Image from Author

     

    위의 정의에서 '암호기법을 통해 연결된다'와 같은 부분들은 실제로 간단한 블록체인을 만들어보면 복잡한 설명이 필요없이 쉽게 이해가 됩니다. 이번 글에서는 파이썬으로 구현된 간단한(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개의 블록이 아래와 같은 형태로 연결되어 있다는 사실을 알 수 있습니다:

    2 Blocks in a chain - Image from Author

     

    이러한 블록체인이 규칙에 맞게 생성되어 있는지 확인하기 위해서는 아래의 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으로 호출하여 확인해 볼 수 있습니다.

     

     

    GET /mine_block - Image from Author

     

    GET /get_chain - Image from Author

    최종 코드 [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

    [1] https://en.wikipedia.org/wiki/Blockchain

    [2] https://www.udemy.com/course/build-your-blockchain-az/

    반응형
Kaden Sungbin Cho