-
Notifications
You must be signed in to change notification settings - Fork 31
/
rlpx.nim
1403 lines (1186 loc) · 47.4 KB
/
rlpx.nim
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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# nim-eth
# Copyright (c) 2018-2024 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at
# https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at
# https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except
# according to those terms.
## This module implements the `RLPx` Transport Protocol defined at
## `RLPx <https://github.com/ethereum/devp2p/blob/5713591d0366da78a913a811c7502d9ca91d29a8/rlpx.md>`_
## in its EIP-8 version.
##
## This modules implements version 5 of the p2p protocol as defined by EIP-706 -
## earlier versions are not supported.
##
## Both, the message ID and the request/response ID are now unsigned. This goes
## along with the RLPx specs (see above) and the sub-protocol specs at
## `sub-proto <https://github.com/ethereum/devp2p/tree/master/caps>`_ plus the
## fact that RLP is defined for non-negative integers smaller than 2^64 only at
## `Yellow Paper <https://ethereum.github.io/yellowpaper/paper.pdf#appendix.B>`_,
## Appx B, clauses (195) ff and (199).
##
{.push raises: [].}
import
std/[algorithm, deques, options, os, sequtils, strutils, typetraits],
stew/byteutils,
stew/shims/macros,
chronicles,
chronos,
metrics,
snappy,
../rlp,
./private/p2p_types,
./[kademlia, auth, rlpxcrypt, enode, p2p_protocol_dsl]
const
devp2pSnappyVersion* = 5
## EIP-706 version of devp2p, with snappy compression - no support offered
## for earlier versions
maxMsgSize = 1024 * 1024 * 16
## The maximum message size is normally limited by the 24-bit length field in
## the message header but in the case of snappy, we need to protect against
## decompression bombs:
## https://eips.ethereum.org/EIPS/eip-706#avoiding-dos-attacks
connectionTimeout = 10.seconds
msgIdHello = byte 0
msgIdDisconnect = byte 1
msgIdPing = byte 2
msgIdPong = byte 3
# TODO: chronicles re-export here is added for the error
# "undeclared identifier: 'activeChroniclesStream'", when the code using p2p
# does not import chronicles. Need to resolve this properly.
export options, p2pProtocol, rlp, chronicles, metrics
declarePublicGauge rlpx_connected_peers, "Number of connected peers in the pool"
declarePublicCounter rlpx_connect_success, "Number of successfull rlpx connects"
declarePublicCounter rlpx_connect_failure,
"Number of rlpx connects that failed", labels = ["reason"]
declarePublicCounter rlpx_accept_success, "Number of successful rlpx accepted peers"
declarePublicCounter rlpx_accept_failure,
"Number of rlpx accept attempts that failed", labels = ["reason"]
logScope:
topics = "eth p2p rlpx"
type
ResponderWithId*[MsgType] = object
peer*: Peer
reqId*: uint64
ResponderWithoutId*[MsgType] = distinct Peer
# We need these two types in rlpx/devp2p as no parameters or single parameters
# are not getting encoded in an rlp list.
# TODO: we could generalize this in the protocol dsl but it would need an
# `alwaysList` flag as not every protocol expects lists in these cases.
EmptyList = object
DisconnectionReasonList = object
value: DisconnectionReason
proc read(
rlp: var Rlp, T: type DisconnectionReasonList
): T {.gcsafe, raises: [RlpError].} =
## Rlp mixin: `DisconnectionReasonList` parser
if rlp.isList:
# Be strict here: The expression `rlp.read(DisconnectionReasonList)`
# accepts lists with at least one item. The array expression wants
# exactly one item.
if rlp.rawData.len < 3:
# avoids looping through all items when parsing for an overlarge array
return DisconnectionReasonList(value: rlp.read(array[1, DisconnectionReason])[0])
# Also accepted: a single byte reason code. Is is typically used
# by variants of the reference implementation `Geth`
elif rlp.blobLen <= 1:
return DisconnectionReasonList(value: rlp.read(DisconnectionReason))
# Also accepted: a blob of a list (aka object) of reason code. It is
# used by `bor`, a `geth` fork
elif rlp.blobLen < 4:
var subList = rlp.toBytes.rlpFromBytes
if subList.isList:
# Ditto, see above.
return
DisconnectionReasonList(value: subList.read(array[1, DisconnectionReason])[0])
raise newException(RlpTypeMismatch, "Single entry list expected")
include p2p_tracing
when tracingEnabled:
import eth/common/eth_types_json_serialization
export
# XXX: This is a work-around for a Nim issue.
# See a more detailed comment in p2p_tracing.nim
init,
writeValue,
getOutput
proc init*[MsgName](T: type ResponderWithId[MsgName], peer: Peer, reqId: uint64): T =
T(peer: peer, reqId: reqId)
proc init*[MsgName](T: type ResponderWithoutId[MsgName], peer: Peer): T =
T(peer)
chronicles.formatIt(Peer):
$(it.remote)
chronicles.formatIt(Opt[uint64]):
(if it.isSome(): $it.value else: "-1")
include p2p_backends_helpers
proc requestResolver[MsgType](msg: pointer, future: FutureBase) {.gcsafe.} =
var f = Future[Option[MsgType]](future)
if not f.finished:
if msg != nil:
f.complete some(cast[ptr MsgType](msg)[])
else:
f.complete none(MsgType)
proc linkSendFailureToReqFuture[S, R](sendFut: Future[S], resFut: Future[R]) =
sendFut.addCallback do(arg: pointer):
# Avoiding potentially double future completions
if not resFut.finished:
if sendFut.failed:
resFut.fail(sendFut.error)
proc messagePrinter[MsgType](msg: pointer): string {.gcsafe.} =
result = ""
# TODO: uncommenting the line below increases the compile-time
# tremendously (for reasons not yet known)
# result = $(cast[ptr MsgType](msg)[])
proc disconnect*(
peer: Peer, reason: DisconnectionReason, notifyOtherPeer = false
) {.async: (raises: []).}
# TODO Rework the disconnect-and-raise flow to not do both raising
# and disconnection - this results in convoluted control flow and redundant
# disconnect calls
template raisePeerDisconnected(msg: string, r: DisconnectionReason) =
var e = newException(PeerDisconnected, msg)
e.reason = r
raise e
proc disconnectAndRaise(
peer: Peer, reason: DisconnectionReason, msg: string
) {.async: (raises: [PeerDisconnected]).} =
if reason == BreachOfProtocol:
warn "TODO Raising protocol breach",
remote = peer.remote, clientId = peer.clientId, msg
await peer.disconnect(reason)
raisePeerDisconnected(msg, reason)
proc handshakeImpl[T](
peer: Peer,
sendFut: Future[void],
responseFut: auto, # Future[T].Raising([CancelledError, EthP2PError]),
timeout: Duration,
): Future[T] {.async: (raises: [CancelledError, EthP2PError]).} =
sendFut.addCallback do(arg: pointer) {.gcsafe.}:
if sendFut.failed:
debug "Handshake message not delivered", peer
doAssert timeout.milliseconds > 0
try:
let res = await responseFut.wait(timeout)
return res
except AsyncTimeoutError:
# TODO: Really shouldn't disconnect and raise everywhere. In order to avoid
# understanding what error occured where.
# And also, incoming and outgoing disconnect errors should be seperated,
# probably by seperating the actual disconnect call to begin with.
await disconnectAndRaise(peer, TcpError, T.name() & " was not received in time.")
# Dispatcher
#
proc describeProtocols(d: Dispatcher): string =
d.activeProtocols.mapIt($it.capability).join(",")
proc numProtocols(d: Dispatcher): int =
d.activeProtocols.len
proc getDispatcher(
node: EthereumNode, otherPeerCapabilities: openArray[Capability]
): Opt[Dispatcher] =
let dispatcher = Dispatcher()
newSeq(dispatcher.protocolOffsets, protocolCount())
dispatcher.protocolOffsets.fill Opt.none(uint64)
var nextUserMsgId = 0x10u64
for localProtocol in node.protocols:
let idx = localProtocol.index
block findMatchingProtocol:
for remoteCapability in otherPeerCapabilities:
if localProtocol.capability == remoteCapability:
dispatcher.protocolOffsets[idx] = Opt.some(nextUserMsgId)
nextUserMsgId += localProtocol.messages.len.uint64
break findMatchingProtocol
template copyTo(src, dest; index: int) =
for i in 0 ..< src.len:
dest[index + i] = src[i]
dispatcher.messages = newSeq[MessageInfo](nextUserMsgId)
devp2pInfo.messages.copyTo(dispatcher.messages, 0)
for localProtocol in node.protocols:
let idx = localProtocol.index
if dispatcher.protocolOffsets[idx].isSome:
dispatcher.activeProtocols.add localProtocol
localProtocol.messages.copyTo(
dispatcher.messages, dispatcher.protocolOffsets[idx].value.int
)
if dispatcher.numProtocols == 0:
Opt.none(Dispatcher)
else:
Opt.some(dispatcher)
proc getMsgName*(peer: Peer, msgId: uint64): string =
if not peer.dispatcher.isNil and msgId < peer.dispatcher.messages.len.uint64 and
not peer.dispatcher.messages[msgId].isNil:
return peer.dispatcher.messages[msgId].name
else:
return
case msgId
of msgIdHello:
"hello"
of msgIdDisconnect:
"disconnect"
of msgIdPing:
"ping"
of msgIdPong:
"pong"
else:
$msgId
# Protocol info objects
#
proc initProtocol(
name: string,
version: uint64,
peerInit: PeerStateInitializer,
networkInit: NetworkStateInitializer,
): ProtocolInfo =
ProtocolInfo(
capability: Capability(name: name, version: version),
messages: @[],
peerStateInitializer: peerInit,
networkStateInitializer: networkInit,
)
proc setEventHandlers(
p: ProtocolInfo,
onPeerConnected: OnPeerConnectedHandler,
onPeerDisconnected: OnPeerDisconnectedHandler,
) =
p.onPeerConnected = onPeerConnected
p.onPeerDisconnected = onPeerDisconnected
proc cmp*(lhs, rhs: ProtocolInfo): int =
let c = cmp(lhs.capability.name, rhs.capability.name)
if c == 0:
# Highest version first!
-cmp(lhs.capability.version, rhs.capability.version)
else:
c
proc nextMsgResolver[MsgType](
msgData: Rlp, future: FutureBase
) {.gcsafe, raises: [RlpError].} =
var reader = msgData
Future[MsgType](future).complete reader.readRecordType(
MsgType, MsgType.rlpFieldsCount > 1
)
proc failResolver[MsgType](reason: DisconnectionReason, future: FutureBase) =
Future[MsgType](future).fail(
(ref PeerDisconnected)(msg: "Peer disconnected during handshake", reason: reason),
warn = false,
)
proc registerMsg(
protocol: ProtocolInfo,
msgId: uint64,
name: string,
thunk: ThunkProc,
printer: MessageContentPrinter,
requestResolver: RequestResolver,
nextMsgResolver: NextMsgResolver,
failResolver: FailResolver,
) =
if protocol.messages.len.uint64 <= msgId:
protocol.messages.setLen(msgId + 1)
protocol.messages[msgId] = MessageInfo(
id: msgId,
name: name,
thunk: thunk,
printer: printer,
requestResolver: requestResolver,
nextMsgResolver: nextMsgResolver,
failResolver: failResolver,
)
# Message composition and encryption
#
proc perPeerMsgIdImpl(peer: Peer, proto: ProtocolInfo, msgId: uint64): uint64 =
result = msgId
if not peer.dispatcher.isNil:
result += peer.dispatcher.protocolOffsets[proto.index].value
template getPeer(peer: Peer): auto =
peer
template getPeer(responder: ResponderWithId): auto =
responder.peer
template getPeer(responder: ResponderWithoutId): auto =
Peer(responder)
proc supports*(peer: Peer, proto: ProtocolInfo): bool =
peer.dispatcher.protocolOffsets[proto.index].isSome
proc supports*(peer: Peer, Protocol: type): bool =
## Checks whether a Peer supports a particular protocol
peer.supports(Protocol.protocolInfo)
template perPeerMsgId(peer: Peer, MsgType: type): uint64 =
perPeerMsgIdImpl(peer, MsgType.msgProtocol.protocolInfo, MsgType.msgId)
proc invokeThunk*(
peer: Peer, msgId: uint64, msgData: Rlp
): Future[void] {.async: (raises: [CancelledError, EthP2PError]).} =
template invalidIdError(): untyped =
raise newException(
UnsupportedMessageError,
"RLPx message with an invalid id " & $msgId & " on a connection supporting " &
peer.dispatcher.describeProtocols,
)
if msgId >= peer.dispatcher.messages.len.uint64 or
peer.dispatcher.messages[msgId].isNil:
invalidIdError()
let msgInfo = peer.dispatcher.messages[msgId]
doAssert peer.dispatcher.messages.len == peer.awaitedMessages.len,
"Should have been set up in peer constructor"
# Check if the peer is "expecting" this message as part of a handshake
if peer.awaitedMessages[msgId] != nil:
let awaited = move(peer.awaitedMessages[msgId])
peer.awaitedMessages[msgId] = nil
try:
msgInfo.nextMsgResolver(msgData, awaited)
except rlp.RlpError:
await peer.disconnectAndRaise(
BreachOfProtocol, "Could not decode rlp for " & $msgId
)
else:
await msgInfo.thunk(peer, msgData)
template compressMsg(peer: Peer, data: seq[byte]): seq[byte] =
if peer.snappyEnabled:
snappy.encode(data)
else:
data
proc recvMsg(
peer: Peer
): Future[tuple[msgId: uint64, msgRlp: Rlp]] {.
async: (raises: [CancelledError, EthP2PError])
.} =
var msgBody: seq[byte]
try:
msgBody = await peer.transport.recvMsg()
trace "Received message",
remote = peer.remote,
clientId = peer.clientId,
data = toHex(msgBody.toOpenArray(0, min(255, msgBody.high)))
# TODO we _really_ need an rlp decoder that doesn't require this many
# copies of each message...
var tmp = rlpFromBytes(msgBody)
let msgId = tmp.read(uint64)
if peer.snappyEnabled and tmp.hasData():
let decoded =
snappy.decode(msgBody.toOpenArray(tmp.position, msgBody.high), maxMsgSize)
if decoded.len == 0:
if msgId == 0x01 and msgBody.len > 1 and msgBody.len < 16 and msgBody[1] == 0xc1:
# Nethermind sends its TooManyPeers uncompressed but we want to be nice!
# https://github.com/NethermindEth/nethermind/issues/7726
debug "Trying to decode disconnect uncompressed",
remote = peer.remote, clientId = peer.clientId, data = toHex(msgBody)
else:
await peer.disconnectAndRaise(
BreachOfProtocol, "Could not decompress snappy data"
)
else:
trace "Decoded message",
remote = peer.remote,
clientId = peer.clientId,
decoded = toHex(decoded.toOpenArray(0, min(255, decoded.high)))
tmp = rlpFromBytes(decoded)
return (msgId, tmp)
except TransportError as exc:
await peer.disconnectAndRaise(TcpError, exc.msg)
except RlpxTransportError as exc:
await peer.disconnectAndRaise(BreachOfProtocol, exc.msg)
except RlpError as exc:
# TODO remove this warning before using in production
warn "TODO: RLP decoding failed for msgId",
remote = peer.remote,
clientId = peer.clientId,
err = exc.msg,
rawData = toHex(msgBody)
await peer.disconnectAndRaise(BreachOfProtocol, "Could not decode msgId")
proc encodeMsg(msg: auto): seq[byte] =
var rlpWriter = initRlpWriter()
rlpWriter.appendRecordType(msg, typeof(msg).rlpFieldsCount > 1)
rlpWriter.finish
proc sendMsg(
peer: Peer, msgId: uint64, payload: seq[byte]
): Future[void] {.async: (raises: [CancelledError, EthP2PError]).} =
try:
let
msgIdBytes = rlp.encodeInt(msgId)
payloadBytes = peer.compressMsg(payload)
var msg = newSeqOfCap[byte](msgIdBytes.data.len + payloadBytes.len)
msg.add msgIdBytes.data()
msg.add payloadBytes
trace "Sending message",
remote = peer.remote,
clientId = peer.clientId,
msgId,
data = toHex(msg.toOpenArray(0, min(255, msg.high))),
payload = toHex(payload.toOpenArray(0, min(255, payload.high)))
await peer.transport.sendMsg(msg)
except TransportError as exc:
await peer.disconnectAndRaise(TcpError, exc.msg)
except RlpxTransportError as exc:
await peer.disconnectAndRaise(BreachOfProtocol, exc.msg)
proc send*[Msg](
peer: Peer, msg: Msg
): Future[void] {.async: (raises: [CancelledError, EthP2PError], raw: true).} =
logSentMsg(peer, msg)
peer.sendMsg perPeerMsgId(peer, Msg), encodeMsg(msg)
proc registerRequest(
peer: Peer, timeout: Duration, responseFuture: FutureBase, responseMsgId: uint64
): uint64 =
result =
if peer.lastReqId.isNone:
0u64
else:
peer.lastReqId.value + 1u64
peer.lastReqId = Opt.some(result)
let timeoutAt = Moment.fromNow(timeout)
let req = OutstandingRequest(id: result, future: responseFuture)
peer.outstandingRequests[responseMsgId].addLast req
doAssert(not peer.dispatcher.isNil)
let requestResolver = peer.dispatcher.messages[responseMsgId].requestResolver
proc timeoutExpired(udata: pointer) {.gcsafe.} =
requestResolver(nil, responseFuture)
discard setTimer(timeoutAt, timeoutExpired, nil)
proc resolveResponseFuture(peer: Peer, msgId: uint64, msg: pointer) =
## This function is a split off from the previously combined version with
## the same name using optional request ID arguments. This here is the
## version without a request ID (there is the other part below.).
##
## Optional arguments for macro helpers seem easier to handle with
## polymorphic functions (than a `Opt[]` prototype argument.)
##
let msgInfo = peer.dispatcher.messages[msgId]
logScope:
msg = msgInfo.name
msgContents = msgInfo.printer(msg)
receivedReqId = -1
remotePeer = peer.remote
template outstandingReqs(): auto =
peer.outstandingRequests[msgId]
block: # no request ID
# XXX: This is a response from an ETH-like protocol that doesn't feature
# request IDs. Handling the response is quite tricky here because this may
# be a late response to an already timed out request or a valid response
# from a more recent one.
#
# We can increase the robustness by recording enough features of the
# request so we can recognize the matching response, but this is not very
# easy to do because our peers are allowed to send partial responses.
#
# A more generally robust approach is to maintain a set of the wanted
# data items and then to periodically look for items that have been
# requested long time ago, but are still missing. New requests can be
# issues for such items potentially from another random peer.
var expiredRequests = 0
for req in outstandingReqs:
if not req.future.finished:
break
inc expiredRequests
outstandingReqs.shrink(fromFirst = expiredRequests)
if outstandingReqs.len > 0:
let oldestReq = outstandingReqs.popFirst
msgInfo.requestResolver(msg, oldestReq.future)
else:
trace "late or dup RPLx reply ignored", msgId
proc resolveResponseFuture(peer: Peer, msgId: uint64, msg: pointer, reqId: uint64) =
## Variant of `resolveResponseFuture()` for request ID argument.
let msgInfo = peer.dispatcher.messages[msgId]
logScope:
msg = msgInfo.name
msgContents = msgInfo.printer(msg)
receivedReqId = reqId
remotePeer = peer.remote
template outstandingReqs(): auto =
peer.outstandingRequests[msgId]
block: # have request ID
# TODO: This is not completely sound because we are still using a global
# `reqId` sequence (the problem is that we might get a response ID that
# matches a request ID for a different type of request). To make the code
# correct, we can use a separate sequence per response type, but we have
# to first verify that the other Ethereum clients are supporting this
# correctly (because then, we'll be reusing the same reqIds for different
# types of requests). Alternatively, we can assign a separate interval in
# the `reqId` space for each type of response.
if peer.lastReqId.isNone or reqId > peer.lastReqId.value:
debug "RLPx response without matching request", msgId, reqId
return
var idx = 0
while idx < outstandingReqs.len:
template req(): auto =
outstandingReqs()[idx]
if req.future.finished:
# Here we'll remove the expired request by swapping
# it with the last one in the deque (if necessary):
if idx != outstandingReqs.len - 1:
req = outstandingReqs.popLast
continue
else:
outstandingReqs.shrink(fromLast = 1)
# This was the last item, so we don't have any
# more work to do:
return
if req.id == reqId:
msgInfo.requestResolver msg, req.future
# Here we'll remove the found request by swapping
# it with the last one in the deque (if necessary):
if idx != outstandingReqs.len - 1:
req = outstandingReqs.popLast
else:
outstandingReqs.shrink(fromLast = 1)
return
inc idx
trace "late or dup RPLx reply ignored"
proc checkedRlpRead(
peer: Peer, r: var Rlp, MsgType: type
): auto {.raises: [RlpError].} =
when defined(release):
return r.read(MsgType)
else:
try:
return r.read(MsgType)
except rlp.RlpError as e:
debug "Failed rlp.read",
peer = peer, dataType = MsgType.name, err = e.msg, errName = e.name
#, rlpData = r.inspect -- don't use (might crash)
raise e
proc nextMsg*(
peer: Peer, MsgType: type
): Future[MsgType] {.async: (raises: [CancelledError, EthP2PError], raw: true).} =
## This procs awaits a specific RLPx message.
## Any messages received while waiting will be dispatched to their
## respective handlers. The designated message handler will also run
## to completion before the future returned by `nextMsg` is resolved.
let wantedId = peer.perPeerMsgId(MsgType)
let f = peer.awaitedMessages[wantedId]
if not f.isNil:
return Future[MsgType].Raising([CancelledError, EthP2PError])(f)
initFuture result
peer.awaitedMessages[wantedId] = result
proc dispatchMessages*(peer: Peer) {.async: (raises: []).} =
try:
while peer.connectionState notin {Disconnecting, Disconnected}:
var (msgId, msgData) = await peer.recvMsg()
await peer.invokeThunk(msgId, msgData)
except EthP2PError:
# TODO Is this needed? Most such exceptions are raised with an accompanying
# disconnect already .. ClientQuitting isn't a great error but as good
# as any since it will have no effect if the disconnect already happened
await peer.disconnect(ClientQuitting)
except CancelledError:
await peer.disconnect(ClientQuitting)
proc p2pProtocolBackendImpl*(protocol: P2PProtocol): Backend =
let
resultIdent = ident "result"
Peer = bindSym "Peer"
EthereumNode = bindSym "EthereumNode"
initRlpWriter = bindSym "initRlpWriter"
append = bindSym("append", brForceOpen)
read = bindSym("read", brForceOpen)
checkedRlpRead = bindSym "checkedRlpRead"
startList = bindSym "startList"
tryEnterList = bindSym "tryEnterList"
finish = bindSym "finish"
messagePrinter = bindSym "messagePrinter"
nextMsgResolver = bindSym "nextMsgResolver"
failResolver = bindSym "failResolver"
registerRequest = bindSym "registerRequest"
requestResolver = bindSym "requestResolver"
resolveResponseFuture = bindSym "resolveResponseFuture"
sendMsg = bindSym "sendMsg"
nextMsg = bindSym "nextMsg"
initProtocol = bindSym"initProtocol"
registerMsg = bindSym "registerMsg"
perPeerMsgId = bindSym "perPeerMsgId"
perPeerMsgIdImpl = bindSym "perPeerMsgIdImpl"
linkSendFailureToReqFuture = bindSym "linkSendFailureToReqFuture"
handshakeImpl = bindSym "handshakeImpl"
ResponderWithId = bindSym "ResponderWithId"
ResponderWithoutId = bindSym "ResponderWithoutId"
isSubprotocol = protocol.rlpxName != "p2p"
if protocol.rlpxName.len == 0:
protocol.rlpxName = protocol.name
# By convention, all Ethereum protocol names have at least 3 characters.
doAssert protocol.rlpxName.len >= 3
new result
result.registerProtocol = bindSym "registerProtocol"
result.setEventHandlers = bindSym "setEventHandlers"
result.PeerType = Peer
result.NetworkType = EthereumNode
result.ResponderType =
if protocol.useRequestIds: ResponderWithId else: ResponderWithoutId
result.implementMsg = proc(msg: Message) =
var
msgIdValue = msg.id
msgIdent = msg.ident
msgName = $msgIdent
msgRecName = msg.recName
responseMsgId =
if msg.response.isNil:
Opt.none(uint64)
else:
Opt.some(msg.response.id)
hasReqId = msg.hasReqId
protocol = msg.protocol
# variables used in the sending procs
peerOrResponder = ident"peerOrResponder"
rlpWriter = ident"writer"
perPeerMsgIdVar = ident"perPeerMsgId"
# variables used in the receiving procs
receivedRlp = ident"rlp"
receivedMsg = ident"msg"
var
readParams = newNimNode(nnkStmtList)
paramsToWrite = newSeq[NimNode](0)
appendParams = newNimNode(nnkStmtList)
if hasReqId:
# Messages using request Ids
readParams.add quote do:
let `reqIdVar` = `read`(`receivedRlp`, uint64)
case msg.kind
of msgRequest:
doAssert responseMsgId.isSome
let reqToResponseOffset = responseMsgId.value - msgIdValue
let responseMsgId = quote:
`perPeerMsgIdVar` + `reqToResponseOffset`
# Each request is registered so we can resolve it when the response
# arrives. There are two types of protocols: newer protocols use
# explicit `reqId` sent over the wire, while old versions of the ETH wire
# protocol assume response order matches requests.
let registerRequestCall =
newCall(registerRequest, peerVar, timeoutVar, resultIdent, responseMsgId)
if hasReqId:
appendParams.add quote do:
initFuture `resultIdent`
let `reqIdVar` = `registerRequestCall`
paramsToWrite.add reqIdVar
else:
appendParams.add quote do:
initFuture `resultIdent`
discard `registerRequestCall`
of msgResponse:
if hasReqId:
paramsToWrite.add newDotExpr(peerOrResponder, reqIdVar)
of msgHandshake, msgNotification:
discard
for param, paramType in msg.procDef.typedParams(skip = 1):
# This is a fragment of the sending proc that
# serializes each of the passed parameters:
paramsToWrite.add param
# The received RLP data is deserialized to a local variable of
# the message-specific type. This is done field by field here:
readParams.add quote do:
`receivedMsg`.`param` = `checkedRlpRead`(`peerVar`, `receivedRlp`, `paramType`)
let
paramCount = paramsToWrite.len
readParamsPrelude =
if paramCount > 1:
newCall(tryEnterList, receivedRlp)
else:
newStmtList()
when tracingEnabled:
readParams.add newCall(bindSym"logReceivedMsg", peerVar, receivedMsg)
let callResolvedResponseFuture =
if msg.kind != msgResponse:
newStmtList()
elif hasReqId:
newCall(
resolveResponseFuture,
peerVar,
newCall(perPeerMsgId, peerVar, msgRecName),
newCall("addr", receivedMsg),
reqIdVar,
)
else:
newCall(
resolveResponseFuture,
peerVar,
newCall(perPeerMsgId, peerVar, msgRecName),
newCall("addr", receivedMsg),
)
var userHandlerParams = @[peerVar]
if hasReqId:
userHandlerParams.add reqIdVar
let
awaitUserHandler = msg.genAwaitUserHandler(receivedMsg, userHandlerParams)
thunkName = ident(msgName & "Thunk")
msg.defineThunk quote do:
proc `thunkName`(
`peerVar`: `Peer`, data: Rlp
) {.async: (raises: [CancelledError, EthP2PError]).} =
var `receivedRlp` = data
var `receivedMsg`: `msgRecName`
try:
`readParamsPrelude`
`readParams`
`awaitUserHandler`
`callResolvedResponseFuture`
except rlp.RlpError as exc:
# TODO this is a pre-release warning - we should turn this into an
# actual BreachOfProtocol disconnect down the line
warn "TODO: RLP decoding failed for incoming message",
msg = name(`msgRecName`),
remote = `peerVar`.remote,
clientId = `peerVar`.clientId,
err = exc.msg
await `peerVar`.disconnectAndRaise(
BreachOfProtocol, "Invalid RLP in parameter list for " & $(`msgRecName`)
)
var sendProc = msg.createSendProc(isRawSender = (msg.kind == msgHandshake))
sendProc.def.params[1][0] = peerOrResponder
let
msgBytes = ident"msgBytes"
finalizeRequest = quote:
let `msgBytes` = `finish`(`rlpWriter`)
perPeerMsgIdValue =
if isSubprotocol:
newCall(perPeerMsgIdImpl, peerVar, protocol.protocolInfo, newLit(msgIdValue))
else:
newLit(msgIdValue)
var sendCall = newCall(sendMsg, peerVar, perPeerMsgIdVar, msgBytes)
let senderEpilogue =
if msg.kind == msgRequest:
# In RLPx requests, the returned future was allocated here and passed
# to `registerRequest`. It's already assigned to the result variable
# of the proc, so we just wait for the sending operation to complete
# and we return in a normal way. (the waiting is done, so we can catch
# any possible errors).
quote:
`linkSendFailureToReqFuture`(`sendCall`, `resultIdent`)
else:
# In normal RLPx messages, we are returning the future returned by the
# `sendMsg` call.
quote:
return `sendCall`
if paramCount > 1:
# In case there are more than 1 parameter,
# the params must be wrapped in a list:
appendParams =
newStmtList(newCall(startList, rlpWriter, newLit(paramCount)), appendParams)
for param in paramsToWrite:
appendParams.add newCall(append, rlpWriter, param)
let initWriter = quote:
var `rlpWriter` = `initRlpWriter`()
let `perPeerMsgIdVar` = `perPeerMsgIdValue`
when tracingEnabled:
appendParams.add logSentMsgFields(peerVar, protocol, msgId, paramsToWrite)
# let paramCountNode = newLit(paramCount)
sendProc.setBody quote do:
let `peerVar` = getPeer(`peerOrResponder`)
`initWriter`
`appendParams`
`finalizeRequest`
`senderEpilogue`
if msg.kind == msgHandshake:
discard msg.createHandshakeTemplate(sendProc.def.name, handshakeImpl, nextMsg)
protocol.outProcRegistrations.add(
newCall(
registerMsg,
protocolVar,
newLit(msgIdValue),
newLit(msgName),
thunkName,
newTree(nnkBracketExpr, messagePrinter, msgRecName),
newTree(nnkBracketExpr, requestResolver, msgRecName),
newTree(nnkBracketExpr, nextMsgResolver, msgRecName),
newTree(nnkBracketExpr, failResolver, msgRecName),
)
)
result.implementProtocolInit = proc(protocol: P2PProtocol): NimNode =
return newCall(
initProtocol,
newLit(protocol.rlpxName),
newLit(protocol.version),
protocol.peerInit,
protocol.netInit,
)
p2pProtocol DevP2P(version = devp2pSnappyVersion, rlpxName = "p2p"):
proc hello(
peer: Peer,
version: uint64,
clientId: string,
capabilities: seq[Capability],
listenPort: uint,
nodeId: array[RawPublicKeySize, byte],
) =
# The first hello message gets processed during the initial handshake - this
# version is used for any subsequent messages
# TODO investigate and turn warning into protocol breach
warn "TODO Multiple hello messages received",
remote = peer.remote, clientId = clientId
# await peer.disconnectAndRaise(BreachOfProtocol, "Multiple hello messages")
proc sendDisconnectMsg(peer: Peer, reason: DisconnectionReasonList) =
## Notify other peer that we're about to disconnect them for the given
## reason
if reason.value == BreachOfProtocol:
# TODO This is a temporary log message at warning level to aid in
# debugging in pre-release versions - it should be removed before
# release
# TODO Nethermind sends BreachOfProtocol on network id mismatch:
# https://github.com/NethermindEth/nethermind/issues/7727
if not peer.clientId.startsWith("Nethermind"):
warn "TODO Peer sent BreachOfProtocol error!",
remote = peer.remote, clientId = peer.clientId
else:
trace "disconnect message received", reason = reason.value, peer
await peer.disconnect(reason.value, false)
# Adding an empty RLP list as the spec defines.
# The parity client specifically checks if there is rlp data.
proc ping(peer: Peer, emptyList: EmptyList) =
discard peer.pong(EmptyList())
proc pong(peer: Peer, emptyList: EmptyList) =
discard
proc removePeer(network: EthereumNode, peer: Peer) =
# It is necessary to check if peer.remote still exists. The connection might
# have been dropped already from the peers side.
# E.g. when receiving a p2p.disconnect message from a peer, a race will happen
# between which side disconnects first.
if network.peerPool != nil and not peer.remote.isNil and
peer.remote in network.peerPool.connectedNodes:
network.peerPool.connectedNodes.del(peer.remote)
rlpx_connected_peers.dec()
# Note: we need to do this check as disconnect (and thus removePeer)
# currently can get called before the dispatcher is initialized.
if not peer.dispatcher.isNil:
for observer in network.peerPool.observers.values:
if not observer.onPeerDisconnected.isNil:
if observer.protocol.isNil or peer.supports(observer.protocol):
observer.onPeerDisconnected(peer)
proc callDisconnectHandlers(
peer: Peer, reason: DisconnectionReason
): Future[void] {.async: (raises: []).} =
let futures = peer.dispatcher.activeProtocols
.filterIt(it.onPeerDisconnected != nil)
.mapIt(it.onPeerDisconnected(peer, reason))
await noCancel allFutures(futures)
proc disconnect*(
peer: Peer, reason: DisconnectionReason, notifyOtherPeer = false
) {.async: (raises: []).} =
if reason == BreachOfProtocol:
# TODO remove warning after all protocol breaches have been investigated
# TODO https://github.com/NethermindEth/nethermind/issues/7727