rostrum 14.0.1

An efficient implementation of Electrum Server with token support
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

import asyncio
import random
import secrets
import socket
import psutil

# Don't assign rpc or p2p ports lower than this
PORT_MIN = 1024
PORT_MAX = 65535

mapped_ports = {}


async def get_port(n, service):
    key = f"{service}:{n}"
    mapped = mapped_ports.get(key)
    if mapped is not None:
        return mapped

    port = await get_random_unused_port()
    print(f"providing port {port} to {service}:{n}")
    mapped_ports[key] = port
    return port


async def try_bind_port(port: int) -> bool:
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(("localhost", port))
            return True
    except OSError:
        return False


async def is_port_in_use(port: int) -> bool:
    async def check_port():
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            return s.connect_ex(("localhost", port)) == 0

    if await check_port():
        # Random sleep and check again to mitigate race conditions
        await asyncio.sleep(random.uniform(0, 3))
        return await check_port()

    return False


taken_ports = []


async def get_random_unused_port():
    # pylint: disable=global-statement
    global taken_ports
    if len(taken_ports) == 0:
        # first time function is called

        # seed randomness
        seed = int.from_bytes(secrets.token_bytes(16), "big")
        random.seed(seed)

        # try to prefill taken_ports
        try:
            connections = psutil.net_connections()
            taken_ports = {
                conn.laddr.port for conn in connections if conn.status == "LISTEN"
            }
        except psutil.AccessDenied:
            # don't have privileges
            taken_ports = set()
        except Exception as e:
            print(f"Unexpected exception: {e}")
            taken_ports = set()

    while True:
        port = random.randint(PORT_MIN, PORT_MAX)
        if port in taken_ports:
            continue
        taken_ports.add(port)
        if await try_bind_port(port):
            return port


async def p2p_port(n):
    return await get_port(n, "p2p")


async def rpc_port(n):
    return await get_port(n, "rpc")


async def electrum_rpc_port(n):
    return await get_port(n, "electrum_rpc")


async def electrum_ws_port(n):
    return await get_port(n, "electrum_ws")


async def electrum_http_port(n):
    return await get_port(n, "electrum_http")


async def electrum_monitoring_port(n):
    return await get_port(n, "electrum_monitoring")


async def electrum_rpc_ssl_port(n):
    return await get_port(n, "electrum_rpc_ssl")


async def electrum_ws_ssl_port(n):
    return await get_port(n, "electrum_ws_ssl")


async def electrum_https_port(n):
    return await get_port(n, "electrum_https")