Symbol Peer ノードのブロック高を取得するスクリプト #Symbol

Symbol Peerノードのブロック高を取得するスクリプトについてまとめました。
以下の記事を参考にやってみました。

参考記事

今回想定している環境

[インターネット上のサーバー]<-->[自宅Wifiルーター]<-->[Symbol peerノード]

前提

ルーターは7900番ポートを開けておく
peerノードに対してポートフォワーディングしておく

概要

インターネット上のサーバーのOSはubuntuとします。
適当なディレクトリを作成します
python3コマンドをインストールしgithubからsymbol-node-configuratorをクローンします。
python3スクリプトを用意します

コマンド

sudo apt-get update
sudo apt-get install python3 python3-pip openssl git

mkdir testdir
cd testdir

openssl genpkey -algorithm ed25519 -outform PEM -out ca.key.pem
git clone https://github.com/symbol/symbol-node-configurator.git
PYTHONPATH=./symbol-node-configurator python3 symbol-node-configurator/certtool.py --working cert --name-ca "my cool CA" --name-node "my cool node name" --ca ca.key.pem
cat cert/node.crt.pem cert/ca.crt.pem > cert/node.full.crt.pem
openssl verify -CAfile cert/ca.crt.pem cert/node.full.crt.pem

vim peer-node-check.py
python3 peer-node-check.py

peer-node-check.py

import socket
import ssl
from pathlib import Path
from symbolchain.core.BufferReader import BufferReader
from symbolchain.core.BufferWriter import BufferWriter

class ChainStatistics:
    def __init__(self):
        self.height = 0
        self.finalized_height = 0
        self.score_high = 0
        self.score_low = 0


    def __str__(self):
        score = self.score_high << 64 | self.score_low
        return '\n'.join([
            f'          height: {self.height}',
            f'finalized height: {self.finalized_height}',
            f'           score: {score}'
        ])


class SymbolPeerClient:
    def __init__(self, host, port, certificate_directory):
        (self.node_host, self.node_port) = (host, port)
        self.certificate_directory = Path(certificate_directory)
        self.timeout = 10

        self.ssl_context = ssl.create_default_context()
        self.ssl_context.check_hostname = False
        self.ssl_context.verify_mode = ssl.CERT_NONE
        self.ssl_context.load_cert_chain(
            self.certificate_directory / 'node.full.crt.pem',
            keyfile=self.certificate_directory / 'node.key.pem')


    def _send_socket_request(self, packet_type, parser):
        try:
            with socket.create_connection((self.node_host, self.node_port), self.timeout) as sock:
                with self.ssl_context.wrap_socket(sock) as ssock:
                    self._send_simple_request(ssock, packet_type)
                    return parser(self._read_packet_data(ssock, packet_type))
        except socket.timeout as ex:
            raise ConnectionRefusedError from ex


    @staticmethod
    def _send_simple_request(ssock, packet_type):
        writer = BufferWriter()
        writer.write_int(8, 4)
        writer.write_int(packet_type, 4)
        ssock.send(writer.buffer)


    def _read_packet_data(self, ssock, packet_type):
        read_buffer = ssock.read()

        if 0 == len(read_buffer):
            raise ConnectionRefusedError(f'socket returned empty data for {self.node_host}')

        size = BufferReader(read_buffer).read_int(4)

        while len(read_buffer) < size:
            read_buffer += ssock.read()

        reader = BufferReader(read_buffer)
        size = reader.read_int(4)
        actual_packet_type = reader.read_int(4)

        if packet_type != actual_packet_type:
            raise ConnectionRefusedError(f'socket returned packet type {actual_packet_type} but expected {packet_type}')

        return reader


    def get_chain_statistics(self):
        packet_type = 5
        return self._send_socket_request(packet_type, self._parse_chain_statistics_response)


    @staticmethod
    def _parse_chain_statistics_response(reader):
        chain_statistics = ChainStatistics()

        chain_statistics.height = reader.read_int(8)
        chain_statistics.finalized_height = reader.read_int(8)
        chain_statistics.score_high = reader.read_int(8)
        chain_statistics.score_low = reader.read_int(8)

        return chain_statistics


peer_client = SymbolPeerClient('自宅のIPアドレス', 7900, '/home/xxxxx/testdir/cert')
chain_statistics_peer = peer_client.get_chain_statistics()
print(chain_statistics_peer)

※接続先のIPアドレスとディレクトリまでのパスは適宜書き換えてください
※なんか環境によってはうまくいかないかも…
※symbol-bootrtrapインストール済み環境では動き、まっさらのUbuntuに実行してもエラーに遭遇することは確認しました。
コアデブ情報

Symbol Peer ノードのブロック高を取得するスクリプト

参考記事

今回想定している環境

前提

概要

関連する記事