ceremonyclient/node/consensus/data/message_validators.go
petricadaipegsp d1e65c1c92
Support frame fragmentation and dissemination (#396)
* Add clock frame fragment message

* Add clock frame fragment validation

* Add clock frame fragmentation utilities

* Add clock frame fragmentation message handling

* Report publication errors

* Publish info list after frame

* Add frame publish configuration

* Publish clock frame fragments

* Update BlossomSub dashboard

* Publish clock frame fragments in parallel
2024-12-02 16:25:10 -06:00

159 lines
4.7 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package data
import (
"encoding/binary"
"fmt"
"time"
"github.com/libp2p/go-libp2p/core/peer"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/anypb"
"source.quilibrium.com/quilibrium/monorepo/go-libp2p-blossomsub/pb"
"source.quilibrium.com/quilibrium/monorepo/node/p2p"
"source.quilibrium.com/quilibrium/monorepo/node/protobufs"
)
func (e *DataClockConsensusEngine) validateFrameMessage(peerID peer.ID, message *pb.Message) p2p.ValidationResult {
msg := &protobufs.Message{}
if err := proto.Unmarshal(message.Data, msg); err != nil {
return p2p.ValidationResultReject
}
a := &anypb.Any{}
if err := proto.Unmarshal(msg.Payload, a); err != nil {
return p2p.ValidationResultReject
}
switch a.TypeUrl {
case protobufs.ClockFrameType:
frame := &protobufs.ClockFrame{}
if err := proto.Unmarshal(a.Value, frame); err != nil {
return p2p.ValidationResultReject
}
if ts := time.UnixMilli(frame.Timestamp); time.Since(ts) > 2*time.Minute {
return p2p.ValidationResultIgnore
}
return p2p.ValidationResultAccept
default:
return p2p.ValidationResultReject
}
}
func (e *DataClockConsensusEngine) validateFrameFragmentMessage(peerID peer.ID, message *pb.Message) p2p.ValidationResult {
msg := &protobufs.Message{}
if err := proto.Unmarshal(message.Data, msg); err != nil {
return p2p.ValidationResultReject
}
a := &anypb.Any{}
if err := proto.Unmarshal(msg.Payload, a); err != nil {
return p2p.ValidationResultReject
}
switch a.TypeUrl {
case protobufs.ClockFrameFragmentType:
fragment := &protobufs.ClockFrameFragment{}
if err := proto.Unmarshal(a.Value, fragment); err != nil {
return p2p.ValidationResultReject
}
if err := fragment.Validate(); err != nil {
return p2p.ValidationResultReject
}
if ts := time.UnixMilli(fragment.Timestamp); time.Since(ts) > 2*time.Minute {
return p2p.ValidationResultIgnore
}
return p2p.ValidationResultAccept
default:
return p2p.ValidationResultReject
}
}
func (e *DataClockConsensusEngine) validateTxMessage(peerID peer.ID, message *pb.Message) p2p.ValidationResult {
msg := &protobufs.Message{}
if err := proto.Unmarshal(message.Data, msg); err != nil {
return p2p.ValidationResultReject
}
a := &anypb.Any{}
if err := proto.Unmarshal(msg.Payload, a); err != nil {
return p2p.ValidationResultReject
}
switch a.TypeUrl {
case protobufs.TokenRequestType:
tx := &protobufs.TokenRequest{}
if err := proto.Unmarshal(a.Value, tx); err != nil {
return p2p.ValidationResultReject
}
if err := tx.Validate(); err != nil {
return p2p.ValidationResultReject
}
if mint := tx.GetMint(); mint != nil {
if len(mint.Proofs) < 3 {
return p2p.ValidationResultReject
}
if len(mint.Proofs[1]) != 4 {
return p2p.ValidationResultReject
}
if len(mint.Proofs[2]) != 8 {
return p2p.ValidationResultReject
}
// cheap hack for handling protobuf trickery: because protobufs can be
// serialized in infinite ways, message ids can be regenerated simply by
// modifying the data without affecting the underlying signed message.
// if this is encountered, go scorched earth on the sender a thank you
// message for destabilizing the network.
frameNumber := binary.BigEndian.Uint64(mint.Proofs[2])
id := fmt.Sprintf(
"mint-sign-%d-%x",
frameNumber,
mint.Signature.PublicKey.KeyValue,
)
e.validationFilterMx.Lock()
_, ok := e.validationFilter[id]
e.validationFilter[id] = struct{}{}
e.validationFilterMx.Unlock()
if ok {
e.pubSub.AddPeerScore(message.From, -1000000)
return p2p.ValidationResultIgnore
}
head, err := e.dataTimeReel.Head()
if err != nil {
panic(err)
}
if frameNumber+2 < head.FrameNumber {
return p2p.ValidationResultIgnore
}
}
if ts := time.UnixMilli(tx.Timestamp); time.Since(ts) > 10*time.Minute {
return p2p.ValidationResultIgnore
}
return p2p.ValidationResultAccept
default:
return p2p.ValidationResultReject
}
}
func (e *DataClockConsensusEngine) validateInfoMessage(peerID peer.ID, message *pb.Message) p2p.ValidationResult {
msg := &protobufs.Message{}
if err := proto.Unmarshal(message.Data, msg); err != nil {
return p2p.ValidationResultReject
}
a := &anypb.Any{}
if err := proto.Unmarshal(msg.Payload, a); err != nil {
return p2p.ValidationResultReject
}
switch a.TypeUrl {
case protobufs.DataPeerListAnnounceType:
announce := &protobufs.DataPeerListAnnounce{}
if err := proto.Unmarshal(a.Value, announce); err != nil {
return p2p.ValidationResultReject
}
if announce.Peer == nil {
return p2p.ValidationResultIgnore
}
if ts := time.UnixMilli(announce.Peer.Timestamp); time.Since(ts) > 10*time.Minute {
return p2p.ValidationResultIgnore
}
return p2p.ValidationResultAccept
default:
return p2p.ValidationResultReject
}
}