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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
# Copyright 2014-2022 Vincent Texier <vit@free.fr>
#
# DuniterPy is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# DuniterPy is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
from typing import Union
import jsonschema
from duniterpy.api import bma, ws2p
from duniterpy.api.client import Client, WSConnection
from duniterpy.api.endpoint import BMAEndpoint, SecuredBMAEndpoint, WS2PEndpoint
from duniterpy.constants import G1_CURRENCY_CODENAME
from duniterpy.documents.ws2p.messages import Ack, Connect, Ok
from duniterpy.key import SigningKey
def handshake(
ws: WSConnection, signing_key: SigningKey, currency: str = G1_CURRENCY_CODENAME
):
"""
Perform ws2p handshake on the web socket connection using the signing_key instance
:param ws: Web socket connection instance
:param signing_key: SigningKey instance
:param currency: Currency codename (default=constants.CURRENCY_CODENAME_G1)
:return:
"""
# START HANDSHAKE #######################################################
logging.debug("\nSTART HANDSHAKE...")
connect_document = Connect(signing_key.pubkey, currency=currency)
connect_message = connect_document.get_signed_json(signing_key)
logging.debug("Send CONNECT message")
ws.send_str(connect_message)
loop = True
remote_connect_document = None
# Iterate on each message received...
while loop:
data = ws.receive_json()
if "auth" in data and data["auth"] == "CONNECT":
jsonschema.validate(data, ws2p.network.WS2P_CONNECT_MESSAGE_SCHEMA)
logging.debug("Received a CONNECT message")
remote_connect_document = Connect(
data["pub"],
challenge=data["challenge"],
signature=data["sig"],
currency=currency,
)
logging.debug("Received CONNECT message signature is valid")
ack_message = Ack(
signing_key.pubkey, remote_connect_document.challenge, currency=currency
).get_signed_json(signing_key)
# Send ACK message
logging.debug("Send ACK message...")
ws.send_str(ack_message)
if "auth" in data and data["auth"] == "ACK":
jsonschema.validate(data, ws2p.network.WS2P_ACK_MESSAGE_SCHEMA)
logging.debug("Received an ACK message")
# Create ACK document from ACK response to verify signature
Ack(
data["pub"],
connect_document.challenge,
signature=data["sig"],
currency=currency,
)
logging.debug("Received ACK message signature is valid")
# If ACK response is ok, create OK message
ok_message = Ok(
signing_key.pubkey, connect_document.challenge, currency=currency
).get_signed_json(signing_key)
# Send OK message
logging.debug("Send OK message...")
ws.send_str(ok_message)
if (
remote_connect_document is not None
and "auth" in data
and data["auth"] == "OK"
):
jsonschema.validate(data, ws2p.network.WS2P_OK_MESSAGE_SCHEMA)
logging.debug("Received an OK message")
Ok(
remote_connect_document.pubkey,
connect_document.challenge,
signature=data["sig"],
currency=currency,
)
logging.debug("Received OK message signature is valid")
# END HANDSHAKE #######################################################
logging.debug("END OF HANDSHAKE\n")
# exit loop
break
def generate_ws2p_endpoint(
bma_endpoint: Union[str, BMAEndpoint, SecuredBMAEndpoint]
) -> WS2PEndpoint:
"""
Retrieve WS2P endpoints from BMA peering
Take the first one found
"""
bma_client = Client(bma_endpoint)
peering = bma_client(bma.network.peering)
for endpoint in peering["endpoints"]:
if endpoint.startswith("WS2P"):
return WS2PEndpoint.from_inline(endpoint)
raise ValueError("No WS2P endpoint found")
|
