Manage Bor

The script is designed for the maintenance and management of peers in a Bor network, disconnecting peers that do not meet certain version criteria, and blocking traffic from the IPs associated with those peers.

import subprocess
import re
import json
import os

def remove_peer(enode):
    # Command to remove the peer using the Geth console
    command = f'bor attach /var/lib/bor/bor.ipc --exec "admin.removePeer(\'{enode}\')"'
    
    try:
        # Execute the command
        subprocess.run(command, shell=True, check=True)
        print(f"Peer removed: {enode}")
    except subprocess.CalledProcessError as e:
        print(f"Error removing peer {enode}: {e}")

def block_ip(ip_address):
    # Command to block incoming traffic for the IP using iptables
    block_command = f'sudo iptables -A INPUT -s {ip_address} -j DROP'
    os.system(block_command)
    print(f"Traffic from IP {ip_address} blocked using iptables")

# Command to get the list of peers using the Geth console
get_peers_command = 'bor attach /var/lib/bor/bor.ipc --exec "admin.peers"'

def version_compare(v1, v2):
    """
    Compare version strings v1 and v2.
    Returns True if v1 is less than v2.
    """
    return v1 < v2

try:
    # Execute the command and capture the output
    result = subprocess.check_output(get_peers_command, shell=True, text=True)

    # Add double quotes around keys in JSON
    result = re.sub(r'([{,]\s*)([A-Za-z_][a-zA-Z_0-9]*)(\s*:)','\\1"\\2"\\3', result)

    # Load the output as a list of JSON objects
    peers = json.loads(result)

    # Iterate over the found peers
    for peer in peers:
        caps = peer.get('caps', [])
        name_version = peer.get('name', '').split('/')[1] if '/' in peer.get('name', '') else ''
        ip_match = re.search(r'@(\d+\.\d+\.\d+\.\d+):30303', peer['enode'])

        # Check if the peer does not have "snap/1" in its capabilities
        # and if the version of the name is less than "v1"
        if "snap/1" not in caps or version_compare(name_version, 'v1'):
            # Disconnect the peer
            remove_peer(peer['enode'])

            # Block traffic from the IP associated with the enode
            if ip_match:
                ip_address = ip_match.group(1)
                block_ip(ip_address)
        else:
            print(f"Healthy peer: {peer['enode']}")

except subprocess.CalledProcessError as e:
    print(f"Error getting the list of peers: {e}")

Python and Requests Installation:

Make sure you have Python installed on your system. Open the terminal or command prompt. Execute the following command to install the requests library:

pip install requests

Script Execution:

Save the Python script (Your_script.py) in the directory of your choice.

Run the Script:

In the terminal or command prompt, navigate to the directory where the script is located. Execute the script using the following command:

python3 your_script.py

Results:

Iterates over the list of peers obtained from the Geth console. For each peer, it checks whether it has the "snap/1" capability and if the name version is less than "v1". If the above conditions are not met, it removes the peer using admin.removePeer and blocks the traffic from the associated IP using iptables. Note: You can use cron to automate the periodic execution of the script.

Last updated