AtHeartEngineer/prysm
1
0
Fork 0
mirror of https://github.com/OffchainLabs/prysm.git synced 2026-01-10 13:58:09 -05:00
Code Issues Packages Projects Releases Wiki Activity

P2P handshake handling (#2306)

Browse Source 
* proto: illegal wireType 7 :(

* set addr to todo for now so somethign gets sent

* push latest progress

* Add feedback from @raulk. Stream never connects

* working handshake handler

* add exclusions for relay/bootstrap node

* fix tests, still need to add new ones

* remove race, fails coverage

* Add test for negotiation

* gazelle

* regen pb

* Update shared/p2p/handshake_handler.go

Co-Authored-By: prestonvanloon <preston@prysmaticlabs.com>
This commit is contained in:
Preston Van Loon
2019-04-27 15:08:27 -04:00
committed by Raul Jordan
parent 7a04af7621
commit 210edfc940
13 changed files with 515 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
beacon-chain/node/BUILD.bazel
View File

@@ -3,6 +3,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
go_library(
name = "go_default_library",
srcs = [
"fetch_contract_address.go",
"node.go",
"p2p_config.go",
],

41
beacon-chain/node/fetch_contract_address.go Normal file
View File

@@ -0,0 +1,41 @@
package node
import (
"io/ioutil"
"net/http"
"sync"
"github.com/prysmaticlabs/prysm/shared/params"
)
var cachedDepositAddress string
var fetchLock sync.Mutex
// fetchDepositContract from the cluster endpoint.
func fetchDepositContract() (string, error) {
fetchLock.Lock()
defer fetchLock.Unlock()
if cachedDepositAddress != "" {
return cachedDepositAddress, nil
}
log.WithField(
"endpoint",
params.BeaconConfig().TestnetContractEndpoint,
).Info("Fetching testnet cluster address")
resp, err := http.Get(params.BeaconConfig().TestnetContractEndpoint)
if err != nil {
return "", err
}
contractResponse, err := ioutil.ReadAll(resp.Body)
if err != nil {
return "", err
}
if err := resp.Body.Close(); err != nil {
return "", err
}
cachedDepositAddress = string(contractResponse)
return cachedDepositAddress, nil
}

19
beacon-chain/node/node.go
View File

@@ -4,8 +4,6 @@ package node
import (
"context"
"fmt"
"io/ioutil"
"net/http"
"os"
"os/signal"
"path"
@@ -248,22 +246,11 @@ func (b *BeaconNode) registerPOWChainService(cliCtx *cli.Context) error {
depAddress := cliCtx.GlobalString(utils.DepositContractFlag.Name)
if depAddress == "" {
log.Infof("Fetching testnet cluster address from %s...", params.BeaconConfig().TestnetContractEndpoint)
resp, err := http.Get(params.BeaconConfig().TestnetContractEndpoint)
var err error
depAddress, err = fetchDepositContract()
if err != nil {
log.Fatalf("Could not get latest deposit contract address: %v", err)
log.WithError(err).Fatal("Cannot fetch deposit contract")
}
defer func() {
if err := resp.Body.Close(); err != nil {
log.Fatal(err)
}
}()
contractResponse, err := ioutil.ReadAll(resp.Body)
if err != nil {
log.Fatal(err)
}
depAddress = string(contractResponse)
}
if !common.IsHexAddress(depAddress) {

19
beacon-chain/node/p2p_config.go
View File

@@ -2,6 +2,7 @@ package node
import (
"github.com/gogo/protobuf/proto"
"github.com/prysmaticlabs/prysm/beacon-chain/utils"
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
"github.com/prysmaticlabs/prysm/shared/cmd"
"github.com/prysmaticlabs/prysm/shared/p2p"
@@ -27,11 +28,21 @@ var topicMappings = map[pb.Topic]proto.Message{
}
func configureP2P(ctx *cli.Context) (*p2p.Server, error) {
contractAddress := ctx.GlobalString(utils.DepositContractFlag.Name)
if contractAddress == "" {
var err error
contractAddress, err = fetchDepositContract()
if err != nil {
return nil, err
}
}
s, err := p2p.NewServer(&p2p.ServerConfig{
NoDiscovery: ctx.GlobalBool(cmd.NoDiscovery.Name),
BootstrapNodeAddr: ctx.GlobalString(cmd.BootstrapNode.Name),
RelayNodeAddr: ctx.GlobalString(cmd.RelayNode.Name),
Port: ctx.GlobalInt(cmd.P2PPort.Name),
NoDiscovery: ctx.GlobalBool(cmd.NoDiscovery.Name),
BootstrapNodeAddr: ctx.GlobalString(cmd.BootstrapNode.Name),
RelayNodeAddr: ctx.GlobalString(cmd.RelayNode.Name),
Port: ctx.GlobalInt(cmd.P2PPort.Name),
DepositContractAddress: contractAddress,
})
if err != nil {
return nil, err

293
proto/beacon/p2p/v1/messages.pb.go generated
View File

@@ -1435,6 +1435,53 @@ func (m *ExitResponse) GetVoluntaryExit() *VoluntaryExit {
return nil
}
type Handshake struct {
DepositContractAddress string `protobuf:"bytes,1,opt,name=deposit_contract_address,json=depositContractAddress,proto3" json:"deposit_contract_address,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *Handshake) Reset() { *m = Handshake{} }
func (m *Handshake) String() string { return proto.CompactTextString(m) }
func (*Handshake) ProtoMessage() {}
func (*Handshake) Descriptor() ([]byte, []int) {
return fileDescriptor_a1d590cda035b632, []int{27}
}
func (m *Handshake) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *Handshake) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_Handshake.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalTo(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *Handshake) XXX_Merge(src proto.Message) {
xxx_messageInfo_Handshake.Merge(m, src)
}
func (m *Handshake) XXX_Size() int {
return m.Size()
}
func (m *Handshake) XXX_DiscardUnknown() {
xxx_messageInfo_Handshake.DiscardUnknown(m)
}
var xxx_messageInfo_Handshake proto.InternalMessageInfo
func (m *Handshake) GetDepositContractAddress() string {
if m != nil {
return m.DepositContractAddress
}
return ""
}
func init() {
proto.RegisterEnum("ethereum.beacon.p2p.v1.Topic", Topic_name, Topic_value)
proto.RegisterType((*Envelope)(nil), "ethereum.beacon.p2p.v1.Envelope")
@@ -1464,68 +1511,71 @@ func init() {
proto.RegisterType((*ExitAnnounce)(nil), "ethereum.beacon.p2p.v1.ExitAnnounce")
proto.RegisterType((*ExitRequest)(nil), "ethereum.beacon.p2p.v1.ExitRequest")
proto.RegisterType((*ExitResponse)(nil), "ethereum.beacon.p2p.v1.ExitResponse")
proto.RegisterType((*Handshake)(nil), "ethereum.beacon.p2p.v1.Handshake")
}
func init() { proto.RegisterFile("proto/beacon/p2p/v1/messages.proto", fileDescriptor_a1d590cda035b632) }
var fileDescriptor_a1d590cda035b632 = []byte{
// 884 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x96, 0xd1, 0x72, 0xda, 0x46,
0x14, 0x86, 0xab, 0xd8, 0x0e, 0xce, 0x01, 0x13, 0xb2, 0x6e, 0x63, 0xec, 0x26, 0xd8, 0x56, 0xea,
0xa9, 0xdb, 0x99, 0xe0, 0x89, 0x73, 0x95, 0x8b, 0x4e, 0x47, 0xc2, 0x9a, 0x92, 0xc4, 0x15, 0xa9,
0x80, 0x76, 0x7a, 0xb5, 0x5d, 0xc4, 0xd6, 0x30, 0xc5, 0xbb, 0x5b, 0x76, 0x61, 0xec, 0xde, 0xf7,
0x19, 0xfa, 0x04, 0x7d, 0x89, 0x3e, 0x41, 0x2f, 0xfb, 0x08, 0x1d, 0x3f, 0x49, 0x47, 0xab, 0x15,
0x16, 0x20, 0xcb, 0x5c, 0xf4, 0x0e, 0x9d, 0xf3, 0xff, 0xff, 0x9e, 0x6f, 0x39, 0x62, 0x00, 0x5b,
0x8c, 0xb9, 0xe2, 0x27, 0x3d, 0x4a, 0x42, 0xce, 0x4e, 0xc4, 0xa9, 0x38, 0x99, 0xbe, 0x3a, 0xb9,
0xa4, 0x52, 0x92, 0x0b, 0x2a, 0xeb, 0xba, 0x89, 0x9e, 0x52, 0x35, 0xa0, 0x63, 0x3a, 0xb9, 0xac,
0xc7, 0xb2, 0xba, 0x38, 0x15, 0xf5, 0xe9, 0xab, 0xbd, 0xfd, 0x2c, 0xaf, 0xba, 0x16, 0x89, 0xd1,
0xfe, 0x06, 0x36, 0x3d, 0x36, 0xa5, 0x23, 0x2e, 0x28, 0x3a, 0x84, 0x92, 0x14, 0x84, 0xe1, 0x90,
0x33, 0x45, 0xaf, 0x54, 0xd5, 0x3a, 0xb0, 0x8e, 0x4b, 0x41, 0x31, 0xaa, 0x35, 0xe2, 0x12, 0xaa,
0x42, 0x41, 0x90, 0xeb, 0x11, 0x27, 0xfd, 0xea, 0x03, 0xdd, 0x4d, 0x1e, 0xed, 0x77, 0xb0, 0xed,
0xea, 0x53, 0xdc, 0x11, 0x0f, 0x7f, 0x71, 0x18, 0xe3, 0x13, 0x16, 0x52, 0x84, 0x60, 0x7d, 0x40,
0xe4, 0xc0, 0x64, 0xe9, 0xcf, 0x68, 0x1f, 0x8a, 0x72, 0xc4, 0x15, 0x66, 0x93, 0xcb, 0x1e, 0x1d,
0xeb, 0xa0, 0xf5, 0x00, 0xa2, 0x92, 0xaf, 0x2b, 0xf6, 0x31, 0xa0, 0x54, 0x56, 0x40, 0x7f, 0x9d,
0x50, 0xa9, 0xb2, 0xa2, 0x6c, 0x07, 0x6a, 0xcb, 0x4a, 0xf7, 0xba, 0x3d, 0xcb, 0x5a, 0x3c, 0xcc,
0x5a, 0x3a, 0xec, 0x0f, 0x6b, 0x6e, 0xf2, 0x80, 0x4a, 0xc1, 0x99, 0xa4, 0xe8, 0x0d, 0x6c, 0xf4,
0xa2, 0x82, 0xb6, 0x14, 0x4f, 0x5f, 0xd4, 0xb3, 0xaf, 0xb8, 0x9e, 0xf6, 0xc6, 0x0e, 0xe4, 0x41,
0x91, 0x28, 0x45, 0xa5, 0x22, 0x6a, 0xc8, 0x99, 0x06, 0xcc, 0x09, 0x70, 0x6e, 0xa5, 0x41, 0xda,
0x67, 0x77, 0x61, 0xd7, 0x25, 0x2a, 0x1c, 0xd0, 0x7e, 0xc6, 0x6d, 0x3c, 0x07, 0x90, 0x8a, 0x8c,
0x15, 0x8e, 0x50, 0x0c, 0xd6, 0x23, 0x5d, 0x89, 0xe0, 0xd1, 0x2e, 0x6c, 0x52, 0xd6, 0x8f, 0x9b,
0xf1, 0x05, 0x17, 0x28, 0xeb, 0x47, 0x2d, 0x7b, 0x00, 0x7b, 0x59, 0xb1, 0x06, 0xfb, 0x1d, 0x94,
0x7b, 0x71, 0x17, 0x6b, 0x18, 0x59, 0xb5, 0x0e, 0xd6, 0x56, 0xe5, 0xdf, 0x32, 0x56, 0xfd, 0x24,
0x6d, 0x04, 0x95, 0xc6, 0x80, 0x0c, 0x59, 0x93, 0x92, 0xbe, 0x99, 0xdb, 0xfe, 0xcb, 0x82, 0x27,
0xa9, 0xa2, 0x39, 0xf5, 0x08, 0xca, 0x21, 0x61, 0x9c, 0x0d, 0x43, 0x32, 0x4a, 0x13, 0x6d, 0xcd,
0xaa, 0x9a, 0xea, 0x2b, 0xf8, 0x34, 0x25, 0x53, 0x44, 0x51, 0x3c, 0xe6, 0x5c, 0xe1, 0x68, 0x17,
0x5e, 0x9f, 0x9a, 0x95, 0xac, 0xde, 0x7a, 0x22, 0x45, 0xc0, 0xb9, 0x6a, 0xea, 0x3e, 0xfa, 0x1a,
0x9e, 0xfd, 0x3c, 0x64, 0x64, 0x34, 0xfc, 0x8d, 0xf6, 0x97, 0xed, 0xb2, 0xba, 0xa6, 0xfd, 0xbb,
0x33, 0xcd, 0x82, 0x5f, 0xda, 0x2f, 0x61, 0x27, 0xc6, 0xd5, 0x9d, 0xa8, 0x9a, 0xb7, 0xe8, 0x76,
0x37, 0xd9, 0xe3, 0x38, 0xc8, 0x7c, 0x73, 0xf7, 0x4d, 0x61, 0xdd, 0x37, 0x45, 0x98, 0x2c, 0xac,
0x89, 0x35, 0x77, 0x78, 0x0e, 0x8f, 0x17, 0x72, 0x57, 0x5b, 0xdd, 0x38, 0xa5, 0x3c, 0x7f, 0x9e,
0xfd, 0x05, 0x6c, 0xa7, 0x16, 0x33, 0x17, 0xf3, 0x18, 0x50, 0x7a, 0x87, 0x73, 0x5e, 0x57, 0x31,
0x17, 0x3a, 0x9b, 0x3c, 0xeb, 0x47, 0xe2, 0x7f, 0x7a, 0x87, 0xea, 0x50, 0xfd, 0x30, 0xe6, 0x82,
0x4b, 0x3a, 0x6e, 0x8f, 0x88, 0x1c, 0x0c, 0xd9, 0x45, 0x2e, 0xcb, 0x4b, 0xd8, 0x59, 0xd4, 0xe7,
0x01, 0xfd, 0x6e, 0x2d, 0xe7, 0xe7, 0x62, 0x75, 0xe1, 0x89, 0x30, 0x7a, 0x2c, 0x8d, 0xc1, 0xc0,
0x1d, 0xdf, 0x05, 0xb7, 0x74, 0x40, 0x45, 0x2c, 0x54, 0x22, 0xcc, 0xf8, 0x0a, 0x56, 0xc7, 0x5c,
0xd4, 0xdf, 0x87, 0xb9, 0xac, 0xcf, 0xc7, 0x4c, 0xf4, 0x2b, 0x63, 0x2e, 0x1d, 0x50, 0x59, 0xac,
0xd8, 0x47, 0xf0, 0xf8, 0x8c, 0x0a, 0x2e, 0x87, 0x2a, 0x97, 0xee, 0x33, 0x28, 0x1b, 0x59, 0x1e,
0xd4, 0x4f, 0xb3, 0xb0, 0x5c, 0x94, 0x37, 0x50, 0xe8, 0xc7, 0x32, 0x03, 0xb0, 0x7f, 0x17, 0x40,
0x92, 0x96, 0xe8, 0x6d, 0x1b, 0x4a, 0xde, 0xd5, 0x3d, 0xb3, 0x1e, 0x42, 0x31, 0xd2, 0xe4, 0xbf,
0x35, 0xa5, 0x58, 0x92, 0x33, 0xe5, 0x39, 0x94, 0xa7, 0x7c, 0x34, 0x61, 0x8a, 0x8c, 0xaf, 0x31,
0xbd, 0x9a, 0x0d, 0x7b, 0x74, 0xd7, 0xb0, 0xdf, 0x27, 0x6a, 0x1d, 0xbd, 0x35, 0x4d, 0x3f, 0x7e,
0xf9, 0xe7, 0x1a, 0x6c, 0x74, 0xb8, 0x18, 0x86, 0xa8, 0x08, 0x85, 0xae, 0xff, 0xde, 0x6f, 0xfd,
0xe0, 0x57, 0x3e, 0x42, 0xbb, 0xf0, 0x89, 0xeb, 0x39, 0x8d, 0x96, 0x8f, 0xdd, 0xf3, 0x56, 0xe3,
0x3d, 0x76, 0x7c, 0xbf, 0xd5, 0xf5, 0x1b, 0x5e, 0xc5, 0x42, 0x55, 0xf8, 0x78, 0xae, 0x15, 0x78,
0xdf, 0x75, 0xbd, 0x76, 0xa7, 0xf2, 0x00, 0x7d, 0x0e, 0x2f, 0xb2, 0x3a, 0xd8, 0xfd, 0x11, 0xb7,
0xcf, 0x5b, 0x1d, 0xec, 0x77, 0xbf, 0x75, 0xbd, 0xa0, 0xb2, 0xb6, 0x94, 0x1e, 0x78, 0xed, 0x0f,
0x2d, 0xbf, 0xed, 0x55, 0xd6, 0xd1, 0x01, 0x3c, 0x73, 0x9d, 0x4e, 0xa3, 0xe9, 0x9d, 0xe1, 0xcc,
0x53, 0x36, 0xd0, 0x21, 0x3c, 0xbf, 0x43, 0x61, 0x42, 0x1e, 0xa2, 0xa7, 0x80, 0x1a, 0x4d, 0xe7,
0xad, 0x8f, 0x9b, 0x9e, 0x73, 0x36, 0xb3, 0x16, 0xd0, 0x0e, 0x6c, 0xcf, 0xd5, 0x8d, 0x61, 0x13,
0xd5, 0x60, 0xcf, 0x64, 0xb5, 0x3b, 0x4e, 0xc7, 0xc3, 0x4d, 0xa7, 0xdd, 0xbc, 0x65, 0x7e, 0x94,
0x62, 0x8e, 0xfb, 0x49, 0x24, 0xa4, 0x50, 0x92, 0x8e, 0x09, 0x2d, 0x46, 0x26, 0xa7, 0xd3, 0xf1,
0xa2, 0xfa, 0xdb, 0x96, 0x7f, 0x1b, 0x57, 0x8a, 0xe6, 0x48, 0x77, 0x92, 0xb4, 0xad, 0x45, 0xcb,
0x2c, 0xac, 0xec, 0x96, 0xfe, 0xbe, 0xa9, 0x59, 0xff, 0xdc, 0xd4, 0xac, 0x7f, 0x6f, 0x6a, 0x56,
0xef, 0xa1, 0xfe, 0x4b, 0xf7, 0xfa, 0xbf, 0x00, 0x00, 0x00, 0xff, 0xff, 0xb1, 0xd9, 0xa8, 0x18,
0x31, 0x0a, 0x00, 0x00,
// 921 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x96, 0xdf, 0x72, 0xdb, 0x44,
0x14, 0xc6, 0x51, 0x93, 0xd4, 0xc9, 0xb1, 0xe3, 0xba, 0x1b, 0x48, 0x94, 0xd0, 0x3a, 0x89, 0x4a,
0x86, 0xc0, 0x4c, 0x9d, 0x69, 0x7a, 0x43, 0x2f, 0x18, 0x46, 0x72, 0x34, 0xb8, 0x6d, 0x90, 0x8b,
0x6c, 0xc3, 0x70, 0xb5, 0xac, 0xa5, 0xa5, 0xf6, 0xd4, 0xd9, 0x15, 0xda, 0xb5, 0x27, 0xe1, 0x9e,
0x67, 0xe0, 0x09, 0x78, 0x09, 0x9e, 0x80, 0x4b, 0x1e, 0x81, 0xc9, 0x93, 0x30, 0x5a, 0xad, 0x1c,
0xf9, 0x4f, 0x94, 0x5c, 0x70, 0x17, 0x9d, 0xf3, 0x7d, 0xdf, 0x9e, 0xdf, 0xe6, 0xc8, 0x23, 0xb0,
0xa2, 0x98, 0x4b, 0x7e, 0xd2, 0xa7, 0x24, 0xe0, 0xec, 0x24, 0x3a, 0x8d, 0x4e, 0x26, 0x2f, 0x4e,
0x2e, 0xa8, 0x10, 0xe4, 0x3d, 0x15, 0x0d, 0xd5, 0x44, 0xdb, 0x54, 0x0e, 0x68, 0x4c, 0xc7, 0x17,
0x8d, 0x54, 0xd6, 0x88, 0x4e, 0xa3, 0xc6, 0xe4, 0xc5, 0xde, 0xfe, 0x32, 0xaf, 0xbc, 0x8a, 0x32,
0xa3, 0xf5, 0x2d, 0xac, 0xbb, 0x6c, 0x42, 0x47, 0x3c, 0xa2, 0xe8, 0x10, 0x2a, 0x22, 0x22, 0x0c,
0x07, 0x9c, 0x49, 0x7a, 0x29, 0x4d, 0xe3, 0xc0, 0x38, 0xae, 0xf8, 0xe5, 0xa4, 0xd6, 0x4c, 0x4b,
0xc8, 0x84, 0x52, 0x44, 0xae, 0x46, 0x9c, 0x84, 0xe6, 0x03, 0xd5, 0xcd, 0x1e, 0xad, 0x37, 0xb0,
0xe5, 0xa8, 0x53, 0x9c, 0x11, 0x0f, 0x3e, 0xd8, 0x8c, 0xf1, 0x31, 0x0b, 0x28, 0x42, 0xb0, 0x3a,
0x20, 0x62, 0xa0, 0xb3, 0xd4, 0xdf, 0x68, 0x1f, 0xca, 0x62, 0xc4, 0x25, 0x66, 0xe3, 0x8b, 0x3e,
0x8d, 0x55, 0xd0, 0xaa, 0x0f, 0x49, 0xc9, 0x53, 0x15, 0xeb, 0x18, 0x50, 0x2e, 0xcb, 0xa7, 0xbf,
0x8e, 0xa9, 0x90, 0xcb, 0xa2, 0x2c, 0x1b, 0xea, 0x8b, 0x4a, 0xe7, 0xaa, 0x33, 0xcd, 0x9a, 0x3f,
0xcc, 0x58, 0x38, 0xec, 0x0f, 0x63, 0x66, 0x72, 0x9f, 0x8a, 0x88, 0x33, 0x41, 0xd1, 0x2b, 0x58,
0xeb, 0x27, 0x05, 0x65, 0x29, 0x9f, 0x3e, 0x6b, 0x2c, 0xbf, 0xe2, 0x46, 0xde, 0x9b, 0x3a, 0x90,
0x0b, 0x65, 0x22, 0x25, 0x15, 0x92, 0xc8, 0x21, 0x67, 0x0a, 0xb0, 0x20, 0xc0, 0xbe, 0x91, 0xfa,
0x79, 0x9f, 0xd5, 0x83, 0x5d, 0x87, 0xc8, 0x60, 0x40, 0xc3, 0x25, 0xb7, 0xf1, 0x14, 0x40, 0x48,
0x12, 0x4b, 0x9c, 0xa0, 0x68, 0xac, 0x0d, 0x55, 0x49, 0xe0, 0xd1, 0x2e, 0xac, 0x53, 0x16, 0xa6,
0xcd, 0xf4, 0x82, 0x4b, 0x94, 0x85, 0x49, 0xcb, 0x1a, 0xc0, 0xde, 0xb2, 0x58, 0x8d, 0xfd, 0x06,
0xaa, 0xfd, 0xb4, 0x8b, 0x15, 0x8c, 0x30, 0x8d, 0x83, 0x95, 0xfb, 0xf2, 0x6f, 0x6a, 0xab, 0x7a,
0x12, 0x16, 0x82, 0x5a, 0x73, 0x40, 0x86, 0xac, 0x45, 0x49, 0xa8, 0xe7, 0xb6, 0xfe, 0x32, 0xe0,
0x71, 0xae, 0xa8, 0x4f, 0x3d, 0x82, 0x6a, 0x40, 0x18, 0x67, 0xc3, 0x80, 0x8c, 0xf2, 0x44, 0x9b,
0xd3, 0xaa, 0xa2, 0xfa, 0x1a, 0x3e, 0xcd, 0xc9, 0x24, 0x91, 0x14, 0xc7, 0x9c, 0x4b, 0x9c, 0xec,
0xc2, 0xcb, 0x53, 0xbd, 0x92, 0xe6, 0x8d, 0x27, 0x51, 0xf8, 0x9c, 0xcb, 0x96, 0xea, 0xa3, 0x6f,
0xe0, 0xc9, 0x2f, 0x43, 0x46, 0x46, 0xc3, 0xdf, 0x68, 0xb8, 0x68, 0x17, 0xe6, 0x8a, 0xf2, 0xef,
0x4e, 0x35, 0x73, 0x7e, 0x61, 0x3d, 0x87, 0x9d, 0x14, 0x57, 0x75, 0x92, 0x6a, 0xd1, 0xa2, 0x5b,
0xbd, 0x6c, 0x8f, 0xd3, 0x20, 0xfd, 0x9f, 0xbb, 0x6b, 0x0a, 0xe3, 0xae, 0x29, 0x82, 0x6c, 0x61,
0x75, 0xac, 0xbe, 0xc3, 0x73, 0x78, 0x34, 0x97, 0x7b, 0xbf, 0xd5, 0x4d, 0x53, 0xaa, 0xb3, 0xe7,
0x59, 0x5f, 0xc0, 0x56, 0x6e, 0x31, 0x0b, 0x31, 0x8f, 0x01, 0xe5, 0x77, 0xb8, 0xe0, 0x75, 0x8d,
0x66, 0x42, 0xa7, 0x93, 0x2f, 0xfb, 0x91, 0xf8, 0x9f, 0xde, 0xa1, 0x06, 0x98, 0xef, 0x62, 0x1e,
0x71, 0x41, 0xe3, 0xce, 0x88, 0x88, 0xc1, 0x90, 0xbd, 0x2f, 0x64, 0x79, 0x0e, 0x3b, 0xf3, 0xfa,
0x22, 0xa0, 0xdf, 0x8d, 0xc5, 0xfc, 0x42, 0xac, 0x1e, 0x3c, 0x8e, 0xb4, 0x1e, 0x0b, 0x6d, 0xd0,
0x70, 0xc7, 0xb7, 0xc1, 0x2d, 0x1c, 0x50, 0x8b, 0xe6, 0x2a, 0x09, 0x66, 0x7a, 0x05, 0xf7, 0xc7,
0x9c, 0xd7, 0xdf, 0x85, 0xb9, 0xa8, 0x2f, 0xc6, 0xcc, 0xf4, 0xf7, 0xc6, 0x5c, 0x38, 0xa0, 0x36,
0x5f, 0xb1, 0x8e, 0xe0, 0xd1, 0x19, 0x8d, 0xb8, 0x18, 0xca, 0x42, 0xba, 0xcf, 0xa0, 0xaa, 0x65,
0x45, 0x50, 0x3f, 0x4f, 0xc3, 0x0a, 0x51, 0x5e, 0x41, 0x29, 0x4c, 0x65, 0x1a, 0x60, 0xff, 0x36,
0x80, 0x2c, 0x2d, 0xd3, 0x5b, 0x16, 0x54, 0xdc, 0xcb, 0x3b, 0x66, 0x3d, 0x84, 0x72, 0xa2, 0x29,
0x7e, 0x6b, 0x2a, 0xa9, 0xa4, 0x60, 0xca, 0x73, 0xa8, 0x4e, 0xf8, 0x68, 0xcc, 0x24, 0x89, 0xaf,
0x30, 0xbd, 0x9c, 0x0e, 0x7b, 0x74, 0xdb, 0xb0, 0x3f, 0x64, 0x6a, 0x15, 0xbd, 0x39, 0xc9, 0x3f,
0x5a, 0x2e, 0x6c, 0xb4, 0x08, 0x0b, 0xc5, 0x80, 0x7c, 0xa0, 0xe8, 0x2b, 0x30, 0x35, 0x90, 0xfa,
0x32, 0x88, 0x49, 0x20, 0x31, 0x09, 0xc3, 0x98, 0x8a, 0xf4, 0xb7, 0x6a, 0xc3, 0xdf, 0xd6, 0xfd,
0xa6, 0x6e, 0xdb, 0x69, 0xf7, 0xcb, 0x3f, 0x57, 0x60, 0xad, 0xcb, 0xa3, 0x61, 0x80, 0xca, 0x50,
0xea, 0x79, 0x6f, 0xbd, 0xf6, 0x8f, 0x5e, 0xed, 0x23, 0xb4, 0x0b, 0x9f, 0x38, 0xae, 0xdd, 0x6c,
0x7b, 0xd8, 0x39, 0x6f, 0x37, 0xdf, 0x62, 0xdb, 0xf3, 0xda, 0x3d, 0xaf, 0xe9, 0xd6, 0x0c, 0x64,
0xc2, 0xc7, 0x33, 0x2d, 0xdf, 0xfd, 0xbe, 0xe7, 0x76, 0xba, 0xb5, 0x07, 0xe8, 0x73, 0x78, 0xb6,
0xac, 0x83, 0x9d, 0x9f, 0x70, 0xe7, 0xbc, 0xdd, 0xc5, 0x5e, 0xef, 0x3b, 0xc7, 0xf5, 0x6b, 0x2b,
0x0b, 0xe9, 0xbe, 0xdb, 0x79, 0xd7, 0xf6, 0x3a, 0x6e, 0x6d, 0x15, 0x1d, 0xc0, 0x13, 0xc7, 0xee,
0x36, 0x5b, 0xee, 0x19, 0x5e, 0x7a, 0xca, 0x1a, 0x3a, 0x84, 0xa7, 0xb7, 0x28, 0x74, 0xc8, 0x43,
0xb4, 0x0d, 0xa8, 0xd9, 0xb2, 0x5f, 0x7b, 0xb8, 0xe5, 0xda, 0x67, 0x53, 0x6b, 0x09, 0xed, 0xc0,
0xd6, 0x4c, 0x5d, 0x1b, 0xd6, 0x51, 0x1d, 0xf6, 0x74, 0x56, 0xa7, 0x6b, 0x77, 0x5d, 0xdc, 0xb2,
0x3b, 0xad, 0x1b, 0xe6, 0x8d, 0x1c, 0x73, 0xda, 0xcf, 0x22, 0x21, 0x87, 0x92, 0x75, 0x74, 0x68,
0x39, 0x31, 0xd9, 0xdd, 0xae, 0x9b, 0xd4, 0x5f, 0xb7, 0xbd, 0x9b, 0xb8, 0x4a, 0x32, 0x47, 0xbe,
0x93, 0xa5, 0x6d, 0xce, 0x5b, 0xa6, 0x61, 0x55, 0xa7, 0xf2, 0xf7, 0x75, 0xdd, 0xf8, 0xe7, 0xba,
0x6e, 0xfc, 0x7b, 0x5d, 0x37, 0xfa, 0x0f, 0xd5, 0x97, 0xe1, 0xcb, 0xff, 0x02, 0x00, 0x00, 0xff,
0xff, 0xec, 0x05, 0xc3, 0xc5, 0x78, 0x0a, 0x00, 0x00,
}
func (m *Envelope) Marshal() (dAtA []byte, err error) {
@@ -2350,6 +2400,33 @@ func (m *ExitResponse) MarshalTo(dAtA []byte) (int, error) {
return i, nil
}
func (m *Handshake) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalTo(dAtA)
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *Handshake) MarshalTo(dAtA []byte) (int, error) {
var i int
_ = i
var l int
_ = l
if len(m.DepositContractAddress) > 0 {
dAtA[i] = 0xa
i++
i = encodeVarintMessages(dAtA, i, uint64(len(m.DepositContractAddress)))
i += copy(dAtA[i:], m.DepositContractAddress)
}
if m.XXX_unrecognized != nil {
i += copy(dAtA[i:], m.XXX_unrecognized)
}
return i, nil
}
func encodeVarintMessages(dAtA []byte, offset int, v uint64) int {
for v >= 1<<7 {
dAtA[offset] = uint8(v&0x7f | 0x80)
@@ -2828,6 +2905,22 @@ func (m *ExitResponse) Size() (n int) {
return n
}
func (m *Handshake) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
l = len(m.DepositContractAddress)
if l > 0 {
n += 1 + l + sovMessages(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
func sovMessages(x uint64) (n int) {
for {
n++
@@ -5498,6 +5591,92 @@ func (m *ExitResponse) Unmarshal(dAtA []byte) error {
}
return nil
}
func (m *Handshake) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowMessages
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: Handshake: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: Handshake: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field DepositContractAddress", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowMessages
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthMessages
}
postIndex := iNdEx + intStringLen
if postIndex < 0 {
return ErrInvalidLengthMessages
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.DepositContractAddress = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipMessages(dAtA[iNdEx:])
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthMessages
}
if (iNdEx + skippy) < 0 {
return ErrInvalidLengthMessages
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func skipMessages(dAtA []byte) (n int, err error) {
l := len(dAtA)
iNdEx := 0

4
proto/beacon/p2p/v1/messages.proto
View File

@@ -138,3 +138,7 @@ message ExitResponse {
bytes hash = 1;
VoluntaryExit voluntary_exit = 2;
}
message Handshake {
string deposit_contract_address = 1;
}

5
shared/p2p/BUILD.bazel
View File

@@ -7,9 +7,11 @@ go_library(
"dial_relay_node.go",
"discovery.go",
"feed.go",
"handshake_handler.go",
"interfaces.go",
"message.go",
"monitoring.go",
"negotiation.go",
"options.go",
"p2p.go",
"service.go",
@@ -54,6 +56,7 @@ go_test(
"feed_test.go",
"message_test.go",
"monitoring_test.go",
"negotiation_test.go",
"options_test.go",
"register_topic_example_test.go",
"service_test.go",
@@ -71,6 +74,8 @@ go_test(
"@com_github_gogo_protobuf//proto:go_default_library",
"@com_github_golang_mock//gomock:go_default_library",
"@com_github_libp2p_go_libp2p_blankhost//:go_default_library",
"@com_github_libp2p_go_libp2p_net//:go_default_library",
"@com_github_libp2p_go_libp2p_peer//:go_default_library",
"@com_github_libp2p_go_libp2p_peerstore//:go_default_library",
"@com_github_libp2p_go_libp2p_protocol//:go_default_library",
"@com_github_libp2p_go_libp2p_pubsub//:go_default_library",

14
shared/p2p/discovery.go
View File

@@ -37,11 +37,7 @@ func startDHTDiscovery(ctx context.Context, host host.Host, bootstrapAddr string
ctx, span := trace.StartSpan(ctx, "p2p_startDHTDiscovery")
defer span.End()
addr, err := iaddr.ParseString(bootstrapAddr)
if err != nil {
return err
}
peerinfo, err := ps.InfoFromP2pAddr(addr.Multiaddr())
peerinfo, err := peerInfoFromAddr(bootstrapAddr)
if err != nil {
return err
}
@@ -50,6 +46,14 @@ func startDHTDiscovery(ctx context.Context, host host.Host, bootstrapAddr string
return err
}
func peerInfoFromAddr(address string) (*ps.PeerInfo, error) {
addr, err := iaddr.ParseString(address)
if err != nil {
return nil, err
}
return ps.InfoFromP2pAddr(addr.Multiaddr())
}
// Discovery implements mDNS notifee interface.
type discovery struct {
ctx context.Context

23
shared/p2p/handshake_handler.go Normal file
View File

@@ -0,0 +1,23 @@
package p2p
import (
ggio "github.com/gogo/protobuf/io"
host "github.com/libp2p/go-libp2p-host"
inet "github.com/libp2p/go-libp2p-net"
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
)
// setHandshakeHandler to respond to requests for p2p handshake messages.
func setHandshakeHandler(host host.Host, contractAddress string) {
host.SetStreamHandler(handshakeProtocol, func(stream inet.Stream) {
defer stream.Close()
log.Debug("Handling handshake stream")
w := ggio.NewDelimitedWriter(stream)
defer w.Close()
hs := &pb.Handshake{DepositContractAddress: contractAddress}
if err := w.WriteMsg(hs); err != nil {
log.WithError(err).Error("Failed to write handshake response")
}
})
}

93
shared/p2p/negotiation.go Normal file
View File

@@ -0,0 +1,93 @@
package p2p
import (
"context"
ggio "github.com/gogo/protobuf/io"
host "github.com/libp2p/go-libp2p-host"
inet "github.com/libp2p/go-libp2p-net"
peer "github.com/libp2p/go-libp2p-peer"
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
"github.com/sirupsen/logrus"
)
const handshakeProtocol = prysmProtocolPrefix + "/handshake"
// setupPeerNegotiation adds a "Connected" event handler which checks a peer's
// handshake to ensure the peer is on the same blockchain. This currently
// checks only the deposit contract address. Some peer IDs may be excluded.
// For example, a relay or bootnode will not support the handshake protocol,
// but we would not want to disconnect from those well known peer IDs.
func setupPeerNegotiation(h host.Host, contractAddress string, exclusions []peer.ID) {
h.Network().Notify(&inet.NotifyBundle{
ConnectedF: func(net inet.Network, conn inet.Conn) {
// Must be handled in a goroutine as this callback cannot be blocking.
go func() {
// Exclude bootstrap node, relay node, etc.
for _, exclusion := range exclusions {
if conn.RemotePeer() == exclusion {
return
}
}
log.WithField("peer", conn.RemotePeer()).Debug(
"Checking connection to peer",
)
s, err := h.NewStream(
context.Background(),
conn.RemotePeer(),
handshakeProtocol,
)
if err != nil {
log.WithError(err).WithField("peer", conn.RemotePeer()).Error(
"Failed to open stream with newly connected peer",
)
if err := h.Network().ClosePeer(conn.RemotePeer()); err != nil {
log.WithError(err).Error("failed to disconnect peer")
}
return
}
defer s.Close()
w := ggio.NewDelimitedWriter(s)
defer w.Close()
hs := &pb.Handshake{DepositContractAddress: contractAddress}
if err := w.WriteMsg(hs); err != nil {
log.WithError(err).Error("Failed to write handshake to peer")
if err := h.Network().ClosePeer(conn.RemotePeer()); err != nil {
log.WithError(err).Error("failed to disconnect peer")
}
return
}
r := ggio.NewDelimitedReader(s, maxMessageSize)
resp := &pb.Handshake{}
if err := r.ReadMsg(resp); err != nil {
log.WithError(err).Error("Failed to read message")
if err := h.Network().ClosePeer(conn.RemotePeer()); err != nil {
log.WithError(err).Error("failed to disconnect peer")
}
return
}
log.WithField("msg", resp).Debug("Handshake received")
if resp.DepositContractAddress != contractAddress {
log.WithFields(logrus.Fields{
"peerContract": resp.DepositContractAddress,
"expectedContract": contractAddress,
}).Warn("Disconnecting from peer on different contract")
if err := h.Network().ClosePeer(conn.RemotePeer()); err != nil {
log.WithError(err).Error("failed to disconnect peer")
}
}
}()
},
})
}

58
shared/p2p/negotiation_test.go Normal file
View File

@@ -0,0 +1,58 @@
package p2p
import (
"context"
"testing"
"time"
bhost "github.com/libp2p/go-libp2p-blankhost"
libp2pnet "github.com/libp2p/go-libp2p-net"
peer "github.com/libp2p/go-libp2p-peer"
pstore "github.com/libp2p/go-libp2p-peerstore"
swarmt "github.com/libp2p/go-libp2p-swarm/testing"
)
func TestNegotiation_AcceptsValidPeer(t *testing.T) {
ctx := context.Background()
hostA := bhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
hostB := bhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
address := "same"
setHandshakeHandler(hostA, address)
setHandshakeHandler(hostB, address)
setupPeerNegotiation(hostA, address, []peer.ID{})
setupPeerNegotiation(hostB, address, []peer.ID{})
if err := hostA.Connect(ctx, pstore.PeerInfo{ID: hostB.ID(), Addrs: hostB.Addrs()}); err != nil {
t.Fatal(err)
}
// Allow short delay for async negotiation.
time.Sleep(200 * time.Millisecond)
if hostA.Network().Connectedness(hostB.ID()) != libp2pnet.Connected {
t.Error("hosts are not connected")
}
}
func TestNegotiation_DisconnectsDifferentDepositContract(t *testing.T) {
ctx := context.Background()
hostA := bhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
hostB := bhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
setHandshakeHandler(hostA, "hostA")
setHandshakeHandler(hostB, "hostB")
setupPeerNegotiation(hostA, "hostA", []peer.ID{})
setupPeerNegotiation(hostB, "hostB", []peer.ID{})
if err := hostA.Connect(ctx, pstore.PeerInfo{ID: hostB.ID(), Addrs: hostB.Addrs()}); err != nil {
t.Fatal(err)
}
// Allow short delay for async negotiation.
time.Sleep(200 * time.Millisecond)
if hostA.Network().Connectedness(hostB.ID()) == libp2pnet.Connected {
t.Error("hosts are connected, but should not be connected")
}
}

25
shared/p2p/service.go
View File

@@ -60,10 +60,11 @@ type Server struct {
// ServerConfig for peer to peer networking.
type ServerConfig struct {
NoDiscovery bool
BootstrapNodeAddr string
RelayNodeAddr string
Port int
NoDiscovery bool
BootstrapNodeAddr string
RelayNodeAddr string
Port int
DepositContractAddress string
}
// NewServer creates a new p2p server instance.
@@ -103,6 +104,22 @@ func NewServer(cfg *ServerConfig) (*Server, error) {
return nil, err
}
// Blockchain peering negitiation; excludes negotiating with bootstrap or
// relay nodes.
exclusions := []peer.ID{}
for _, addr := range []string{cfg.BootstrapNodeAddr, cfg.RelayNodeAddr} {
if addr == "" {
continue
}
info, err := peerInfoFromAddr(addr)
if err != nil {
return nil, err
}
exclusions = append(exclusions, info.ID)
}
setupPeerNegotiation(h, cfg.DepositContractAddress, exclusions)
setHandshakeHandler(h, cfg.DepositContractAddress)
return &Server{
ctx: ctx,
cancel: cancel,

15
shared/p2p/service_test.go
View File

@@ -39,19 +39,14 @@ func init() {
logrus.SetLevel(logrus.DebugLevel)
}
func TestStartDialRelayNode_InvalidMultiaddress(t *testing.T) {
hook := logTest.NewGlobal()
s, err := NewServer(&ServerConfig{
func TestNewServer_InvalidMultiaddress(t *testing.T) {
_, err := NewServer(&ServerConfig{
RelayNodeAddr: "bad",
})
if err != nil {
t.Fatalf("Unable to create server: %v", err)
if err.Error() != "invalid multiaddr, must begin with /" {
t.Fatal("expected invalid multiaddr err")
}
s.Start()
logContains(t, hook, "Could not dial relay node: invalid multiaddr, must begin with /", logrus.ErrorLevel)
}
func TestP2P_PortTaken(t *testing.T) {
@@ -448,6 +443,8 @@ func simulateIncomingMessage(t *testing.T, s *Server, topic string, msg proto.Me
ctx := context.Background()
h := bhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
setHandshakeHandler(h, "")
gsub, err := pubsub.NewFloodSub(ctx, h)
if err != nil {
return err