diff --git a/Connection.go b/Connection.go index f1b8c69..c95743c 100644 --- a/Connection.go +++ b/Connection.go @@ -426,7 +426,7 @@ func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *protocol.Packet } if sequenceData != nil { - packet.Sequence = ArbitrarySequence(receiverPublicKey, sequenceData).SequenceNumber + packet.Sequence = networks.Sequences.ArbitrarySequence(receiverPublicKey, sequenceData).SequenceNumber } err = (&Connection{Network: network, Address: remote, PortInternal: receiverPortInternal, traversePeer: traversePeer}).send(packet, receiverPublicKey, isFirstPacket) isFirstPacket = false diff --git a/Message Send.go b/Message Send.go index 386db6b..6dae818 100644 --- a/Message Send.go +++ b/Message Send.go @@ -15,7 +15,7 @@ import ( // pingConnection sends a ping to the target peer via the specified connection func (peer *PeerInfo) pingConnection(connection *Connection) { - raw := &protocol.PacketRaw{Command: protocol.CommandPing, Sequence: NewSequence(peer.PublicKey, &peer.messageSequence, nil).SequenceNumber} + raw := &protocol.PacketRaw{Command: protocol.CommandPing, Sequence: networks.Sequences.NewSequence(peer.PublicKey, &peer.messageSequence, nil).SequenceNumber} Filters.MessageOutPing(peer, raw, connection) err := peer.sendConnection(raw, connection) @@ -37,7 +37,7 @@ func (peer *PeerInfo) sendAnnouncement(sendUA, findSelf bool, findPeer []KeyHash packets := EncodeAnnouncement(sendUA, findSelf, findPeer, findValue, files, FeatureSupport(), blockchainHeight, blockchainVersion) for _, packet := range packets { - raw := &protocol.PacketRaw{Command: protocol.CommandAnnouncement, Payload: packet, Sequence: NewSequence(peer.PublicKey, &peer.messageSequence, sequenceData).SequenceNumber} + raw := &protocol.PacketRaw{Command: protocol.CommandAnnouncement, Payload: packet, Sequence: networks.Sequences.NewSequence(peer.PublicKey, &peer.messageSequence, sequenceData).SequenceNumber} Filters.MessageOutAnnouncement(peer.PublicKey, peer, raw, findSelf, findPeer, findValue, files) peer.send(raw) } diff --git a/Message Sequence.go b/Message Sequence.go index 8464eb4..b4fa1ac 100644 --- a/Message Sequence.go +++ b/Message Sequence.go @@ -25,13 +25,16 @@ import ( "github.com/btcsuite/btcd/btcec" ) -// SequenceReplyTimeout is the round-trip timeout for message sequences. -var SequenceReplyTimeout = 20 +// SequenceManager stores all message sequence numbers that are valid at the moment +type SequenceManager struct { + ReplyTimeout int // The round-trip timeout for message sequences. -// sequences stores all sequence numbers that are valid at the moment. The value represents the time the sequence number was used. -// Key = Peer ID + Sequence Number -var sequences map[string]*SequenceExpiry -var sequencesMutex sync.Mutex + // sequences is the list of sequence numbers that are valid at the moment. The value represents the time the sequence number. + // Key = Peer ID + Sequence Number + sequences map[string]*SequenceExpiry + + sync.Mutex // synchronized access to the sequences +} // SequenceExpiry contains the decoded sequence information of a message. type SequenceExpiry struct { @@ -42,65 +45,73 @@ type SequenceExpiry struct { Data interface{} // Optional high-level data associated with the sequence } -func initMessageSequence() { - sequences = make(map[string]*SequenceExpiry) +// NewSequenceManager creates a new sequence manager. The ReplyTimeout is in seconds. The expiration function is started immediately. +func NewSequenceManager(ReplyTimeout int) (manager *SequenceManager) { + manager = &SequenceManager{ + ReplyTimeout: ReplyTimeout, + sequences: make(map[string]*SequenceExpiry), + } - // auto-delete worker to remove expired sequences - go func() { - for { - time.Sleep(time.Duration(SequenceReplyTimeout) * time.Second) - now := time.Now() + go manager.autoDeleteExpired() - sequencesMutex.Lock() - for key, sequence := range sequences { - if sequence.expires.Before(now) { - delete(sequences, key) - } + return +} + +// autoDeleteExpired deletes all sequences that are expired. +func (manager *SequenceManager) autoDeleteExpired() { + for { + time.Sleep(time.Duration(manager.ReplyTimeout) * time.Second) + now := time.Now() + + manager.Lock() + for key, sequence := range manager.sequences { + if sequence.expires.Before(now) { + delete(manager.sequences, key) } - sequencesMutex.Unlock() } - }() + manager.Unlock() + } } // NewSequence returns a new sequence and registers it. messageSequence must point to the variable holding the continuous next sequence number. // Use only for Announcement and Ping messages. -func NewSequence(publicKey *btcec.PublicKey, messageSequence *uint32, data interface{}) (info *SequenceExpiry) { +func (manager *SequenceManager) NewSequence(publicKey *btcec.PublicKey, messageSequence *uint32, data interface{}) (info *SequenceExpiry) { info = &SequenceExpiry{ SequenceNumber: atomic.AddUint32(messageSequence, 1), created: time.Now(), - expires: time.Now().Add(time.Duration(SequenceReplyTimeout) * time.Second), + expires: time.Now().Add(time.Duration(manager.ReplyTimeout) * time.Second), Data: data, } // Add the sequence to the list. Sequences are unique enough that collisions are unlikely and negligible. key := string(publicKey.SerializeCompressed()) + strconv.FormatUint(uint64(info.SequenceNumber), 10) - sequencesMutex.Lock() - sequences[key] = info - sequencesMutex.Unlock() + manager.Lock() + manager.sequences[key] = info + manager.Unlock() return } // ArbitrarySequence returns an arbitrary sequence to be used for uncontacted peers -func ArbitrarySequence(publicKey *btcec.PublicKey, data interface{}) (info *SequenceExpiry) { +func (manager *SequenceManager) ArbitrarySequence(publicKey *btcec.PublicKey, data interface{}) (info *SequenceExpiry) { info = &SequenceExpiry{ SequenceNumber: rand.Uint32(), created: time.Now(), - expires: time.Now().Add(time.Duration(SequenceReplyTimeout) * time.Second), + expires: time.Now().Add(time.Duration(manager.ReplyTimeout) * time.Second), Data: data, } // Add the sequence to the list. Sequences are unique enough that collisions are unlikely and negligible. key := string(publicKey.SerializeCompressed()) + strconv.FormatUint(uint64(info.SequenceNumber), 10) - sequencesMutex.Lock() - sequences[key] = info - sequencesMutex.Unlock() + manager.Lock() + manager.sequences[key] = info + manager.Unlock() return } // ValidateSequence validates the sequence number of an incoming message. It will set raw.sequence if valid. -func ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Duration) { +func (manager *SequenceManager) ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Duration) { // Only Response and Pong if raw.Command != protocol.CommandResponse && raw.Command != protocol.CommandPong { return true, rtt @@ -108,11 +119,11 @@ func ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Du key := string(raw.SenderPublicKey.SerializeCompressed()) + strconv.FormatUint(uint64(raw.Sequence), 10) - sequencesMutex.Lock() - defer sequencesMutex.Unlock() + manager.Lock() + defer manager.Unlock() // lookup the sequence - sequence, ok := sequences[key] + sequence, ok := manager.sequences[key] if !ok { return false, rtt } @@ -127,10 +138,10 @@ func ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Du // invalidate the sequence immediately? if invalidate { - delete(sequences, key) + delete(manager.sequences, key) } else if raw.Command == protocol.CommandResponse { // Special case CommandResponse: Extend validity in case there are follow-up responses by half of the round-trip time since they will be sent one-way. - sequence.expires = time.Now().Add(time.Duration(SequenceReplyTimeout) * time.Second / 2) + sequence.expires = time.Now().Add(time.Duration(manager.ReplyTimeout) * time.Second / 2) } raw.SequenceInfo = sequence @@ -139,7 +150,7 @@ func ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Du } // InvalidateSequence invalidates the sequence number. -func InvalidateSequence(raw *MessageRaw) { +func (manager *SequenceManager) InvalidateSequence(raw *MessageRaw) { // Only Response if raw.Command != protocol.CommandResponse { return @@ -147,7 +158,7 @@ func InvalidateSequence(raw *MessageRaw) { key := string(raw.SenderPublicKey.SerializeCompressed()) + strconv.FormatUint(uint64(raw.Sequence), 10) - sequencesMutex.Lock() - delete(sequences, key) - sequencesMutex.Unlock() + manager.Lock() + delete(manager.sequences, key) + manager.Unlock() } diff --git a/Network.go b/Network.go index d6135ce..85b287f 100644 --- a/Network.go +++ b/Network.go @@ -195,7 +195,7 @@ func packetWorker(packets <-chan networkWire) { case protocol.CommandResponse: // Response if response, _ := msgDecodeResponse(raw); response != nil { // Validate sequence number which prevents unsolicited responses. - if valid, rtt := ValidateSequence(raw, response.Actions&(1< 0); !valid { + if valid, rtt := networks.Sequences.ValidateSequence(raw, response.Actions&(1< 0); !valid { //Filters.LogError("packetWorker", "message with invalid sequence %d command %d from %s\n", raw.Sequence, raw.Command, raw.connection.Address.String()) // Only log for debug purposes. continue } else if rtt > 0 { @@ -237,7 +237,7 @@ func packetWorker(packets <-chan networkWire) { case protocol.CommandPong: // Ping // Validate sequence number which prevents unsolicited responses. - if valid, rtt := ValidateSequence(raw, true); !valid { + if valid, rtt := networks.Sequences.ValidateSequence(raw, true); !valid { //Filters.LogError("packetWorker", "message with invalid sequence %d command %d from %s\n", raw.Sequence, raw.Command, raw.connection.Address.String()) // Only log for debug purposes. continue } else if rtt > 0 { diff --git a/Networks.go b/Networks.go new file mode 100644 index 0000000..ccafd9e --- /dev/null +++ b/Networks.go @@ -0,0 +1,34 @@ +/* +File Name: Networks.go +Copyright: 2021 Peernet s.r.o. +Author: Peter Kleissner +*/ + +package core + +// Networks is the collection of all connected networks +type Networks struct { + // networks is a list of all connected networks + //networks4, networks6 []*Network + + // Mutex for both network lists. Higher granularity currently not needed. + //sync.RWMutex + //networksMutex sync.RWMutex + + // countListenX is the number of networks listened to, excluding link-local only listeners. This number might be different than len(networksN). + // This is useful to determine if there are any IPv4 or IPv6 listeners for potential external connections. This can be used to determine IPv4_LISTEN and IPv6_LISTEN. + //countListen4, countListen6 int64 + + Sequences *SequenceManager +} + +// ReplyTimeout is the round-trip timeout for message sequences. +const ReplyTimeout = 20 + +var networks *Networks + +func initMessageSequence() { + networks = &Networks{} + + networks.Sequences = NewSequenceManager(ReplyTimeout) +}