P2P 네트워크 / 블록체인 주고받기

1/ P2P 네트워크란?

Peer To Peer

기존에 배웠던 서버-클라이언트로 구분된 통신이 아닌 노드-노드간의 통신으로 블록체인의 탈중앙화 개념과 일맥상통하다.

한 컴퓨터가 서버가 되기도 하고 클라이언트가 되기도 하기 때문에, 이전에 frt/back을 나눠서 코드를 작성해준 것과 달리 서버쪽 코드와 클라이언트쪽 코드를 같이 한 파일 안에서 작성해줌으로써 P2P 네트워크를 구축한다.

 

2/ P2P 네트워크 구축하기

1. BlockChain 클래스 정의 

우선, 기존에 만들어둔 Chain 클래스를 사용해서 블록체인 클래스를 새로 정의해준다. 

해당 블록체인 클래스를 사용해서 클래스 내부의 메소드를 사용해 블록체인 내용을 조회해보는 api 나 블록 채굴 api를 서버 쪽에 만들어줄 수 있다.

// src/core/index.ts

import { Chain } from "./blockchain/chain";

export class BlockChain {
  public chain: Chain;
  constructor() {
    this.chain = new Chain();
  }
}

// index.ts (최상단 루트디렉토리)

import { BlockChain } from "@core/index";
import express from "express";

const app = express();
const bc = new Blockchain();

app.use(express.json());

app.get("/", (req, res) => {
  res.send("yjchain");
});

// 블록채굴 api
app.post("/mineBlock", (req, res) => {
  const { data } = req.body;
  const newBlock = bc.chain.addBlock(data);
  if (newBlock.isError) return res.status(500).json(newBlock.error);
  console.log(newBlock);
  res.json(newBlock.value);
});

// 블록체인 조회 api
app.get("/chains", (req, res) => {
  res.json(bc.chain.getChain());
});

app.listen(3000, () => {
  console.log("서버 시작");
});

 

 

2. 기본 http서버와 ws를 사용한 P2P서버 세팅

항상 일반적으로 쓰는 express 서버 즉, http서버를 기본서버로 하고, P2P 네트워크를 위해 websocket을 사용한다.

 

 - express 를 사용한 http 서버를 구축하고 이 http서버가 실행될 때 P2P 서버 또한 실행되게끔 ws.listen() 을 추가해줬다.

해당 ws(P2PServer 클래스의 인스턴스)의 listen 메서드는 P2PServer 클래스의 메서드로 서버를 시작하는 코드이고 아래서 설명한다.

// index.ts (최상단 루트디렉토리)

import express from "express";
import { P2PServer } from "./src/serve/p2p";

const app = express();
const ws = new P2PServer();

app.use(express.json());

app.get("/", (req, res) => {
  res.send("yjchain");
});

app.post("/addToPeer", (req, res) => {
  const { peer } = req.body;
  ws.connectToPeer(peer);
});

app.listen(3000, () => {
  console.log("서버 시작");
  ws.listen();
});

 

- P2PServer 클래스

나에게 요청/응답오는 소켓들을 담기 위해 sockets 라는 웹소켓을 담는 배열을 속성으로 가진다. 

// src/serve/p2p.ts

import { WebSocket } from "ws";
import { Chain } from "@core/blockchain/chain";

export class P2PServer extends Chain {
	public sockets: WebSocket[];
    
    constructor() {
    	super();
        this.sockets = [];
    }
    // 연결된 socket들 가져오는 메소드
    getSockets() : {
    	return this.sockets;
    }
    
    // 서버시작 실행코드
    listen() {
    	const server = new WebSocket.Server({ port: 7545 });
        server.on("connection", (socket) => {
            console.log("ws connection");
            this.connectSocket(socket);
        });
    }
    
    // 클라이언트 연결코드
    connectToPeer(newPeer: string) {
    	const socket = new Websocket(newPeer);
        socket.on("open", () => {
             this.connectSocket(socket);
        })
    }
}

 

listen() 메서드 : 서버 역할의 코드

해당 코드에선 7545번 포트에서 WebSocket Server가 연결 요청을 받을 수 있는 상태로 만들어준 것이고, 클라이언트 입장의 다른 노드가 소켓연결 요청을 보내면 그로 인해 connection 핸드쉐이크가 일어난다. 이 때 클라이언트 정보가 담긴 socket을 받아 클래스 내부의 connectSocket() 메서드가 실행된다.

 

connectToPeer : 클라이언트 역할의 코드

해당 인자로 들어오는 newPeer는 '/addToPeer' 라우터에서 post요청으로 넘겨준 req.body에 담긴 ip 주소이다.

핸드쉐이크 ok -> open 이벤트가 발동, 이 ip 주소를 가지고 있는 서버 입장의 노드와 웹소켓이 연결되고, 해당 요청을 받은 그 ip주소의 노드는 앞서 만들어둔 서버 입장에서의 listen 메소드를 실행한다. 

 

 

3. connectSocket 메서드 

해당 메소드는 위의 서버(connection)와 클라이언트(open) 역할의 연결 코드에서 동시에 존재했다.

즉, 해당 connectSocket 메소드 내부에도 서버쪽, 클라이언트쪽 코드가 동시에 존재한다.

// src/serve/p2p.ts 의 P2PServer 클래스 내부 메서드

connectSocket(socket: WebSocket) {
    this.sockets.push(socket);
    this.messageHandler(socket);
    
    const data: Message = {
       type: MessageType.latest_block,
       payload: [],
    };
    
    this.send(socket)(data)
    // const send = this.send(socket);
    // send(data);
}

// 연결된 소켓에 Message 타입 데이터 전달
send(_socket: WebSocket) {
    return (_data: Message) => {
    	_socket.send(JSON.stringify(_data))
    }
}

this.sockets.push(socket)

: P2PServer의 속성인 sockets 웹소켓배열에 들어가는 socket은 위의 각 연결 메소드에서 인자로 받은 소켓으로, 클라이언트 입장에선 서버쪽 소켓정보, 서버 입장이면 클라이언트쪽 소켓정보를 추가해주는 거다.

 

나중에 broadcasting 을 하기 위해서 이렇게 연결된 소켓 정보들을 배열에 담는다고 한다.

broadcasting은 간단히 설명하면, 한 P2P 네트워크 안에 있는 각 소켓들이 각자 블럭을 마이닝하게 되면 체인이 업데이트되는데, 이 업데이트된 체인 정보를 모두가 같이 공유해야하기 때문이다. 

 

여기서 각 소켓에 Message 데이터 전달을 위해 send 고차함수를 만들어주었다. 

Message 는 아래와 같은 형태이며, 소켓통신 시 이벤트 구분을 위해서 미리 만들어놓은 인터페이스이다.

MessageType은 딱 정해져있기 때문에 그냥 변수가 아닌 enum으로 정의해주었다. 이렇게 하면, Message 작성 시 다른 MessageType이 들어가는걸 미리 막아주고, 코드 자동완성으로 좀 더 수월하게 코딩할수 있다.

(다만 enum은 declare가 되지 않더라..)

enum MessageType {
  latest_block = 0,
  all_block = 1,
  rcvdChain = 2,
}

interface Message {
  type: MessageType;
  payload: any;
}

 

4. messageHandler

socket.on("message", cb) : message 이벤트 즉, 메시지/데이터를 전달받았을 때 콜백함수를 실행한다.

해당 이벤트는 어쨌든 connectSocket을 각 클라이언트/서버 코드 쪽에 심어두었으므로, 두 쪽 다 데이터를 전달하고, 받아볼수도 있는 것이다.

 

전체적으로 흐름을 정리하자면, 아래와 같다.

 

1. 클라이언트 역할의 노드(이하 클노드)가 서버역할의 노드(이하 서노드)에게 소켓 연결을 시도 (open)

2. 서노드는 연결을 받아서 (connection), connectSocket() 실행 + this.send(socket)(data) 발동하여 데이터를 클노드에게 전달

3. 이때 전달한 Message의 MessageType은 latest_block 즉, 가장 최신 블락을 클노드에게 '요청' 한것

    (그래서 서노드가 보내는 맨 처음 data(Message)의 payload 가 null값이었다. 우선 요청만 때린 것)

4. 클노드는 해당 MessageType이 latest_block 인 switch 문을 발동

   -> 요청메시지에 맞게 payload로 가장 최근 블럭을 던져주고, 동시에 MessageType으로는 all_block을 서노드에게 요청

5. 서노드는 all_block 요청을 받고 가지고 있는 블록체인을 클노드에게 payload로 전달

    이 때, 서노드가 클노드에게 자신의 체인을 전달하기 전에!!

   서노드는 클노드에게서 받은 최신 블럭([this.getLatestBlock])을 자신의 블록체인에 추가할지 말지 검증을 거쳐 결정해야한다. 

6-1. 블록검증을 거쳐 블록이 맞다면 블록체인에 추가 (addToChain())

6-2. 블록검증에서 실패해서 클노드의 latest_block이 체인에 추가되지 못했을 경우, 클노드에게 payload로 서노드의 전체 블록체인과 MessageType rcvdChain을 전달해준다. 

7. 클노드는 서노드로부터 가장 최신 블록체인과 함께 MessageType이 rcvdChain인 Message를 받고, 받은 체인을 검증하여 조건에 맞는 체인이라면, 내가 가지고 있던 기존의 블럭체인과 바꿔치기한다. (handleChainResponse) 여기서 현재 같은 P2P 네트워크 상에 연결된 모든 소켓들에게 이를 broadcast해줘야하는데, 이부분은 다음 포스팅에서 정리할 예정이다.  

 

 // src/serve/p2p.ts
 
 messageHandler(socket: WebSocket) {
       const cb = (data: string) => {
       const result: Message = P2PServer.dataParse<Message>(data);
       // 위에서 다른 노드가 보낸 type, payload를 담은 Message
       const send = this.send(socket);
       
       switch(result.type){
       	 case MessageType.latest_block: {
            const message: Message = {
               type: MessageType.all_block,
               payload: [this.getLatestBlock()]; 
                // P2PServer 클래스는 Chain 클래스를 extends 했기 때문에 Chain쪽 메소드를 쓸 수 있다. 
            }
            send(message);
            break;
         }
         case MessageType.all_block: {
           const message: Message = {
               type: MessageType.rcvdChain,
               payload: this.getChain(); 
            }
            // 블럭검증코드 이후 체인에 블럭을 추가할지말지 결정한다.
            const [rcvdBlock] = result.payload; // [this.getLatestBlock()]
            const isValid = this.addToChain(rcvdBlock);
           // addToChain에 성공했을 경우 굳이 전체 체인을 요청할 필요가 없으니 send를 실행시키지 않고 끝낸다.
            if (!isValid.isError) break;
            send(message);
            break;
         }
         case MessageType.rcvdChain: {
           const rcvdChain: IBlock[] = result.payload; 
           console.log(rcvdChain);
           // 체인을 통으로 교체하는 코드, 내꺼보다 긴 체인 선택해서 교체
           this.handleChainResponse(rcvdChain);
           break;
         }
       }
    }
    socket.on("message", cb);
 }

socket으로 받은 데이터는 Buffer 형태이므로 이를 다시 원래 형태로 바꿔주는 함수가 필요하다. 

위에서 이 dataParse 함수를 써서 원래 Message 형태로 돌려준다. 

  static dataParse<T>(data: string): T {
    return JSON.parse(Buffer.from(data).toString());
  }

 

addToChain(rcvdBlock)

: 블럭을 검증하여 내 체인에 추가할지말지 결정하는 코드 

 

다른 노드로부터 받은 블록을 내 체인에 넣을지 말지 결정하는 메서드를 만들어줌

- 우선 새로 받은 블럭이 유효한지 검사부터 해야함 : 이미 Block 클래스에서 블럭 검증 코드를 넣어두었으니 끌어다 쓴다. 

 isValidNewBlock() 코드 : https://yjleekr.tistory.com/76

- 유효하다면 서버역할 노드가 가지고 있는 체인에 넣어주고 여기서 messageHandler는 종료된다. 

- 유효하지 않다면, 이젠 전체 체인을 payload에 실어서 다시 클라이언트쪽 노드에게 전달해준다. 

// src/core/blockchain/chain.ts/Chain 클래스

  public addToChain(_rcvdBlock: Block): Failable<undefined, string> {
    const isValid = Block.isValidNewBlock(_rcvdBlock, this.getLatestBlock());
    if (isValid.isError) return { isError: true, error: isValid.error };
    this.blockchain.push(_rcvdBlock);
    return { isError: false, value: undefined };
  }

handleChainResponse()

: 서버노드로부터 업데이트 받은 체인을 검사해서 교체하고 broadcast로 같은 네트워크 상의 노드들에게 던져주는 코드

 

1/ 다른 노드로부터 받은 체인이 유효한지 검사하는 메소드인 isValidChain() 와 체인을 바꾸고 체인 대체 성공 여부를 반환하는 replaceChain()은 어쨌든 Chain과 관련된 메소드이므로 Chain 클래스로 빼주었다. 아래서 정리한다.

 

2/ 결국 체인을 교체하는 데 성공했으면 broadcast로 소켓으로 연결된 모든 노드들에게 이 최신 체인을 업데이트해준다.  

// src/serve/p2p.ts

public handleChainResponse(rcvdChain: IBlock[]): Failable<Message | undefined, string> {
	// 서버역할 노드로부터 받은 체인이 유효한가 검사
    const isValidChain = this.isValidChain(rcvdChain); // 내 체인 vs 상대노드 체인
    if(isValidChain.isError) return { isError: true, error: isValidChain.error };
    
    // 체인 바꾼 결과
    const isValid = this.replaceChain(rcvdChain)
    if(isValid.isError) return { isError: true, error: isValid.error };
    
    // 체인 바꾼 게 성공했으면 broadcast로 던져주기
    const msg: Message = {
    	type: MessageType.rcvdChain,
        payload: rcvdChain,
    };
    this.broadcast(msg);
    return { isError: false, value: undefined }
}

 

isValidChain() 

: 다른 노드로부터 받은 체인을 검사한다.

사실 여기서 제네시스 블록 자체가 맞는지에 대한 검증도 진행해야한다. 왜냐면 같은 제네시스여야만 같은 블록체인에 들어올 수 있기 때문이다. 하지만 이번엔 이 과정은 생략한다. 

 

검사 방식은 체인 내에서 블록을 하나하나 전 블록과 비교하는 Block.isValidNewBlock() 메소드를 사용해준다. 

다만, 위에서 제네시스 블록을 검사하는 코드를 넣어줄 것이기 때문에, 블록 검사는 height 0번째인 제네시스 블록부터가 아닌 그 다음블록인 height 1번째부터 실시한다. 

// src/core/blockchain/chain.ts

public isValidChain(_chain: Block[]): Failable<undefined, string> {
    // 제네시스 블록을 검사하는 코드가 들어가야한다
	// const genesis = _chain[0]
    
    // 이제 체인 내 나머지 블럭 검사하는 코드 시작
    for(let i=1; i< _chain.length; i++){
    	const newBlock = _chain[i];
        const previousBlock = _chain[i-1];
        const isValid = Block.isValidNewBlock(newBlock, previousBlock);
        if (isValid.isError) return { isError: true, error: isValid.error };
    }
    
    return { isError: false, value: undefined };
}

 

replaceChain()

isValidChain()으로 다른 노드로부터 받은 체인의 유효성을 검사한 이후에, 이젠 기존에 가지고 있던 블록체인과 다른 노드로부터 받은 체인을 바꿀지 말지 결정해서 조건에 맞는다면 바꿔주는 코드를 작성해준다. 

 

내 블록체인 vs. 상대 노드의 블록체인 비교검사

아래의 조건에 걸린다면 상대 노드의 블록체인과 바꾸지않고 리턴해버린다.

 

1/ 받은 체인의 길이가 0일 때

2/ 받은 체인의 최신블럭 height <= 내 체인의 최신블럭 height 일때

3/ 받은 체인의 최신블럭 이전 해시 === 내 체인의 해시값 일때 

 

3번째 조건은 사실 왜 넣나 의문이었는데, 이미 앞의 검증과정으로 인해 저 조건문까지 가지않을뿐더러,  우선 이미 앞의 addToChain() 함수에서 isValidNewBlock()으로 이미 검증을 거쳐 체인에 포함되었을 것이기도 하고, 나중에 다른데서도 쓰인다고 하니 우선 대충 이해하고 넘어간다..

// src/core/blockchain/chain.ts

public replaceChain(rcvdChain: Block[]): Failable<undefined, string> {
    const latestRcvdBlock: Block = rcvdChain[rcvdChain.length - 1];
    const latestBlock: Block = this.getlatestBlock();
    
    if (latestRcvdBlock.height === 0) {
      return { isError: true, error: "받은 최신블록이 제네시스 블록입니다." };
    }
    if (latestRcvdBlock.height <= latestBlock.height) {
      return { isError: true, error: "보유한 블록체인이 받은 체인보다 더 길거나 같습니다." };
    }
    if (latestRcvdBlock.previousHash === latestBlock.hash) {
      // addToChain()
      return { isError: true, error: "블록이 하나 모자릅니다." };
    }
    
    // 체인 바꿔주는 코드
    this.blockchain = rcvdChain;
    return { isError: false, value: undefined }
}

5. broadcast

받은 Message (type과 payload를 가진)를 한 네트워크 안에 있는 모든 소켓들(노드들)에게 보내주는 함수이다.

// src/serve/p2p.ts

 broadcast(message: Message) {
    this.sockets.forEach((socket) => this.send(socket)(message));
  }
  
 send(_socket: WebSocket) {
    return (_data: Message) => {
      _socket.send(JSON.stringify(_data));
    };
  }

 

6. errorHandler

특정 소켓이 나가거나 다른 소켓에러가 발생했을 시 그 소켓을 아예 P2P네트워크에서 없애버리는 방식으로 예외처리를 해준다.

 errorHandler(socket: WebSocket) {
    const close = () => this.sockets.splice(this.sockets.indexOf(socket), 1);
    socket.on("close", close);
    socket.on("error", close);
  }

해당 에러핸들러는 connectSocket 함수 상에서 호출한다.

// src/serve/p2p.ts

connectSocket(socket: WebSocket) {
    this.sockets.push(socket);
    this.messageHandler(socket);

    const data: Message = {
      type: MessageType.latest_block,
      payload: [],
    };

    this.errorHandler(socket);
    this.send(socket)(data);
  }

 

같은 소켓으로 연결된 노드의 아이피를 알고 싶다면 아래처럼 해보자

// index.ts

app.get("/peers", (req, res) => {
  const sockets = ws.getSockets().map((s: any) => s._socket.remoteAddress + ":" + s._socket.remotePort);
  res.json(sockets);
});