diff --git a/sei-tendermint/autobahn/types/msg.go b/sei-tendermint/autobahn/types/msg.go index 8f80dd244a..75b3829cf8 100644 --- a/sei-tendermint/autobahn/types/msg.go +++ b/sei-tendermint/autobahn/types/msg.go @@ -83,6 +83,11 @@ func (k SecretKey) Public() PublicKey { return PublicKey{key: k.key.Public()} } +// SignWithTag signs msg with a domain-separation tag. +func (k SecretKey) SignWithTag(tag ed25519.Tag, msg []byte) ed25519.Signature { + return k.key.SignWithTag(tag, msg) +} + // PublicKey is the public key of the validator. // nolint:recvcheck type PublicKey struct { @@ -99,6 +104,11 @@ func (k PublicKey) Bytes() []byte { return k.key.Bytes() } // ED25519 returns the underlying Ed25519 public key. func (k PublicKey) ED25519() ed25519.PublicKey { return k.key } +// VerifyWithTag verifies a tagged signature. +func (k PublicKey) VerifyWithTag(tag ed25519.Tag, msg []byte, sig ed25519.Signature) error { + return k.key.VerifyWithTag(tag, msg, sig) +} + // PublicKeyFromBytes constructs a public key from bytes. func PublicKeyFromBytes(b []byte) (PublicKey, error) { k, err := ed25519.PublicKeyFromBytes(b) diff --git a/sei-tendermint/config/autobahn.go b/sei-tendermint/config/autobahn.go index d087af99b8..b5c4481988 100644 --- a/sei-tendermint/config/autobahn.go +++ b/sei-tendermint/config/autobahn.go @@ -16,11 +16,14 @@ type URL struct{ *url.URL } func (u URL) MarshalText() ([]byte, error) { return []byte(u.String()), nil } func (u *URL) UnmarshalText(text []byte) error { - url, err := url.Parse(string(text)) + parsed, err := url.Parse(string(text)) if err != nil { return err } - u.URL = url + if err := utils.CheckHTTPURL(*parsed); err != nil { + return err + } + u.URL = parsed return nil } @@ -107,6 +110,9 @@ func (fc *AutobahnFileConfig) Validate() error { if v.EVMRPC.URL == nil { return fmt.Errorf("validator %s is missing evmrpc URL", v.ValidatorKey) } + if err := utils.CheckHTTPURL(*v.EVMRPC.URL); err != nil { + return fmt.Errorf("validator %s evmrpc: %w", v.ValidatorKey, err) + } } if fc.MaxTxsPerBlock == 0 { return errors.New("max_txs_per_block must be > 0") diff --git a/sei-tendermint/config/autobahn_test.go b/sei-tendermint/config/autobahn_test.go index d9bf43f679..ee3915c8ac 100644 --- a/sei-tendermint/config/autobahn_test.go +++ b/sei-tendermint/config/autobahn_test.go @@ -27,6 +27,11 @@ func TestURLJSONReencode(t *testing.T) { require.Equal(t, want.String(), got.String()) } +func TestURLUnmarshalRejectsNonHTTP(t *testing.T) { + var got URL + require.Error(t, json.Unmarshal([]byte(`"ws://example.com:8545"`), &got)) +} + func TestAutobahnBlockDBConfig_LittBlockConfig(t *testing.T) { dir := t.TempDir() const ( diff --git a/sei-tendermint/internal/p2p/conv.go b/sei-tendermint/internal/p2p/conv.go index e42bfabf69..bbb4ee7336 100644 --- a/sei-tendermint/internal/p2p/conv.go +++ b/sei-tendermint/internal/p2p/conv.go @@ -3,8 +3,10 @@ package p2p import ( "errors" "fmt" + "net/url" "strings" + atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" "github.com/sei-protocol/sei-chain/sei-tendermint/crypto/ed25519" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/conn" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/pb" @@ -84,6 +86,15 @@ var nodePublicKeyConv = protoutils.Conv[NodePublicKey, *pb.NodePublicKey]{ }, } +// gigaHandshakeClaim is the Autobahn committee identity a validator advertises +// on the giga handshake. +type gigaHandshakeClaim struct { + utils.ReadOnly + Validator atypes.PublicKey + sig ed25519.Signature + evmRPC string +} + type handshakeSpec struct { SelfAddr utils.Option[NodeAddress] PexAddrs []NodeAddress @@ -93,6 +104,7 @@ type handshakeSpec struct { type handshakeMsg struct { NodeAuth NodeChallengeSig handshakeSpec + GigaClaim utils.Option[gigaHandshakeClaim] } var handshakeMsgConv = protoutils.Conv[*handshakeMsg, *pb.Handshake]{ @@ -105,14 +117,19 @@ var handshakeMsgConv = protoutils.Conv[*handshakeMsg, *pb.Handshake]{ for i, addr := range m.PexAddrs { pexAddrs[i] = addr.String() } - - return &pb.Handshake{ + out := &pb.Handshake{ NodeAuthKey: nodePublicKeyConv.Encode(m.NodeAuth.Key()), NodeAuthSig: m.NodeAuth.sig.Bytes(), SelfAddr: selfAddr, PexAddrs: pexAddrs, SeiGigaConnection: m.SeiGigaConnection, } + if claim, ok := m.GigaClaim.Get(); ok { + out.ValidatorAuthKey = claim.Validator.Bytes() + out.ValidatorAuthSig = claim.sig.Bytes() + out.EvmRpc = utils.Alloc(claim.evmRPC) + } + return out }, Decode: func(p *pb.Handshake) (*handshakeMsg, error) { nodeAuthKey, err := nodePublicKeyConv.DecodeReq(p.NodeAuthKey) @@ -139,6 +156,10 @@ var handshakeMsgConv = protoutils.Conv[*handshakeMsg, *pb.Handshake]{ } pexAddrs[i] = addr } + claim, err := decodeGigaClaim(p) + if err != nil { + return nil, err + } return &handshakeMsg{ NodeAuth: NodeChallengeSig{key: nodeAuthKey, sig: nodeAuthSig}, handshakeSpec: handshakeSpec{ @@ -146,6 +167,37 @@ var handshakeMsgConv = protoutils.Conv[*handshakeMsg, *pb.Handshake]{ PexAddrs: pexAddrs, SeiGigaConnection: p.SeiGigaConnection, }, + GigaClaim: claim, }, nil }, } + +func decodeGigaClaim(p *pb.Handshake) (utils.Option[gigaHandshakeClaim], error) { + isValidator := p.ValidatorAuthKey != nil || p.ValidatorAuthSig != nil || p.EvmRpc != nil + if !isValidator { + return utils.None[gigaHandshakeClaim](), nil + } + if p.ValidatorAuthKey == nil || p.ValidatorAuthSig == nil || p.EvmRpc == nil { + return utils.None[gigaHandshakeClaim](), fmt.Errorf("giga claim requires validator_auth_key, validator_auth_sig, and evm_rpc") + } + valKey, err := atypes.PublicKeyFromBytes(p.ValidatorAuthKey) + if err != nil { + return utils.None[gigaHandshakeClaim](), fmt.Errorf("ValidatorAuthKey: %w", err) + } + valSig, err := ed25519.SignatureFromBytes(p.ValidatorAuthSig) + if err != nil { + return utils.None[gigaHandshakeClaim](), fmt.Errorf("ValidatorAuthSig: %w", err) + } + u, err := url.Parse(*p.EvmRpc) + if err != nil { + return utils.None[gigaHandshakeClaim](), fmt.Errorf("EvmRpc: %w", err) + } + if err := utils.CheckHTTPURL(*u); err != nil { + return utils.None[gigaHandshakeClaim](), fmt.Errorf("EvmRpc: %w", err) + } + return utils.Some(gigaHandshakeClaim{ + Validator: valKey, + sig: valSig, + evmRPC: *p.EvmRpc, + }), nil +} diff --git a/sei-tendermint/internal/p2p/conv_test.go b/sei-tendermint/internal/p2p/conv_test.go index 8aa4f47da5..4854307b2a 100644 --- a/sei-tendermint/internal/p2p/conv_test.go +++ b/sei-tendermint/internal/p2p/conv_test.go @@ -1,9 +1,15 @@ package p2p import ( + "strings" "testing" + "google.golang.org/protobuf/proto" + + atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/conn" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/pb" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/protoutils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" ) @@ -28,6 +34,72 @@ func TestHandshakeMsgConv(t *testing.T) { } } +func TestDecodeGigaClaimEVMRPC(t *testing.T) { + rng := utils.TestRng() + key := atypes.GenSecretKey(rng) + sig := key.SignWithTag(gigaValidatorHandshakeTag, []byte("x")) + claim := func(evm string) *pb.Handshake { + return &pb.Handshake{ + ValidatorAuthKey: key.Public().Bytes(), + ValidatorAuthSig: sig.Bytes(), + EvmRpc: &evm, + } + } + for _, s := range []string{"http://validator.example:8545", "https://validator.example:8545"} { + got, err := decodeGigaClaim(claim(s)) + require.NoError(t, err) + require.Equal(t, s, got.OrPanic("missing claim").evmRPC) + } + for _, s := range []string{"file://localhost/rpc", "ws://validator.example:8545", "http://", "http://u:p@validator.example:8545"} { + _, err := decodeGigaClaim(claim(s)) + require.Error(t, err) + } +} + +func TestDecodeGigaClaimRequiresAllFields(t *testing.T) { + rng := utils.TestRng() + key := atypes.GenSecretKey(rng) + sig := key.SignWithTag(gigaValidatorHandshakeTag, []byte("x")) + evmRPC := "http://validator.example:8545" + full := func() *pb.Handshake { + return &pb.Handshake{ + ValidatorAuthKey: key.Public().Bytes(), + ValidatorAuthSig: sig.Bytes(), + EvmRpc: &evmRPC, + } + } + for _, drop := range []func(*pb.Handshake){ + func(p *pb.Handshake) { p.ValidatorAuthKey = nil }, + func(p *pb.Handshake) { p.ValidatorAuthSig = nil }, + func(p *pb.Handshake) { p.EvmRpc = nil }, + func(p *pb.Handshake) { p.ValidatorAuthKey, p.ValidatorAuthSig = nil, nil }, + func(p *pb.Handshake) { p.ValidatorAuthKey, p.EvmRpc = nil, nil }, + func(p *pb.Handshake) { p.ValidatorAuthSig, p.EvmRpc = nil, nil }, + } { + p := full() + drop(p) + _, err := decodeGigaClaim(p) + require.Error(t, err) + } + + // None of the three is a peer without a claim, rather than a malformed one. + got, err := decodeGigaClaim(&pb.Handshake{}) + require.NoError(t, err) + require.False(t, got.IsPresent()) +} + +func TestHandshakeWireguardRejectsOversizedEvmRPC(t *testing.T) { + tooLong := strings.Repeat("a", 2049) + raw, err := proto.Marshal(&pb.Handshake{EvmRpc: &tooLong}) + require.NoError(t, err) + require.Error(t, protoutils.Scan[*pb.Handshake](raw)) + + ok := strings.Repeat("a", 2048) + raw, err = proto.Marshal(&pb.Handshake{EvmRpc: &ok}) + require.NoError(t, err) + require.NoError(t, protoutils.Scan[*pb.Handshake](raw)) +} + func TestNodePublicKeyFromString(t *testing.T) { rng := utils.TestRng() key := makeKey(rng).Public() diff --git a/sei-tendermint/internal/p2p/giga_router.go b/sei-tendermint/internal/p2p/giga_router.go index 233bec4a66..a23498ef50 100644 --- a/sei-tendermint/internal/p2p/giga_router.go +++ b/sei-tendermint/internal/p2p/giga_router.go @@ -20,7 +20,7 @@ import ( type GigaNodeAddr struct { Key NodePublicKey HostPort tcp.HostPort - EVMRPC *url.URL + EVMRPC url.URL } func (a GigaNodeAddr) String() string { @@ -65,7 +65,8 @@ type GigaValidatorConfig struct { // GigaRouter is the read-path / Run / EvmProxy surface. Implemented by // *gigaValidatorRouter and *gigaFullnodeRouter; Mempool returns Some only // on validators. RunInboundConn is served by both — non-committee peers -// get the block-sync subset only. +// get the block-sync subset only. A fullnode accepts committee peers but +// has no consensus state to serve them. type GigaRouter interface { Run(ctx context.Context) error RunInboundConn(ctx context.Context, hConn *handshakedConn) error @@ -76,4 +77,5 @@ type GigaRouter interface { EvmProxy(sender common.Address) utils.Option[*rpc.Client] Mempool() utils.Option[*producer.State] Validators(n atypes.GlobalBlockNumber) ([]*types.Validator, atypes.GlobalBlockNumber, error) + fillInboundHandshake(spec handshakeSpec) (handshakeSpec, utils.Option[handshakeOffer]) } diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go index 56b04a87f9..0e4d44f746 100644 --- a/sei-tendermint/internal/p2p/giga_router_common.go +++ b/sei-tendermint/internal/p2p/giga_router_common.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "net/url" "path/filepath" "slices" "sort" @@ -33,15 +34,22 @@ import ( // you need more. const maxInboundFullnodePeers = 10000 +var errGigaMembershipChanged = errors.New("inbound giga peer changed committee membership") + type gigaRouterCommon struct { - cfg *GigaRouterCommonConfig - key NodeSecretKey - data *data.State - service *giga.Service - poolIn *giga.Pool[NodePublicKey, rpc.Server[giga.API]] - poolOut *giga.Pool[NodePublicKey, rpc.Client[giga.API]] - proxies utils.RWMutex[map[atypes.PublicKey]*ethrpc.Client] - app *proxy.Proxy + cfg *GigaRouterCommonConfig + key NodeSecretKey + data *data.State + service *giga.Service + poolIn *giga.Pool[NodePublicKey, rpc.Server[giga.API]] + poolInCommittee *giga.Pool[atypes.PublicKey, rpc.Server[giga.API]] + poolOut *giga.Pool[atypes.PublicKey, rpc.Client[giga.API]] + proxies utils.RWMutex[map[atypes.PublicKey]*ethrpc.Client] + app *proxy.Proxy + offer utils.Option[handshakeOffer] + selfAddr utils.Option[NodeAddress] + liveAddrs utils.RWMutex[map[atypes.PublicKey]GigaNodeAddr] + liveAddrVersion utils.AtomicSend[uint64] // nextCommitEpoch is data.NextCommitEpoch() cached at construction so // EvmProxy can Load() without taking the data lock on every call. nextCommitEpoch utils.AtomicRecv[*atypes.Epoch] @@ -56,6 +64,14 @@ type gigaRouterCommon struct { inboundFullnodeCap int64 } +func (r *gigaRouterCommon) fillInboundHandshake(spec handshakeSpec) (handshakeSpec, utils.Option[handshakeOffer]) { + spec.SeiGigaConnection = true + if r.selfAddr.IsPresent() { + spec.SelfAddr = r.selfAddr + } + return spec, r.offer +} + // BuildDataState validates the common config, constructs the committee, and // returns an initialised data.State backed by blockStore. // @@ -472,13 +488,13 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { // dialAndRunConn dials a peer, handshakes as a SeiGiga connection, // registers the rpc client in poolOut, and runs runClient for the -// connection's lifetime. expectedKey is enforced when Some (validator -// dialing a committee member); fullnodes pass None — block-sync data -// is QC-verified, so the peer's identity doesn't need to be checked -// here. +// connection's lifetime. It verifies the p2p node key against the selected +// address, requires a validator claim for expectedValidatorKey, and registers +// the client under that validator. func (r *gigaRouterCommon) dialAndRunConn( ctx context.Context, - expectedKey utils.Option[NodePublicKey], + expectedValidatorKey atypes.PublicKey, + expectedNodeKey NodePublicKey, hp tcp.HostPort, runClient func(ctx context.Context, client rpc.Client[giga.API]) error, ) error { @@ -496,7 +512,10 @@ func (r *gigaRouterCommon) dialAndRunConn( } s.SpawnBg(func() error { return tcpConn.Run(ctx) }) // TODO: handshake needs a timeout. - hConn, err := handshake(ctx, tcpConn, r.key, handshakeSpec{SeiGigaConnection: true}) + hConn, err := handshake(ctx, tcpConn, r.key, handshakeSpec{ + SelfAddr: r.selfAddr, + SeiGigaConnection: true, + }, r.offer) if err != nil { return fmt.Errorf("handshake(): %w", err) } @@ -504,11 +523,18 @@ func (r *gigaRouterCommon) dialAndRunConn( return fmt.Errorf("not a sei giga connection") } peerKey := hConn.msg.NodeAuth.Key() - if want, ok := expectedKey.Get(); ok && peerKey != want { - return fmt.Errorf("peer key = %v, want %v", peerKey, want) + if peerKey != expectedNodeKey { + return fmt.Errorf("peer node key = %v, want %v", peerKey, expectedNodeKey) + } + claim, ok := hConn.msg.GigaClaim.Get() + if !ok { + return fmt.Errorf("committee member %v: %w", expectedValidatorKey, errMissingGigaClaim) + } + if claim.Validator != expectedValidatorKey { + return fmt.Errorf("peer validator key = %v, want %v", claim.Validator, expectedValidatorKey) } client := rpc.NewClient[giga.API]() - return r.poolOut.InsertAndRun(ctx, peerKey, client, func(ctx context.Context) error { + return r.poolOut.InsertAndRun(ctx, expectedValidatorKey, client, func(ctx context.Context) error { return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { s.Spawn(func() error { return client.Run(ctx, hConn.conn) }) Global.gigaNewConnsAt("out").Add(1) @@ -521,14 +547,15 @@ func (r *gigaRouterCommon) dialAndRunConn( } // committeeMemberTask is work for one reachable committee member. It must run -// until ctx is cancelled; returning earlier leaves the member unmarked in live -// and it is not restarted while it stays in the committee. +// until ctx is cancelled; otherwise it is not restarted while the member stays +// in the committee. type committeeMemberTask func(ctx context.Context, validator atypes.PublicKey, addr GigaNodeAddr) error // memberSession is a committee member's cancellable task session. type memberSession struct { cancel context.CancelFunc done chan struct{} + addr GigaNodeAddr } // keepReplicas is commitEpoch's committee, plus the committee Anchor covers when present. @@ -545,30 +572,105 @@ func keepReplicas(anchor utils.Option[data.Anchor], commitEpoch *atypes.Epoch) m return keep } -// stopDepartingMembers stops sessions for validators outside keepReplicas once -// Anchor is at most one epoch behind commitEpoch. -func stopDepartingMembers( +// validatorAddr returns the address to dial validator at. The configured +// committee book wins; the live overlay only covers members it omits. +func (r *gigaRouterCommon) validatorAddr(validator atypes.PublicKey) (GigaNodeAddr, bool) { + if addr, ok := r.cfg.ValidatorAddrs[validator]; ok { + return addr, true + } + for addrs := range r.liveAddrs.RLock() { + if addr, ok := addrs[validator]; ok { + return addr, true + } + } + return GigaNodeAddr{}, false +} + +func sameGigaNodeAddr(a, b GigaNodeAddr) bool { + return a.Key == b.Key && a.HostPort == b.HostPort && a.EVMRPC.String() == b.EVMRPC.String() +} + +// acceptInbound returns the committee identity a verified inbound connection +// proved, recording the giga address it advertised. None means the peer is +// served the block-sync subset: it carried no claim, or it claimed a validator +// outside the current commit committee. +func (r *gigaRouterCommon) acceptInbound(hConn *handshakedConn) utils.Option[atypes.PublicKey] { + claim, ok := hConn.msg.GigaClaim.Get() + if !ok { + return utils.None[atypes.PublicKey]() + } + if !r.nextCommitEpoch.Load().Committee().HasReplica(claim.Validator) { + return utils.None[atypes.PublicKey]() + } + selfAddr := hConn.msg.SelfAddr.OrPanic("verified giga claim has no SelfAddr") + evmRPC := *utils.OrPanic1(url.Parse(claim.evmRPC)) + // If we will not use its EVMRPC, we learn none of the advertisement. + if utils.IsLoopbackOrLinkLocalURL(evmRPC) { + logger.Error("committee member advertised an unroutable EVM RPC; not learning its address", + "validator", claim.Validator, "evmRPC", evmRPC.String()) + return utils.Some(claim.Validator) + } + addr := GigaNodeAddr{ + Key: hConn.msg.NodeAuth.Key(), + HostPort: tcp.HostPort{Hostname: selfAddr.Hostname, Port: selfAddr.Port}, + EVMRPC: evmRPC, + } + for addrs := range r.liveAddrs.Lock() { + if old, ok := addrs[claim.Validator]; ok && sameGigaNodeAddr(old, addr) { + return utils.Some(claim.Validator) + } + addrs[claim.Validator] = addr + r.liveAddrVersion.Store(r.liveAddrVersion.Load() + 1) + } + logger.Info("learned validator giga address", "validator", claim.Validator, "addr", addr) + return utils.Some(claim.Validator) +} + +// stopStaleSessions stops sessions for validators outside keepReplicas or +// dialing an address that is no longer current, and drops the overlay addresses +// of the departed. Departures are only acted on once Anchor is at most one +// epoch behind commitEpoch. +func (r *gigaRouterCommon) stopStaleSessions( ctx context.Context, live map[atypes.PublicKey]*memberSession, anchor utils.Option[data.Anchor], commitEpoch *atypes.Epoch, ) error { - a, ok := anchor.Get() - if !ok || commitEpoch.EpochIndex() > a.Epoch.EpochIndex()+1 { - return nil - } + a, hasAnchor := anchor.Get() + // Until Anchor is within one epoch, validators of the epochs in between are + // in neither endpoint committee, and the AppQCs for those epochs cannot form + // without them. + settled := hasAnchor && commitEpoch.EpochIndex() <= a.Epoch.EpochIndex()+1 keep := keepReplicas(anchor, commitEpoch) - var departing []*memberSession - // Cancel every departing session before waiting for any of them. + if settled { + for addrs := range r.liveAddrs.Lock() { + dropped := false + for validator := range addrs { + if _, ok := keep[validator]; !ok { + delete(addrs, validator) + dropped = true + } + } + if dropped { + r.liveAddrVersion.Store(r.liveAddrVersion.Load() + 1) + } + } + } + var stale []*memberSession + // Cancel every stale session before waiting for any of them. for validator, session := range live { - if _, ok := keep[validator]; ok { + if _, kept := keep[validator]; !kept { + if !settled { + continue + } + } else if addr, ok := r.validatorAddr(validator); ok && sameGigaNodeAddr(session.addr, addr) { continue } session.cancel() - departing = append(departing, session) + stale = append(stale, session) delete(live, validator) } - for _, session := range departing { + for _, session := range stale { if _, _, err := utils.RecvOrClosed(ctx, session.done); err != nil { return err } @@ -580,6 +682,7 @@ func stopDepartingMembers( func (r *gigaRouterCommon) runPerCommitteeMember(ctx context.Context, tasks ...committeeMemberTask) error { return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { live := map[atypes.PublicKey]*memberSession{} + addrUpdates := r.liveAddrVersion.Subscribe() // End all sessions before the scope waits for them. defer func() { for _, session := range live { @@ -592,23 +695,24 @@ func (r *gigaRouterCommon) runPerCommitteeMember(ctx context.Context, tasks ...c return utils.MapOpt(opt, func(a data.Anchor) atypes.EpochIndex { return a.Epoch.EpochIndex() }) } for ctx.Err() == nil { + addrVersion := addrUpdates.Load() commitEpoch := r.nextCommitEpoch.Load() anchor := r.anchor.Load() - if err := stopDepartingMembers(ctx, live, anchor, commitEpoch); err != nil { + if err := r.stopStaleSessions(ctx, live, anchor, commitEpoch); err != nil { return err } for validator := range keepReplicas(anchor, commitEpoch) { if _, ok := live[validator]; ok { continue } - addr, ok := r.cfg.ValidatorAddrs[validator] + addr, ok := r.validatorAddr(validator) if !ok { logger.Error("committee member has no configured address; not dialing", "validator", validator) continue } taskCtx, cancel := context.WithCancel(ctx) done := make(chan struct{}) - live[validator] = &memberSession{cancel: cancel, done: done} + live[validator] = &memberSession{cancel: cancel, done: done, addr: addr} s.SpawnNamed(addr.String(), func() error { defer close(done) return utils.IgnoreCancel(scope.Run(taskCtx, func(ctx context.Context, ms scope.Scope) error { @@ -621,8 +725,9 @@ func (r *gigaRouterCommon) runPerCommitteeMember(ctx context.Context, tasks ...c } if err := utils.WaitAny(ctx, func() bool { return r.nextCommitEpoch.Load().EpochIndex() != commitEpoch.EpochIndex() || - epochOf(r.anchor.Load()) != epochOf(anchor) - }, r.nextCommitEpoch, r.anchor); err != nil { + epochOf(r.anchor.Load()) != epochOf(anchor) || + addrUpdates.Load() != addrVersion + }, r.nextCommitEpoch, r.anchor, addrUpdates); err != nil { return err } } @@ -630,18 +735,17 @@ func (r *gigaRouterCommon) runPerCommitteeMember(ctx context.Context, tasks ...c }) } -// runUntilMembershipChange runs f while validator's membership matches -// isCommittee. It reports whether a membership change ended f. +// runUntilMembershipChange runs f until validator leaves the current commit +// committee. It reports whether a leave ended f. func (r *gigaRouterCommon) runUntilMembershipChange( ctx context.Context, validator atypes.PublicKey, - isCommittee bool, f func(ctx context.Context) error, ) (changed bool, err error) { err = scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { s.SpawnBg(func() error { _, err := r.nextCommitEpoch.Wait(ctx, func(epoch *atypes.Epoch) bool { - return epoch.Committee().HasReplica(validator) != isCommittee + return !epoch.Committee().HasReplica(validator) }) if err != nil { return err @@ -655,73 +759,79 @@ func (r *gigaRouterCommon) runUntilMembershipChange( return changed, utils.IgnoreCancel(err) } -// RunInboundConn serves an inbound giga connection. Non-committee peers -// get the block-sync subset (StreamFullCommitQCs + GetBlock). Committee peers -// get the full RunServer on validators; on a fullnode the connection is refused. -// -// The role and the fullnode cap are fixed for the lifetime of the connection: a -// membership change ends it, and the peer's dialer reconnects into the role it -// now has. +// RunInboundConn serves an inbound giga connection. A peer proving current +// commit-committee membership is served as a validator; every other peer is +// served as a fullnode for the life of this socket. func (r *gigaRouterCommon) RunInboundConn(ctx context.Context, hConn *handshakedConn) error { if !hConn.msg.SeiGigaConnection { return fmt.Errorf("not a SeiGiga connection") } - // Filter unwanded connections. - key := hConn.msg.NodeAuth.Key() - // TODO: support committee members absent from the address book. - validator := utils.None[atypes.PublicKey]() - for v, addr := range r.cfg.ValidatorAddrs { - if addr.Key == key { - validator = utils.Some(v) - break - } - } - // Inbound role follows nextCommitEpoch only. AppVotes are received on the - // outbound client stream, not this mux, so a departing peer can be - // downgraded here while outbound sessions still collect its votes. - isCommittee := false - if v, ok := validator.Get(); ok { - isCommittee = r.nextCommitEpoch.Load().Committee().HasReplica(v) - } - if !isCommittee { - // Optimistic acquire: Add(1), compare, Add(-1) on overflow. Acquired - // before InsertAndRun, which evicts any live connection for this key. - if r.inboundFullnodeCount.Add(1) > r.inboundFullnodeCap { - r.inboundFullnodeCount.Add(-1) - return fmt.Errorf("inbound fullnode peer limit (%d) reached", r.inboundFullnodeCap) - } - defer r.inboundFullnodeCount.Add(-1) + if member, ok := r.acceptInbound(hConn).Get(); ok { + return r.runInboundValidator(ctx, hConn, member) } + // A member who inbounds before they appear in our nextCommitEpoch view is + // served as a fullnode for this socket's life; their own dialer redials + // after DialInterval and re-handshakes into the validator role. + return r.runInboundFullnode(ctx, hConn) +} + +func (r *gigaRouterCommon) runInboundFullnode(ctx context.Context, hConn *handshakedConn) error { + key := hConn.msg.NodeAuth.Key() server := rpc.NewServer[giga.API]() + // Optimistic acquire: Add(1), compare, Add(-1) on overflow. Acquired + // before InsertAndRun, which evicts any live connection for this key. + if r.inboundFullnodeCount.Add(1) > r.inboundFullnodeCap { + r.inboundFullnodeCount.Add(-1) + return fmt.Errorf("inbound fullnode peer limit (%d) reached", r.inboundFullnodeCap) + } + defer r.inboundFullnodeCount.Add(-1) return r.poolIn.InsertAndRun(ctx, key, server, func(ctx context.Context) error { - return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { - // Background: a membership change must cancel the mux. Spawn would - // keep this scope alive until the peer closes the socket. - s.SpawnBg(func() error { return server.Run(ctx, hConn.conn) }) - Global.gigaNewConnsAt("in").Add(1) - Global.gigaConnsAt("in").Add(1) - defer Global.gigaConnsAt("in").Add(-1) - v, ok := validator.Get() - if !ok { - if err := r.service.RunServer(ctx, server, false); err != nil { - return fmt.Errorf("inbound from %v: %w", key, err) - } - return nil - } - changed, err := r.runUntilMembershipChange(ctx, v, isCommittee, func(ctx context.Context) error { - return r.service.RunServer(ctx, server, isCommittee) + return r.runInboundMux(ctx, server, hConn, func(ctx context.Context) error { + return r.service.RunServer(ctx, server, false) + }) + }) +} + +func (r *gigaRouterCommon) runInboundValidator(ctx context.Context, hConn *handshakedConn, member atypes.PublicKey) error { + key := hConn.msg.NodeAuth.Key() + server := rpc.NewServer[giga.API]() + return r.poolInCommittee.InsertAndRun(ctx, member, server, func(ctx context.Context) error { + return r.runInboundMux(ctx, server, hConn, func(ctx context.Context) error { + changed, err := r.runUntilMembershipChange(ctx, member, func(ctx context.Context) error { + return r.service.RunServer(ctx, server, true) }) if err != nil { - return fmt.Errorf("inbound from %v: %w", key, err) + return err } if changed { - logger.Info("inbound giga peer changed committee membership; closing", "addr", key, "was_committee", isCommittee) + logger.Info("inbound giga peer left the committee; closing", "validator", member, "addr", key) + return errGigaMembershipChanged } return nil }) }) } +func (r *gigaRouterCommon) runInboundMux( + ctx context.Context, + server rpc.Server[giga.API], + hConn *handshakedConn, + serve func(ctx context.Context) error, +) error { + return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + // Background: a membership change must cancel the mux. Spawn would + // keep this scope alive until the peer closes the socket. + s.SpawnBg(func() error { return server.Run(ctx, hConn.conn) }) + Global.gigaNewConnsAt("in").Add(1) + Global.gigaConnsAt("in").Add(1) + defer Global.gigaConnsAt("in").Add(-1) + if err := serve(ctx); err != nil { + return fmt.Errorf("inbound from %v: %w", hConn.msg.NodeAuth.Key(), err) + } + return nil + }) +} + // Validators returns the Autobahn validator set that certified global height n. // Before the first CommitQC, FirstBlock resolves to the genesis committee. func (r *gigaRouterCommon) Validators(n atypes.GlobalBlockNumber) ([]*types.Validator, atypes.GlobalBlockNumber, error) { diff --git a/sei-tendermint/internal/p2p/giga_router_common_test.go b/sei-tendermint/internal/p2p/giga_router_common_test.go index 9752702fb8..b20d2427c4 100644 --- a/sei-tendermint/internal/p2p/giga_router_common_test.go +++ b/sei-tendermint/internal/p2p/giga_router_common_test.go @@ -2,6 +2,7 @@ package p2p import ( "context" + "errors" "fmt" "net/url" "sync/atomic" @@ -18,15 +19,18 @@ import ( "github.com/sei-protocol/sei-chain/sei-tendermint/crypto/ed25519" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/data" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/epoch" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/giga" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/rpc" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/tcp" "github.com/sei-protocol/sei-chain/sei-tendermint/rpc/coretypes" tmtypes "github.com/sei-protocol/sei-chain/sei-tendermint/types" ) -func registerEvmProxyForTest(t *testing.T, router *gigaRouterCommon, validator atypes.PublicKey, rpcURL *url.URL) *ethrpc.Client { +func registerEvmProxyForTest(t *testing.T, router *gigaRouterCommon, validator atypes.PublicKey, rpcURL url.URL) *ethrpc.Client { t.Helper() client, err := ethrpc.DialContext(t.Context(), rpcURL.String()) require.NoError(t, err) @@ -257,8 +261,8 @@ func testAnchor(ep *atypes.Epoch) utils.Option[data.Anchor] { return utils.Some(data.Anchor{Epoch: ep}) } -// settledEpochs drives both keep-set inputs to the same epoch. Tests of the -// window between them send to nextCommitEpoch and anchor separately. +// settledEpochs drives both keep-set inputs to the same epoch. A one-epoch lag +// (commit ahead of Anchor) is sent to each watch separately. type settledEpochs struct { commitEpoch utils.AtomicSend[*atypes.Epoch] anchor utils.AtomicSend[utils.Option[data.Anchor]] @@ -276,6 +280,172 @@ func (e *settledEpochs) store(ep *atypes.Epoch) { e.anchor.Store(testAnchor(ep)) } +func TestGigaRouterCommon_ValidatorAddrPrefersConfiguredBook(t *testing.T) { + rng := utils.TestRng() + validator := atypes.GenSecretKey(rng).Public() + configured := GigaNodeAddr{ + Key: makeKey(rng).Public(), + HostPort: tcp.HostPort{Hostname: "configured.example", Port: 26656}, + EVMRPC: *utils.OrPanic1(url.Parse("http://configured.example:8545")), + } + router := &gigaRouterCommon{ + cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{validator: configured}}, + liveAddrs: utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{validator: { + Key: makeKey(rng).Public(), + HostPort: tcp.HostPort{Hostname: "advertised.example", Port: 26656}, + EVMRPC: *utils.OrPanic1(url.Parse("http://advertised.example:8545")), + }}), + } + got, ok := router.validatorAddr(validator) + require.True(t, ok) + require.NoError(t, utils.TestDiff(configured, got)) +} + +func TestGigaRouterCommon_AcceptInboundRefusesUnroutableEvmRPC(t *testing.T) { + rng := utils.TestRng() + validatorKey := atypes.GenSecretKey(rng) + peerKey := makeKey(rng) + epochs := newSettledEpochs(testEpoch(1, map[atypes.PublicKey]uint64{validatorKey.Public(): 1})) + router := &gigaRouterCommon{ + cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{}}, + nextCommitEpoch: epochs.commitEpoch.Subscribe(), + liveAddrs: utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{}), + liveAddrVersion: utils.NewAtomicSend(uint64(0)), + } + for _, evmRPC := range []string{ + "http://127.0.0.1:8545", + "http://localhost:8545", + "http://169.254.169.254/", + "http://0.0.0.0:8545", + "http://[::]:8545", + } { + // The claim proves committee identity, so the peer is still served as a + // validator; only its advertised address is refused. + accepted := router.acceptInbound(&handshakedConn{msg: &handshakeMsg{ + NodeAuth: NodeChallengeSig{key: peerKey.Public()}, + handshakeSpec: handshakeSpec{ + SelfAddr: utils.Some(NodeAddress{ + NodeID: peerKey.Public().NodeID(), + Hostname: "validator.example", + Port: 26656, + }), + SeiGigaConnection: true, + }, + GigaClaim: utils.Some(gigaHandshakeClaim{Validator: validatorKey.Public(), evmRPC: evmRPC}), + }}) + require.True(t, accepted.IsPresent()) + _, ok := router.validatorAddr(validatorKey.Public()) + require.False(t, ok) + } +} + +func TestGigaRouterCommon_RunInboundConnLearnsMemberAndClosesOnLeave(t *testing.T) { + rng := utils.TestRng() + validatorKey := atypes.GenSecretKey(rng) + localKey := makeKey(rng) + peerKey := makeKey(rng) + selfAddr := NodeAddress{NodeID: peerKey.Public().NodeID(), Hostname: "validator.example", Port: 26656} + evmRPC := *utils.OrPanic1(url.Parse("http://validator.example:8545")) + dummy := atypes.GenSecretKey(rng) + epochs := newSettledEpochs(testEpoch(1, map[atypes.PublicKey]uint64{validatorKey.Public(): 1})) + genesis := map[atypes.PublicKey]GigaNodeAddr{dummy.Public(): {Key: makeKey(rng).Public()}} + router := testGigaRouterWithData(t, genesis) + router.nextCommitEpoch = epochs.commitEpoch.Subscribe() + router.anchor = epochs.anchor.Subscribe() + router.liveAddrs = utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{}) + router.liveAddrVersion = utils.NewAtomicSend(uint64(0)) + router.poolIn = giga.NewPool[NodePublicKey, rpc.Server[giga.API]]() + router.poolInCommittee = giga.NewPool[atypes.PublicKey, rpc.Server[giga.API]]() + router.inboundFullnodeCap = 10 + router.service = giga.NewFullNodeService(router.data) + router.key = localKey + router.cfg = &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{}} + gigaAddr := NodeAddress{NodeID: localKey.Public().NodeID(), Hostname: "giga.example", Port: 26656} + router.selfAddr = utils.Some(gigaAddr) + localSpec := handshakeSpec{SelfAddr: router.selfAddr, SeiGigaConnection: true} + want := GigaNodeAddr{ + Key: peerKey.Public(), + HostPort: tcp.HostPort{Hostname: selfAddr.Hostname, Port: selfAddr.Port}, + EVMRPC: evmRPC, + } + joinerKey := makeKey(rng) + require.False(t, router.acceptInbound(&handshakedConn{msg: &handshakeMsg{ + NodeAuth: NodeChallengeSig{key: joinerKey.Public()}, + handshakeSpec: handshakeSpec{ + SelfAddr: utils.Some(NodeAddress{ + NodeID: joinerKey.Public().NodeID(), + Hostname: "joiner.example", + Port: 26656, + }), + SeiGigaConnection: true, + }, + GigaClaim: utils.Some(gigaHandshakeClaim{ + Validator: dummy.Public(), + evmRPC: "http://joiner.example:8545", + }), + }}).IsPresent()) + _, ok := router.validatorAddr(dummy.Public()) + require.False(t, ok) + + require.NoError(t, scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error { + pair, err := handshakePair(ctx, s, + [2]NodeSecretKey{localKey, peerKey}, + [2]handshakeSpec{ + localSpec, + {SelfAddr: utils.Some(selfAddr), SeiGigaConnection: true}, + }, + [2]utils.Option[handshakeOffer]{ + router.offer, + utils.Some(handshakeOffer{ValidatorKey: validatorKey, EVMRPC: evmRPC}), + }, + ) + if err != nil { + return err + } + if err := utils.TestDiff(gigaAddr, pair[1].msg.SelfAddr.OrPanic("missing SelfAddr")); err != nil { + return err + } + addrUpdates := router.liveAddrVersion.Subscribe() + finished := scope.Spawn1(s, func() (struct{}, error) { + err := router.RunInboundConn(ctx, pair[0]) + if errors.Is(err, errGigaMembershipChanged) { + return struct{}{}, nil + } + if err != nil { + return struct{}{}, err + } + return struct{}{}, fmt.Errorf("RunInboundConn() = nil, want %v", errGigaMembershipChanged) + }) + if _, err := addrUpdates.Wait(ctx, func(v uint64) bool { return v > 0 }); err != nil { + return err + } + got, ok := router.validatorAddr(validatorKey.Public()) + if !ok { + return fmt.Errorf("validatorAddr(%v) missing after inbound", validatorKey.Public()) + } + if err := utils.TestDiff(want, got); err != nil { + return err + } + epochs.store(testEpoch(2, map[atypes.PublicKey]uint64{dummy.Public(): 1})) + _, err = finished.Join(ctx) + if err != nil { + return err + } + if err := router.stopStaleSessions( + ctx, + map[atypes.PublicKey]*memberSession{}, + router.anchor.Load(), + router.nextCommitEpoch.Load(), + ); err != nil { + return err + } + if _, ok := router.validatorAddr(validatorKey.Public()); ok { + return fmt.Errorf("validatorAddr(%v) kept after leave", validatorKey.Public()) + } + return nil + })) +} + func TestGigaRouterCommon_RunPerCommitteeMemberFollowsCommittee(t *testing.T) { rng := utils.TestRng() a := atypes.GenSecretKey(rng).Public() @@ -288,6 +458,8 @@ func TestGigaRouterCommon_RunPerCommitteeMemberFollowsCommittee(t *testing.T) { }}, nextCommitEpoch: epochs.commitEpoch.Subscribe(), anchor: epochs.anchor.Subscribe(), + liveAddrs: utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{}), + liveAddrVersion: utils.NewAtomicSend(uint64(0)), } startedA := make(chan struct{}, 1) @@ -358,6 +530,8 @@ func TestGigaRouterCommon_RunPerCommitteeMemberKeepsLeaversWhileAnchorLags(t *te }}, nextCommitEpoch: nextEpoch.Subscribe(), anchor: anchor.Subscribe(), + liveAddrs: utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{}), + liveAddrVersion: utils.NewAtomicSend(uint64(0)), } started := make(chan atypes.PublicKey, 3) @@ -411,6 +585,8 @@ func TestGigaRouterCommon_RunPerCommitteeMemberDialsBothCommitteesUntilStable(t }}, nextCommitEpoch: nextEpoch.Subscribe(), anchor: anchor.Subscribe(), + liveAddrs: utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{}), + liveAddrVersion: utils.NewAtomicSend(uint64(0)), } started := make(chan atypes.PublicKey, 4) @@ -457,6 +633,8 @@ func TestGigaRouterCommon_RunPerCommitteeMemberRunsOneSessionPerMember(t *testin }}, nextCommitEpoch: epochs.commitEpoch.Subscribe(), anchor: epochs.anchor.Subscribe(), + liveAddrs: utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{}), + liveAddrVersion: utils.NewAtomicSend(uint64(0)), } started := make(chan struct{}, 2) @@ -502,6 +680,8 @@ func TestGigaRouterCommon_RunPerCommitteeMemberCancelsAllDepartingBeforeAwait(t }}, nextCommitEpoch: epochs.commitEpoch.Subscribe(), anchor: epochs.anchor.Subscribe(), + liveAddrs: utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{}), + liveAddrVersion: utils.NewAtomicSend(uint64(0)), } started := make(chan atypes.PublicKey, 2) @@ -536,7 +716,7 @@ func TestGigaRouterCommon_CommitteeTasksReturnWhenWorkReturns(t *testing.T) { nextEpoch := utils.NewAtomicSend(testEpoch(2, map[atypes.PublicKey]uint64{a: 1})) router := &gigaRouterCommon{nextCommitEpoch: nextEpoch.Subscribe()} - changed, err := router.runUntilMembershipChange(t.Context(), a, true, func(context.Context) error { + changed, err := router.runUntilMembershipChange(t.Context(), a, func(context.Context) error { return nil }) require.NoError(t, err) @@ -547,47 +727,29 @@ func TestGigaRouterCommon_RunUntilMembershipChangeCancelsFWhenMembershipChanges( rng := utils.TestRng() a := atypes.GenSecretKey(rng).Public() b := atypes.GenSecretKey(rng).Public() - for _, tc := range []struct { - name string - isCommittee bool - }{ - {"leaves", true}, - {"joins", false}, - } { - t.Run(tc.name, func(t *testing.T) { - members := map[atypes.PublicKey]uint64{b: 1} - if tc.isCommittee { - members[a] = 1 - } - nextEpoch := utils.NewAtomicSend(testEpoch(2, members)) - router := &gigaRouterCommon{nextCommitEpoch: nextEpoch.Subscribe()} - started := make(chan struct{}) - err := scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error { - s.SpawnBg(func() error { - changed, err := router.runUntilMembershipChange(ctx, a, tc.isCommittee, func(ctx context.Context) error { - close(started) - <-ctx.Done() - return ctx.Err() - }) - if err != nil { - return err - } - if !changed { - return fmt.Errorf("membership change must cancel f and report true") - } - return nil - }) - <-started - flipped := map[atypes.PublicKey]uint64{b: 1} - if !tc.isCommittee { - flipped[a] = 1 - } - nextEpoch.Store(testEpoch(3, flipped)) - return nil + nextEpoch := utils.NewAtomicSend(testEpoch(2, map[atypes.PublicKey]uint64{a: 1, b: 1})) + router := &gigaRouterCommon{nextCommitEpoch: nextEpoch.Subscribe()} + started := make(chan struct{}) + err := scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error { + s.SpawnBg(func() error { + changed, err := router.runUntilMembershipChange(ctx, a, func(ctx context.Context) error { + close(started) + <-ctx.Done() + return ctx.Err() }) - require.NoError(t, err) + if err != nil { + return err + } + if !changed { + return fmt.Errorf("membership change must cancel f and report true") + } + return nil }) - } + <-started + nextEpoch.Store(testEpoch(3, map[atypes.PublicKey]uint64{b: 1})) + return nil + }) + require.NoError(t, err) } func TestCommitteeWeights(t *testing.T) { diff --git a/sei-tendermint/internal/p2p/giga_router_fullnode.go b/sei-tendermint/internal/p2p/giga_router_fullnode.go index 23d0407b47..e84c0849ba 100644 --- a/sei-tendermint/internal/p2p/giga_router_fullnode.go +++ b/sei-tendermint/internal/p2p/giga_router_fullnode.go @@ -32,9 +32,12 @@ func NewGigaFullnodeRouter(cfg *GigaRouterCommonConfig, key NodeSecretKey, dataS anchor: dataState.Anchor(), service: giga.NewFullNodeService(dataState), poolIn: giga.NewPool[NodePublicKey, rpc.Server[giga.API]](), - poolOut: giga.NewPool[NodePublicKey, rpc.Client[giga.API]](), + poolInCommittee: giga.NewPool[atypes.PublicKey, rpc.Server[giga.API]](), + poolOut: giga.NewPool[atypes.PublicKey, rpc.Client[giga.API]](), proxies: utils.NewRWMutex(map[atypes.PublicKey]*ethrpc.Client{}), app: cfg.App, + liveAddrs: utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{}), + liveAddrVersion: utils.NewAtomicSend(uint64(0)), inboundFullnodeCap: int64(cfg.MaxInboundFullnodePeers), }, }, nil @@ -107,8 +110,8 @@ func (r *gigaFullnodeRouter) runFullnodeSubscriber(ctx context.Context) error { break } addr := r.cfg.ValidatorAddrs[validator] - left, err := r.runUntilMembershipChange(ctx, validator, true, func(ctx context.Context) error { - return r.dialAndRunConn(ctx, utils.Some(addr.Key), addr.HostPort, func(ctx context.Context, client rpc.Client[giga.API]) error { + left, err := r.runUntilMembershipChange(ctx, validator, func(ctx context.Context) error { + return r.dialAndRunConn(ctx, validator, addr.Key, addr.HostPort, func(ctx context.Context, client rpc.Client[giga.API]) error { // Consensus PublicKey (committee member), not GigaNodeAddr.Key (p2p NodePublicKey). return r.service.RunClient(ctx, client, validator, true) }) diff --git a/sei-tendermint/internal/p2p/giga_router_fullnode_test.go b/sei-tendermint/internal/p2p/giga_router_fullnode_test.go index 9af0a777a2..22da121fee 100644 --- a/sei-tendermint/internal/p2p/giga_router_fullnode_test.go +++ b/sei-tendermint/internal/p2p/giga_router_fullnode_test.go @@ -33,14 +33,13 @@ func TestGigaRouter_Fullnode(t *testing.T) { rng := utils.TestRng() _, validatorKeys := atypes.GenCommittee(rng, 5) addrs := map[atypes.PublicKey]GigaNodeAddr{} - urlByValidator := map[atypes.PublicKey]*url.URL{} + urlByValidator := map[atypes.PublicKey]url.URL{} for i, validatorKey := range validatorKeys { nodeKey := makeKey(rng) // Every committee member needs an EVMRPC URL for fullnode mode — // NewGigaRouter enforces this at construction so a missing URL // can't lead to silently-dropped txs. - rpcURL, err := url.Parse(fmt.Sprintf("http://validator-%d.example.com:8545", i)) - require.NoError(t, err) + rpcURL := *utils.OrPanic1(url.Parse(fmt.Sprintf("http://validator-%d.example.com:8545", i))) addrs[validatorKey.Public()] = GigaNodeAddr{ Key: nodeKey.Public(), HostPort: tcp.HostPort{Hostname: "127.0.0.1", Port: 26657}, diff --git a/sei-tendermint/internal/p2p/giga_router_testhelper_test.go b/sei-tendermint/internal/p2p/giga_router_testhelper_test.go index 07715f534c..b5bc6e3533 100644 --- a/sei-tendermint/internal/p2p/giga_router_testhelper_test.go +++ b/sei-tendermint/internal/p2p/giga_router_testhelper_test.go @@ -200,7 +200,7 @@ func (c *testNodeCfg) GigaNodeAddr() GigaNodeAddr { return GigaNodeAddr{ Key: c.nodeKey.Public(), HostPort: tcp.HostPort{Hostname: c.addr.Addr().String(), Port: c.addr.Port()}, - EVMRPC: utils.OrPanic1(url.Parse(fmt.Sprintf("http://%s:8545", c.addr.Addr().String()))), + EVMRPC: *utils.OrPanic1(url.Parse(fmt.Sprintf("http://%s:8545", c.addr.Addr().String()))), } } diff --git a/sei-tendermint/internal/p2p/giga_router_validator.go b/sei-tendermint/internal/p2p/giga_router_validator.go index b99cc84bf0..1abd66ddb8 100644 --- a/sei-tendermint/internal/p2p/giga_router_validator.go +++ b/sei-tendermint/internal/p2p/giga_router_validator.go @@ -30,6 +30,26 @@ type gigaValidatorRouter struct { // data.State. The caller owns the BlockDB that backs dataState (see BuildDataState); // close it if this constructor returns an error. func NewGigaValidatorRouter(cfg *GigaValidatorConfig, key NodeSecretKey, dataState *data.State) (*gigaValidatorRouter, error) { + validatorKey := cfg.ValidatorKey.Public() + self, ok := cfg.ValidatorAddrs[validatorKey] + if !ok { + return nil, fmt.Errorf("local validator %v has no configured giga address", validatorKey) + } + if self.Key != key.Public() { + return nil, fmt.Errorf("local validator node key = %v, want %v", self.Key, key.Public()) + } + if err := utils.CheckHTTPURL(self.EVMRPC); err != nil { + return nil, fmt.Errorf("local validator %v evmrpc: %w", validatorKey, err) + } + selfAddr := NodeAddress{ + NodeID: key.Public().NodeID(), + Hostname: self.HostPort.Hostname, + Port: self.HostPort.Port, + } + // An invalid local address makes every peer reject our giga claim. + if err := selfAddr.Validate(); err != nil { + return nil, fmt.Errorf("local validator %v address: %w", validatorKey, err) + } consensusState, err := consensus.NewState(&consensus.Config{ Key: cfg.ValidatorKey, ViewTimeout: cfg.ViewTimeout, @@ -42,21 +62,29 @@ func NewGigaValidatorRouter(cfg *GigaValidatorConfig, key NodeSecretKey, dataSta logger.Info("GigaRouter initialized (validator)", "validators", len(cfg.ValidatorAddrs), "dial_interval", cfg.DialInterval, "inbound_fullnode_cap", cfg.MaxInboundFullnodePeers) return &gigaValidatorRouter{ gigaRouterCommon: &gigaRouterCommon{ - cfg: &cfg.GigaRouterCommonConfig, - key: key, - data: dataState, - nextCommitEpoch: dataState.NextCommitEpoch(), - anchor: dataState.Anchor(), - service: giga.NewService(consensusState), - poolIn: giga.NewPool[NodePublicKey, rpc.Server[giga.API]](), - poolOut: giga.NewPool[NodePublicKey, rpc.Client[giga.API]](), - proxies: utils.NewRWMutex(map[atypes.PublicKey]*ethrpc.Client{}), - app: cfg.App, + cfg: &cfg.GigaRouterCommonConfig, + key: key, + data: dataState, + nextCommitEpoch: dataState.NextCommitEpoch(), + anchor: dataState.Anchor(), + service: giga.NewService(consensusState), + poolIn: giga.NewPool[NodePublicKey, rpc.Server[giga.API]](), + poolInCommittee: giga.NewPool[atypes.PublicKey, rpc.Server[giga.API]](), + poolOut: giga.NewPool[atypes.PublicKey, rpc.Client[giga.API]](), + proxies: utils.NewRWMutex(map[atypes.PublicKey]*ethrpc.Client{}), + app: cfg.App, + offer: utils.Some(handshakeOffer{ + ValidatorKey: cfg.ValidatorKey, + EVMRPC: self.EVMRPC, + }), + selfAddr: utils.Some(selfAddr), + liveAddrs: utils.NewRWMutex(map[atypes.PublicKey]GigaNodeAddr{}), + liveAddrVersion: utils.NewAtomicSend(uint64(0)), inboundFullnodeCap: int64(cfg.MaxInboundFullnodePeers), }, consensus: consensusState, producer: producerState, - validatorKey: cfg.ValidatorKey.Public(), + validatorKey: validatorKey, }, nil } @@ -87,7 +115,7 @@ func (r *gigaValidatorRouter) Run(ctx context.Context) error { func (r *gigaValidatorRouter) runCommitteePeer(ctx context.Context, validatorKey atypes.PublicKey, addr GigaNodeAddr) error { getBlock := addr.Key != r.key.Public() for { - err := r.dialAndRunConn(ctx, utils.Some(addr.Key), addr.HostPort, func(ctx context.Context, client rpc.Client[giga.API]) error { + err := r.dialAndRunConn(ctx, validatorKey, addr.Key, addr.HostPort, func(ctx context.Context, client rpc.Client[giga.API]) error { return r.service.RunClient(ctx, client, validatorKey, getBlock) }) logger.Info("giga connection failed", "addr", addr, "err", err) @@ -109,11 +137,7 @@ func (r *gigaValidatorRouter) EvmProxy(sender common.Address) utils.Option[*ethr if r.validatorKey == validator { return utils.None[*ethrpc.Client]() } - target, ok := r.cfg.ValidatorAddrs[validator] - if !ok { - return utils.None[*ethrpc.Client]() - } - if _, ok := r.poolOut.Get(target.Key); !ok { + if _, ok := r.poolOut.Get(validator); !ok { return utils.None[*ethrpc.Client]() } return r.evmProxy(validator) diff --git a/sei-tendermint/internal/p2p/giga_router_validator_test.go b/sei-tendermint/internal/p2p/giga_router_validator_test.go index 4a5e5f04e3..94ec3e790c 100644 --- a/sei-tendermint/internal/p2p/giga_router_validator_test.go +++ b/sei-tendermint/internal/p2p/giga_router_validator_test.go @@ -219,14 +219,13 @@ func TestGigaRouter_EvmProxy(t *testing.T) { _, validatorKeys := atypes.GenCommittee(rng, 10) var nodeKeys []NodeSecretKey addrs := map[atypes.PublicKey]GigaNodeAddr{} - urlByValidator := map[atypes.PublicKey]*url.URL{} + urlByValidator := map[atypes.PublicKey]url.URL{} // NewGigaRouter requires EVMRPC on every committee member in both // validator and fullnode modes. for i, validatorKey := range validatorKeys { nodeKey := makeKey(rng) nodeKeys = append(nodeKeys, nodeKey) - rpcURL, err := url.Parse(fmt.Sprintf("http://validator-%d.example.com:8545", i)) - require.NoError(t, err) + rpcURL := *utils.OrPanic1(url.Parse(fmt.Sprintf("http://validator-%d.example.com:8545", i))) addrs[validatorKey.Public()] = GigaNodeAddr{ Key: nodeKey.Public(), HostPort: tcp.HostPort{Hostname: "127.0.0.1", Port: 26657}, @@ -295,11 +294,10 @@ func TestGigaRouter_EvmProxy(t *testing.T) { err = scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error { for validator := range connectedRemote { - key := addrs[validator].Key ready := make(chan struct{}) - s.SpawnBgNamed(fmt.Sprintf("poolOut[%s]", key), func() error { + s.SpawnBgNamed(fmt.Sprintf("poolOut[%s]", validator), func() error { var client rpc.Client[giga.API] - return utils.IgnoreCancel(router.poolOut.InsertAndRun(ctx, key, client, func(ctx context.Context) error { + return utils.IgnoreCancel(router.poolOut.InsertAndRun(ctx, validator, client, func(ctx context.Context) error { close(ready) <-ctx.Done() return ctx.Err() @@ -340,3 +338,36 @@ func TestGigaRouter_EvmProxy(t *testing.T) { }) require.NoError(t, err) } + +func TestGigaRouter_RejectsInvalidSelfAddr(t *testing.T) { + rng := utils.TestRng() + _, validatorKeys := atypes.GenCommittee(rng, 1) + self := &testNodeCfg{ + validatorKey: validatorKeys[0], + nodeKey: makeKey(rng), + addr: tcp.TestReserveAddr(), + } + for _, tc := range []struct { + name string + hostPort tcp.HostPort + want string + }{ + {"zero port", tcp.HostPort{Hostname: "validator1.example.com"}, "missing port"}, + {"empty hostname", tcp.HostPort{Port: 26656}, "missing hostname"}, + } { + t.Run(tc.name, func(t *testing.T) { + addr := self.GigaNodeAddr() + addr.HostPort = tc.hostPort + _, err := NewGigaValidatorRouter(&GigaValidatorConfig{ + GigaRouterCommonConfig: GigaRouterCommonConfig{ + ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{ + self.validatorKey.Public(): addr, + }, + }, + ValidatorKey: self.validatorKey, + }, self.nodeKey, nil) + require.Error(t, err) + require.Contains(t, err.Error(), tc.want) + }) + } +} diff --git a/sei-tendermint/internal/p2p/handshake.go b/sei-tendermint/internal/p2p/handshake.go index cef3f19790..0e010111ae 100644 --- a/sei-tendermint/internal/p2p/handshake.go +++ b/sei-tendermint/internal/p2p/handshake.go @@ -2,23 +2,80 @@ package p2p import ( "context" + "encoding/binary" + "errors" "fmt" + "net/url" + "slices" gogoproto "github.com/gogo/protobuf/proto" + atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/crypto/ed25519" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/conn" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/pb" gogopb "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/p2p" "github.com/sei-protocol/sei-chain/sei-tendermint/types" ) +var ( + gigaValidatorHandshakeTag = utils.OrPanic1(ed25519.NewTag("SEI_GIGA_VALIDATOR_HANDSHAKE_V1")) + errMissingGigaClaim = errors.New("missing validator_auth_key, validator_auth_sig, and evm_rpc") + errGigaClaimRequiresAddr = errors.New("validator giga handshake requires SelfAddr") + errGigaClaimOnNonGiga = errors.New("validator giga claim on non-giga connection") +) + +// handshakeOffer is the Autobahn committee identity this node claims on a giga handshake. +type handshakeOffer struct { + ValidatorKey atypes.SecretKey + EVMRPC url.URL +} + +// gigaClaimSignBytes is the tagged payload for SEI_GIGA_VALIDATOR_HANDSHAKE_V1: +// challenge || node_public_key || uvarint(len(self_addr)) || self_addr || uvarint(len(evm_rpc)) || evm_rpc. +func gigaClaimSignBytes( + challenge conn.Challenge, + nodeKey NodePublicKey, + selfAddr NodeAddress, + evmRPC string, +) []byte { + b := slices.Clone(challenge[:]) + b = append(b, nodeKey.Bytes()...) + // Length-prefixed: without it, bytes could move across the address/URL + // boundary and two different claims would sign the same payload. + for _, s := range []string{selfAddr.String(), evmRPC} { + b = binary.AppendUvarint(b, uint64(len(s))) + b = append(b, s...) + } + return b +} + type handshakedConn struct { conn *conn.SecretConnection msg *handshakeMsg } -func handshake(ctx context.Context, c conn.Conn, key NodeSecretKey, spec handshakeSpec) (*handshakedConn, error) { +func handshake( + ctx context.Context, + c conn.Conn, + key NodeSecretKey, + spec handshakeSpec, + offer utils.Option[handshakeOffer], +) (*handshakedConn, error) { + // Checked before the connection so a local misconfiguration is reported as + // itself, rather than as whichever handshake step the abort trips first. + o, offering := offer.Get() + selfAddr, hasSelfAddr := spec.SelfAddr.Get() + if offering && !hasSelfAddr { + return nil, errGigaClaimRequiresAddr + } + if offering { + if err := utils.CheckHTTPURL(o.EVMRPC); err != nil { + return nil, fmt.Errorf("EvmRpc: %w", err) + } + } return scope.Run1(ctx, func(ctx context.Context, s scope.Scope) (*handshakedConn, error) { sc, err := conn.MakeSecretConnection(ctx, c) if err != nil { @@ -29,6 +86,16 @@ func handshake(ctx context.Context, c conn.Conn, key NodeSecretKey, spec handsha NodeAuth: key.SignChallenge(sc.Challenge()), handshakeSpec: spec, } + if offering { + msg.GigaClaim = utils.Some(gigaHandshakeClaim{ + Validator: o.ValidatorKey.Public(), + sig: o.ValidatorKey.SignWithTag( + gigaValidatorHandshakeTag, + gigaClaimSignBytes(sc.Challenge(), key.Public(), selfAddr, o.EVMRPC.String()), + ), + evmRPC: o.EVMRPC.String(), + }) + } if err := conn.WriteSizedMsg(ctx, sc, handshakeMsgConv.Marshal(msg)); err != nil { return fmt.Errorf("conn.WriteSizedMsg(): %w", err) } @@ -37,7 +104,7 @@ func handshake(ctx context.Context, c conn.Conn, key NodeSecretKey, spec handsha } return nil }) - msgBytes, err := conn.ReadSizedMsg(ctx, sc, 1024*1024) + msgBytes, err := conn.ReadSizedMsg(ctx, sc, uint64((&pb.Handshake{}).MaxSize())) //nolint:gosec // MaxSize() returns a small positive constant if err != nil { return nil, fmt.Errorf("conn.ReadSizedMsg(): %w", err) } @@ -53,6 +120,22 @@ func handshake(ctx context.Context, c conn.Conn, key NodeSecretKey, spec handsha return nil, fmt.Errorf("handshakeMsg.SelfAddr.NodeID = %v, want %v", got, want) } } + if claim, ok := msg.GigaClaim.Get(); ok { + if !msg.SeiGigaConnection { + return nil, errGigaClaimOnNonGiga + } + selfAddr, ok := msg.SelfAddr.Get() + if !ok { + return nil, errGigaClaimRequiresAddr + } + if err := claim.Validator.VerifyWithTag( + gigaValidatorHandshakeTag, + gigaClaimSignBytes(sc.Challenge(), msg.NodeAuth.Key(), selfAddr, claim.evmRPC), + claim.sig, + ); err != nil { + return nil, fmt.Errorf("handshakeMsg.GigaClaim: %w", err) + } + } if len(msg.PexAddrs) > MaxPexAddrs { return nil, fmt.Errorf("len(handshakeMsg.PexAddrs) = %v, want <= %v", len(msg.PexAddrs), MaxPexAddrs) } diff --git a/sei-tendermint/internal/p2p/handshake_deadline_test.go b/sei-tendermint/internal/p2p/handshake_deadline_test.go index 6dbdacc74e..006431fbfd 100644 --- a/sei-tendermint/internal/p2p/handshake_deadline_test.go +++ b/sei-tendermint/internal/p2p/handshake_deadline_test.go @@ -59,7 +59,7 @@ func TestRouter_InboundNodeInfoBoundedByHandshakeDeadline(t *testing.T) { s.SpawnBg(func() error { return tcpConn.Run(ctx) }) // Complete the handshake, which authenticates us, then send no node info. - if _, err := handshake(ctx, tcpConn, NodeSecretKey(ed25519.GenerateSecretKey()), handshakeSpec{}); err != nil { + if _, err := handshake(ctx, tcpConn, NodeSecretKey(ed25519.GenerateSecretKey()), handshakeSpec{}, utils.None[handshakeOffer]()); err != nil { return err } diff --git a/sei-tendermint/internal/p2p/handshake_test.go b/sei-tendermint/internal/p2p/handshake_test.go new file mode 100644 index 0000000000..8ac1fccc0e --- /dev/null +++ b/sei-tendermint/internal/p2p/handshake_test.go @@ -0,0 +1,256 @@ +package p2p + +import ( + "context" + "fmt" + "net/url" + "testing" + + atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/crypto/ed25519" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/conn" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/pb" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/tcp" +) + +func signGigaClaim( + key atypes.SecretKey, + challenge conn.Challenge, + nodeKey NodePublicKey, + selfAddr NodeAddress, + evmRPC url.URL, +) gigaHandshakeClaim { + return gigaHandshakeClaim{ + Validator: key.Public(), + sig: key.SignWithTag( + gigaValidatorHandshakeTag, + gigaClaimSignBytes(challenge, nodeKey, selfAddr, evmRPC.String()), + ), + evmRPC: evmRPC.String(), + } +} + +// handshakePair runs both ends of a handshake over a pipe held open by s, and +// returns what each end saw: got[i] is the message handshake i received from +// its peer. +func handshakePair( + ctx context.Context, + s scope.Scope, + keys [2]NodeSecretKey, + specs [2]handshakeSpec, + offers [2]utils.Option[handshakeOffer], +) ([2]*handshakedConn, error) { + conns := [2]tcp.Conn{} + conns[0], conns[1] = tcp.TestPipe() + var handles [2]scope.JoinHandle[*handshakedConn] + for i, c := range conns { + // The pipe outlives the handshake, and reports EOF once whichever end + // the test is exercising closes. A handshake failure surfaces below. + s.SpawnBg(func() error { _ = c.Run(ctx); return nil }) + handles[i] = scope.Spawn1(s, func() (*handshakedConn, error) { + return handshake(ctx, c, keys[i], specs[i], offers[i]) + }) + } + var got [2]*handshakedConn + for i, h := range handles { + hConn, err := h.Join(ctx) + if err != nil { + return got, fmt.Errorf("handshake[%v]: %w", i, err) + } + got[i] = hConn + } + return got, nil +} + +// handshakePairIn runs handshakePair in its own scope, for tests that only +// inspect the messages and do not use the connections. +func handshakePairIn( + t *testing.T, + keys [2]NodeSecretKey, + specs [2]handshakeSpec, + offers [2]utils.Option[handshakeOffer], +) [2]*handshakedConn { + t.Helper() + var got [2]*handshakedConn + require.NoError(t, scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error { + var err error + got, err = handshakePair(ctx, s, keys, specs, offers) + return err + })) + return got +} + +func TestHandshakeAuthenticatesGigaClaim(t *testing.T) { + rng := utils.TestRng() + nodeKeys := [2]NodeSecretKey{makeKey(rng), makeKey(rng)} + validatorKeys := [2]atypes.SecretKey{atypes.GenSecretKey(rng), atypes.GenSecretKey(rng)} + selfAddrs := [2]NodeAddress{ + {NodeID: nodeKeys[0].Public().NodeID(), Hostname: "validator-a.example", Port: 26656}, + {NodeID: nodeKeys[1].Public().NodeID(), Hostname: "validator-b.example", Port: 26656}, + } + evmRPCs := [2]url.URL{ + *utils.OrPanic1(url.Parse("http://validator-a.example:8545")), + *utils.OrPanic1(url.Parse("http://validator-b.example:8545")), + } + got := handshakePairIn(t, nodeKeys, [2]handshakeSpec{ + {SelfAddr: utils.Some(selfAddrs[0]), SeiGigaConnection: true}, + {SelfAddr: utils.Some(selfAddrs[1]), SeiGigaConnection: true}, + }, [2]utils.Option[handshakeOffer]{ + utils.Some(handshakeOffer{ValidatorKey: validatorKeys[0], EVMRPC: evmRPCs[0]}), + utils.Some(handshakeOffer{ValidatorKey: validatorKeys[1], EVMRPC: evmRPCs[1]}), + }) + for i := range got { + peer := 1 - i + claim := got[i].msg.GigaClaim.OrPanic("missing peer giga claim") + require.Equal(t, validatorKeys[peer].Public(), claim.Validator) + require.Equal(t, selfAddrs[peer], got[i].msg.SelfAddr.OrPanic("missing peer SelfAddr")) + require.Equal(t, evmRPCs[peer].String(), claim.evmRPC) + } +} + +func TestHandshakeAcceptsFullnodeWithoutClaim(t *testing.T) { + rng := utils.TestRng() + validatorNode := makeKey(rng) + fullnodeNode := makeKey(rng) + validatorKey := atypes.GenSecretKey(rng) + selfAddr := NodeAddress{ + NodeID: validatorNode.Public().NodeID(), + Hostname: "validator.example", + Port: 26656, + } + evmRPC := *utils.OrPanic1(url.Parse("http://validator.example:8545")) + got := handshakePairIn(t, + [2]NodeSecretKey{validatorNode, fullnodeNode}, + [2]handshakeSpec{ + {SelfAddr: utils.Some(selfAddr), SeiGigaConnection: true}, + {SeiGigaConnection: true}, + }, + [2]utils.Option[handshakeOffer]{ + utils.Some(handshakeOffer{ValidatorKey: validatorKey, EVMRPC: evmRPC}), + utils.None[handshakeOffer](), + }, + ) + require.False(t, got[0].msg.GigaClaim.IsPresent()) + require.Equal(t, fullnodeNode.Public(), got[0].msg.NodeAuth.Key()) + require.Equal(t, validatorKey.Public(), got[1].msg.GigaClaim.OrPanic("validator omitted giga claim").Validator) +} + +func TestHandshakeRejectsClaimOnNonGiga(t *testing.T) { + rng := utils.TestRng() + nodeKey := makeKey(rng) + validatorKey := atypes.GenSecretKey(rng) + selfAddr := NodeAddress{NodeID: nodeKey.Public().NodeID(), Hostname: "validator.example", Port: 26656} + evmRPC := *utils.OrPanic1(url.Parse("http://validator.example:8545")) + err := handshakeAgainst(t, nodeKey, handshakeSpec{ + SelfAddr: utils.Some(selfAddr), + SeiGigaConnection: true, + }, utils.Some(handshakeOffer{ValidatorKey: validatorKey, EVMRPC: evmRPC}), + func(ctx context.Context, sc *conn.SecretConnection) error { + return writeHandshake(ctx, sc, nodeKey, handshakeSpec{ + SelfAddr: utils.Some(selfAddr), + SeiGigaConnection: false, + }, utils.Some(signGigaClaim(validatorKey, sc.Challenge(), nodeKey.Public(), selfAddr, evmRPC))) + }, + ) + require.ErrorIs(t, err, errGigaClaimOnNonGiga) +} + +func TestHandshakeRejectsBadGigaClaimSig(t *testing.T) { + rng := utils.TestRng() + nodeKey := makeKey(rng) + validatorKey := atypes.GenSecretKey(rng) + selfAddr := NodeAddress{NodeID: nodeKey.Public().NodeID(), Hostname: "validator.example", Port: 26656} + evmRPC := *utils.OrPanic1(url.Parse("http://validator.example:8545")) + err := handshakeAgainst(t, nodeKey, handshakeSpec{ + SelfAddr: utils.Some(selfAddr), + SeiGigaConnection: true, + }, utils.None[handshakeOffer](), + func(ctx context.Context, sc *conn.SecretConnection) error { + claim := signGigaClaim(validatorKey, sc.Challenge(), nodeKey.Public(), selfAddr, evmRPC) + sig := claim.sig.Bytes() + sig[0] ^= 1 + claim.sig = utils.OrPanic1(ed25519.SignatureFromBytes(sig)) + return writeHandshake(ctx, sc, nodeKey, handshakeSpec{ + SelfAddr: utils.Some(selfAddr), + SeiGigaConnection: true, + }, utils.Some(claim)) + }, + ) + require.Error(t, err) +} + +func TestHandshakeRejectsClaimWithoutSelfAddr(t *testing.T) { + rng := utils.TestRng() + nodeKey := makeKey(rng) + validatorKey := atypes.GenSecretKey(rng) + selfAddr := NodeAddress{NodeID: nodeKey.Public().NodeID(), Hostname: "validator.example", Port: 26656} + evmRPC := *utils.OrPanic1(url.Parse("http://validator.example:8545")) + received := handshakeAgainst(t, nodeKey, handshakeSpec{ + SelfAddr: utils.Some(selfAddr), + SeiGigaConnection: true, + }, utils.None[handshakeOffer](), + func(ctx context.Context, sc *conn.SecretConnection) error { + return writeHandshake(ctx, sc, nodeKey, handshakeSpec{ + SeiGigaConnection: true, + }, utils.Some(signGigaClaim(validatorKey, sc.Challenge(), nodeKey.Public(), selfAddr, evmRPC))) + }, + ) + require.ErrorIs(t, received, errGigaClaimRequiresAddr) + + offered := handshakeAgainst(t, nodeKey, handshakeSpec{SeiGigaConnection: true}, + utils.Some(handshakeOffer{ValidatorKey: validatorKey, EVMRPC: evmRPC}), + func(context.Context, *conn.SecretConnection) error { return nil }, + ) + require.ErrorIs(t, offered, errGigaClaimRequiresAddr) +} + +func handshakeAgainst( + t *testing.T, + localKey NodeSecretKey, + localSpec handshakeSpec, + localOffer utils.Option[handshakeOffer], + remote func(context.Context, *conn.SecretConnection) error, +) error { + t.Helper() + return scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error { + a, b := tcp.TestPipe() + s.SpawnBg(func() error { return utils.IgnoreCancel(a.Run(ctx)) }) + s.SpawnBg(func() error { return utils.IgnoreCancel(b.Run(ctx)) }) + s.Spawn(func() error { + sc, err := conn.MakeSecretConnection(ctx, b) + if err != nil { + return err + } + if err := remote(ctx, sc); err != nil { + return err + } + // Only the local handshake's verdict is asserted. This read keeps + // the pipe open until it reaches one, and sees EOF when the local + // side rejects and closes first. + _, _ = conn.ReadSizedMsg(ctx, sc, uint64((&pb.Handshake{}).MaxSize())) + return nil + }) + _, err := handshake(ctx, a, localKey, localSpec, localOffer) + return err + }) +} + +func writeHandshake( + ctx context.Context, + sc *conn.SecretConnection, + key NodeSecretKey, + spec handshakeSpec, + claim utils.Option[gigaHandshakeClaim], +) error { + msg := &handshakeMsg{NodeAuth: key.SignChallenge(sc.Challenge()), handshakeSpec: spec} + if c, ok := claim.Get(); ok { + msg.GigaClaim = utils.Some(c) + } + if err := conn.WriteSizedMsg(ctx, sc, handshakeMsgConv.Marshal(msg)); err != nil { + return err + } + return sc.Flush(ctx) +} diff --git a/sei-tendermint/internal/p2p/p2p.proto b/sei-tendermint/internal/p2p/p2p.proto index 1a5510a493..608f481b9b 100644 --- a/sei-tendermint/internal/p2p/p2p.proto +++ b/sei-tendermint/internal/p2p/p2p.proto @@ -2,6 +2,8 @@ syntax = "proto3"; package p2p; +import "wireguard/wireguard.proto"; + option go_package = "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/pb"; message PacketPing {} @@ -23,7 +25,8 @@ message Packet { } message NodePublicKey { - bytes ed25519 = 1; + option (wireguard.sized) = true; + bytes ed25519 = 1 [(wireguard.max_size) = 32]; reserved 2, 3; reserved "secp256k1", "sr25519"; } @@ -34,16 +37,27 @@ message Preface { } message Handshake { + option (wireguard.sized) = true; // Node signature on the encrypted session challenge // derived from the secret. It authenticates the node. NodePublicKey node_auth_key = 1; - bytes node_auth_sig = 2; + bytes node_auth_sig = 2 [(wireguard.max_size) = 64]; // NodeAddress that this peer can be dialed at. - optional string self_addr = 4; + // 320 fits the longest NodeAddress: "mconn://" + 40-hex node ID + "@" + a + // 253-char DNS name + ":" + a 5-digit port. + optional string self_addr = 4 [(wireguard.max_size) = 320]; // Initial peer exchange. // It allows to receive peer addresses from a node which doesn't accept any new inbound connections // (it is at full capacity). - repeated string pex_addrs = 5; + repeated string pex_addrs = 5 [ + (wireguard.max_count) = 100, + (wireguard.max_size) = 320 + ]; bool sei_giga_connection = 3; + + // Committee claim; all three or none. + optional bytes validator_auth_key = 6 [(wireguard.max_size) = 32]; + optional bytes validator_auth_sig = 7 [(wireguard.max_size) = 64]; + optional string evm_rpc = 8 [(wireguard.max_size) = 2048]; } diff --git a/sei-tendermint/internal/p2p/pb/p2p.pb.go b/sei-tendermint/internal/p2p/pb/p2p.pb.go index c241bed5ab..8f4a4f04cf 100644 --- a/sei-tendermint/internal/p2p/pb/p2p.pb.go +++ b/sei-tendermint/internal/p2p/pb/p2p.pb.go @@ -7,6 +7,7 @@ package pb import ( + _ "github.com/sei-protocol/sei-chain/sei-tendermint/proto/wireguard" protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" reflect "reflect" @@ -347,14 +348,20 @@ type Handshake struct { NodeAuthKey *NodePublicKey `protobuf:"bytes,1,opt,name=node_auth_key,json=nodeAuthKey,proto3" json:"node_auth_key,omitempty"` NodeAuthSig []byte `protobuf:"bytes,2,opt,name=node_auth_sig,json=nodeAuthSig,proto3" json:"node_auth_sig,omitempty"` // NodeAddress that this peer can be dialed at. + // 320 fits the longest NodeAddress: "mconn://" + 40-hex node ID + "@" + a + // 253-char DNS name + ":" + a 5-digit port. SelfAddr *string `protobuf:"bytes,4,opt,name=self_addr,json=selfAddr,proto3,oneof" json:"self_addr,omitempty"` // Initial peer exchange. // It allows to receive peer addresses from a node which doesn't accept any new inbound connections // (it is at full capacity). PexAddrs []string `protobuf:"bytes,5,rep,name=pex_addrs,json=pexAddrs,proto3" json:"pex_addrs,omitempty"` SeiGigaConnection bool `protobuf:"varint,3,opt,name=sei_giga_connection,json=seiGigaConnection,proto3" json:"sei_giga_connection,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Committee claim; all three or none. + ValidatorAuthKey []byte `protobuf:"bytes,6,opt,name=validator_auth_key,json=validatorAuthKey,proto3,oneof" json:"validator_auth_key,omitempty"` + ValidatorAuthSig []byte `protobuf:"bytes,7,opt,name=validator_auth_sig,json=validatorAuthSig,proto3,oneof" json:"validator_auth_sig,omitempty"` + EvmRpc *string `protobuf:"bytes,8,opt,name=evm_rpc,json=evmRpc,proto3,oneof" json:"evm_rpc,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Handshake) Reset() { @@ -422,11 +429,32 @@ func (x *Handshake) GetSeiGigaConnection() bool { return false } +func (x *Handshake) GetValidatorAuthKey() []byte { + if x != nil { + return x.ValidatorAuthKey + } + return nil +} + +func (x *Handshake) GetValidatorAuthSig() []byte { + if x != nil { + return x.ValidatorAuthSig + } + return nil +} + +func (x *Handshake) GetEvmRpc() string { + if x != nil && x.EvmRpc != nil { + return *x.EvmRpc + } + return "" +} + var File_p2p_p2p_proto protoreflect.FileDescriptor const file_p2p_p2p_proto_rawDesc = "" + "\n" + - "\rp2p/p2p.proto\x12\x03p2p\"\f\n" + + "\rp2p/p2p.proto\x12\x03p2p\x1a\x19wireguard/wireguard.proto\"\f\n" + "\n" + "PacketPing\"\f\n" + "\n" + @@ -443,19 +471,26 @@ const file_p2p_p2p_proto_rawDesc = "" + "packetPong\x12/\n" + "\n" + "packet_msg\x18\x03 \x01(\v2\x0e.p2p.PacketMsgH\x00R\tpacketMsgB\x05\n" + - "\x03sum\"I\n" + - "\rNodePublicKey\x12\x18\n" + - "\aed25519\x18\x01 \x01(\fR\aed25519J\x04\b\x02\x10\x03J\x04\b\x03\x10\x04R\tsecp256k1R\asr25519\"/\n" + + "\x03sum\"Y\n" + + "\rNodePublicKey\x12 \n" + + "\aed25519\x18\x01 \x01(\fB\x06؈\xe2\xab\f R\aed25519:\x06\xe8\x88\xe2\xab\f\x01J\x04\b\x02\x10\x03J\x04\b\x03\x10\x04R\tsecp256k1R\asr25519\"/\n" + "\aPreface\x12$\n" + - "\x0ests_public_key\x18\x01 \x01(\fR\fstsPublicKey\"\xe4\x01\n" + + "\x0ests_public_key\x18\x01 \x01(\fR\fstsPublicKey\"\xe3\x03\n" + "\tHandshake\x126\n" + - "\rnode_auth_key\x18\x01 \x01(\v2\x12.p2p.NodePublicKeyR\vnodeAuthKey\x12\"\n" + - "\rnode_auth_sig\x18\x02 \x01(\fR\vnodeAuthSig\x12 \n" + - "\tself_addr\x18\x04 \x01(\tH\x00R\bselfAddr\x88\x01\x01\x12\x1b\n" + - "\tpex_addrs\x18\x05 \x03(\tR\bpexAddrs\x12.\n" + - "\x13sei_giga_connection\x18\x03 \x01(\bR\x11seiGigaConnectionB\f\n" + + "\rnode_auth_key\x18\x01 \x01(\v2\x12.p2p.NodePublicKeyR\vnodeAuthKey\x12*\n" + + "\rnode_auth_sig\x18\x02 \x01(\fB\x06؈\xe2\xab\f@R\vnodeAuthSig\x12)\n" + + "\tself_addr\x18\x04 \x01(\tB\a؈\xe2\xab\f\xc0\x02H\x00R\bselfAddr\x88\x01\x01\x12*\n" + + "\tpex_addrs\x18\x05 \x03(\tB\rЈ\xe2\xab\fd؈\xe2\xab\f\xc0\x02R\bpexAddrs\x12.\n" + + "\x13sei_giga_connection\x18\x03 \x01(\bR\x11seiGigaConnection\x129\n" + + "\x12validator_auth_key\x18\x06 \x01(\fB\x06؈\xe2\xab\f H\x01R\x10validatorAuthKey\x88\x01\x01\x129\n" + + "\x12validator_auth_sig\x18\a \x01(\fB\x06؈\xe2\xab\f@H\x02R\x10validatorAuthSig\x88\x01\x01\x12%\n" + + "\aevm_rpc\x18\b \x01(\tB\a؈\xe2\xab\f\x80\x10H\x03R\x06evmRpc\x88\x01\x01:\x06\xe8\x88\xe2\xab\f\x01B\f\n" + + "\n" + + "_self_addrB\x15\n" + + "\x13_validator_auth_keyB\x15\n" + + "\x13_validator_auth_sigB\n" + "\n" + - "_self_addrBBZ@github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/pbb\x06proto3" + "\b_evm_rpcBBZ@github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/pbb\x06proto3" var ( file_p2p_p2p_proto_rawDescOnce sync.Once diff --git a/sei-tendermint/internal/p2p/pb/p2p.wireguard.go b/sei-tendermint/internal/p2p/pb/p2p.wireguard.go index 689923c9b5..b90c3ce599 100644 --- a/sei-tendermint/internal/p2p/pb/p2p.wireguard.go +++ b/sei-tendermint/internal/p2p/pb/p2p.wireguard.go @@ -7,6 +7,14 @@ import ( reflect "reflect" ) +func (*NodePublicKey) MaxSize() int { + return 34 +} + +func (*Handshake) MaxSize() int { + return 34878 +} + func init() { // Register the wireguard.Schema generated for p2p.PacketPing. runtime.MustRegister[*PacketPing](runtime.Schema{}) @@ -30,7 +38,7 @@ func init() { // Register the wireguard.Schema generated for p2p.NodePublicKey. runtime.MustRegister[*NodePublicKey](runtime.Schema{ - 1: {MaxCount: 1}, + 1: {MaxCount: 1, MaxSize: 32}, }) // Register the wireguard.Schema generated for p2p.Preface. @@ -41,9 +49,13 @@ func init() { // Register the wireguard.Schema generated for p2p.Handshake. runtime.MustRegister[*Handshake](runtime.Schema{ 1: {MaxCount: 1, Nested: utils.Some(reflect.TypeFor[*NodePublicKey]())}, - 2: {MaxCount: 1}, - 4: {MaxCount: 1}, + 2: {MaxCount: 1, MaxSize: 64}, + 4: {MaxCount: 1, MaxSize: 320}, + 5: {MaxCount: 100, MaxSize: 320}, 3: {MaxCount: 1}, + 6: {MaxCount: 1, MaxSize: 32}, + 7: {MaxCount: 1, MaxSize: 64}, + 8: {MaxCount: 1, MaxSize: 2048}, }) } diff --git a/sei-tendermint/internal/p2p/router.go b/sei-tendermint/internal/p2p/router.go index 4b62fe8827..43036e5fec 100644 --- a/sei-tendermint/internal/p2p/router.go +++ b/sei-tendermint/internal/p2p/router.go @@ -225,20 +225,25 @@ func (r *Router) acceptPeersRoutine(ctx context.Context) error { if r.options.PexOnHandshake { pexAddrs = r.Advertise(MaxPexAddrs) } - hConn, err := handshake(handshakeCtx, tcpConn, r.privKey, handshakeSpec{ + spec := handshakeSpec{ SelfAddr: r.options.SelfAddress, // Listener has to send pex data, so that dialer can learn about more peers in // case listener does not have capacity for new connections. // Dialer also could potentially send pex data, but there is no benefit from doing so: // - if listener is full, then it won't use the new data and it won't gossip it further either, since only verified data is gossiped. // - if it is not full, then the connection will be established and pex data will be sent the regular way using PEX protocol. - PexAddrs: pexAddrs, - SeiGigaConnection: r.giga.IsPresent(), - }) + PexAddrs: pexAddrs, + } + var offer utils.Option[handshakeOffer] + giga, hasGiga := r.giga.Get() + if hasGiga { + spec, offer = giga.fillInboundHandshake(spec) + } + hConn, err := handshake(handshakeCtx, tcpConn, r.privKey, spec, offer) if err != nil { return fmt.Errorf("handshake(): %w", err) } - if giga, ok := r.giga.Get(); ok && hConn.msg.SeiGigaConnection { + if hasGiga && hConn.msg.SeiGigaConnection { release() return giga.RunInboundConn(ctx, hConn) } @@ -297,7 +302,7 @@ func (r *Router) dialPeersRoutine(ctx context.Context) error { hConn, err = handshake(ctx, tcpConn, r.privKey, handshakeSpec{ SelfAddr: r.options.SelfAddress, SeiGigaConnection: false, - }) + }, utils.None[handshakeOffer]()) if err != nil { return fmt.Errorf("handshake(): %w", err) } diff --git a/sei-tendermint/internal/p2p/router_test.go b/sei-tendermint/internal/p2p/router_test.go index cd7b3b8eb8..38793d8468 100644 --- a/sei-tendermint/internal/p2p/router_test.go +++ b/sei-tendermint/internal/p2p/router_test.go @@ -23,7 +23,7 @@ import ( ) func (r *Router) handshakeV2(ctx context.Context, conn tcp.Conn, dialAddr utils.Option[NodeAddress]) (*handshakedConn, types.NodeInfo, error) { - hConn, err := handshake(ctx, conn, r.privKey, handshakeSpec{SeiGigaConnection: false}) + hConn, err := handshake(ctx, conn, r.privKey, handshakeSpec{SeiGigaConnection: false}, utils.None[handshakeOffer]()) if err != nil { return nil, types.NodeInfo{}, err } diff --git a/sei-tendermint/internal/rpc/core/autobahn_env_test.go b/sei-tendermint/internal/rpc/core/autobahn_env_test.go index 9d25739611..e14ddbaf95 100644 --- a/sei-tendermint/internal/rpc/core/autobahn_env_test.go +++ b/sei-tendermint/internal/rpc/core/autobahn_env_test.go @@ -1,6 +1,7 @@ package core import ( + "net/url" "testing" "time" @@ -42,6 +43,7 @@ func newAutobahnBroadcastEnv(t *testing.T) *Environment { valKey.Public(): { Key: nodeKey.Public(), HostPort: tcp.HostPort{Hostname: "127.0.0.1", Port: 26657}, + EVMRPC: *utils.OrPanic1(url.Parse("http://127.0.0.1:8545")), }, } blockStore, err := blockstore.New(memblock.NewBlockDB()) diff --git a/sei-tendermint/libs/utils/url.go b/sei-tendermint/libs/utils/url.go new file mode 100644 index 0000000000..3d99058c76 --- /dev/null +++ b/sei-tendermint/libs/utils/url.go @@ -0,0 +1,36 @@ +package utils + +import ( + "fmt" + "net" + "net/url" + "strings" +) + +// CheckHTTPURL reports whether u is an http or https URL with a host and +// without userinfo. It does not reject loopback, link-local, unspecified, +// zone IDs, or names that resolve to those. +func CheckHTTPURL(u url.URL) error { + if u.Scheme != "http" && u.Scheme != "https" { + return fmt.Errorf("scheme %q, want http or https", u.Scheme) + } + if u.Host == "" { + return fmt.Errorf("missing host") + } + if u.User != nil { + return fmt.Errorf("userinfo not allowed") + } + return nil +} + +// IsLoopbackOrLinkLocalURL reports whether u's host is the name "localhost", a +// loopback IP, an unspecified IP (0.0.0.0 / ::), or a link-local IP. A host +// that resolves to one of those through DNS is not detected. +func IsLoopbackOrLinkLocalURL(u url.URL) bool { + host := u.Hostname() + if strings.EqualFold(host, "localhost") { + return true + } + ip := net.ParseIP(host) + return ip != nil && (ip.IsLoopback() || ip.IsUnspecified() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast()) +} diff --git a/sei-tendermint/libs/utils/url_test.go b/sei-tendermint/libs/utils/url_test.go new file mode 100644 index 0000000000..2c2be5290f --- /dev/null +++ b/sei-tendermint/libs/utils/url_test.go @@ -0,0 +1,41 @@ +package utils + +import ( + "net/url" + "testing" + + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" +) + +func TestCheckHTTPURL(t *testing.T) { + for _, s := range []string{"http://validator.example:8545", "https://validator.example:8545"} { + require.NoError(t, CheckHTTPURL(*OrPanic1(url.Parse(s)))) + } + for _, s := range []string{"file://localhost/rpc", "ws://validator.example:8545", "http://", "http://u:p@validator.example:8545"} { + require.Error(t, CheckHTTPURL(*OrPanic1(url.Parse(s)))) + } +} + +func TestIsLoopbackOrLinkLocalURL(t *testing.T) { + for _, s := range []string{ + "http://localhost:8545", + "http://LocalHost:8545", + "http://127.0.0.1:8545", + "http://127.9.9.9:8545", + "http://[::1]:8545", + "http://169.254.169.254/", + "http://[fe80::1]:8545", + "http://0.0.0.0:8545", + "http://[::]:8545", + } { + require.True(t, IsLoopbackOrLinkLocalURL(*OrPanic1(url.Parse(s)))) + } + for _, s := range []string{ + "http://validator.example:8545", + "http://10.0.0.1:8545", + "http://8.8.8.8:8545", + "http://localhost.validator.example:8545", + } { + require.False(t, IsLoopbackOrLinkLocalURL(*OrPanic1(url.Parse(s)))) + } +} diff --git a/sei-tendermint/node/setup.go b/sei-tendermint/node/setup.go index 310ce54103..7977d5678f 100644 --- a/sei-tendermint/node/setup.go +++ b/sei-tendermint/node/setup.go @@ -215,7 +215,7 @@ func loadAutobahnCommittee(autobahnConfigFile string) (*config.AutobahnFileConfi validatorAddrs[entry.ValidatorKey] = p2p.GigaNodeAddr{ Key: entry.NodeKey, HostPort: entry.Address, - EVMRPC: entry.EVMRPC.URL, + EVMRPC: *entry.EVMRPC.URL, } } return fc, validatorAddrs, nil