Add filters for low-level packet and high-level message interception for debugging purposes. Close #4

Bootstrap.go

@@ -201,12 +201,16 @@ func contactArbitraryPeer(publicKey *btcec.PublicKey, address *net.UDPAddr, rece
 		return false
 	}
 
-	packets := msgEncodeAnnouncement(true, ShouldSendFindSelf(), nil, nil, nil)
+	findSelf := ShouldSendFindSelf()
+	packets := msgEncodeAnnouncement(true, findSelf, nil, nil, nil)
 	if len(packets) == 0 || packets[0].err != nil {
 		return false
 	}
+	raw := &PacketRaw{Command: CommandAnnouncement, Payload: packets[0].raw}
 
-	sendAllNetworks(publicKey, &PacketRaw{Command: CommandAnnouncement, Payload: packets[0].raw}, address, receiverPortInternal, nil, &bootstrapFindSelf{})
+	Filters.MessageOutAnnouncement(publicKey, nil, raw, findSelf, nil, nil, nil)
+
+	sendAllNetworks(publicKey, raw, address, receiverPortInternal, nil, &bootstrapFindSelf{})
 
 	return true
 }

Commands.go

@@ -198,7 +198,11 @@ func (peer *PeerInfo) cmdPing(msg *MessageRaw) {
 		return
 	}
 
-	peer.send(&PacketRaw{Command: CommandPong, Sequence: msg.Sequence})
+	raw := &PacketRaw{Command: CommandPong, Sequence: msg.Sequence}
+
+	Filters.MessageOutPong(peer, raw)
+
+	peer.send(raw)
 }
 
 // cmdPong handles an incoming pong message

Config.go

@@ -16,7 +16,7 @@ import (
 )
 
 // Version is the current core library version
-const Version = "Alpha 3/28.07.2021"
+const Version = "Alpha 3/31.07.2021"
 
 var config struct {
 	LogFile string `yaml:"LogFile"` // Log file

Connection.go

@@ -327,6 +327,8 @@ func (c *Connection) send(packet *PacketRaw, receiverPublicKey *btcec.PublicKey,
 	packet.Protocol = ProtocolVersion
 	packet.setSelfReportedPorts(c.Network)
 
+	Filters.PacketOut(packet, receiverPublicKey, c)
+
 	raw, err := PacketEncrypt(peerPrivateKey, receiverPublicKey, packet)
 	if err != nil {
 		return err

Filter.go

@@ -3,7 +3,7 @@ File Name:  Filter.go
 Copyright:  2021 Peernet s.r.o.
 Author:     Peter Kleissner
 
-Filters allow the caller to intercept events to log, modify, or prevent.
+Filters allow the caller to intercept events. The filter functions must not modify any data.
 */
 
 package core
@@ -12,6 +12,7 @@ import (
 	"log"
 
 	"github.com/PeernetOfficial/core/dht"
+	"github.com/btcsuite/btcd/btcec"
 )
 
 // Filters contains all functions to install the hook. Use nil for unused.
@@ -33,6 +34,33 @@ var Filters struct {
 
 	// IncomingRequest receives all incoming information requests. The action field is set accordingly.
 	IncomingRequest func(peer *PeerInfo, Action int, Key []byte, Info interface{})
+
+	// PacketIn is a low-level filter for incoming packets after they are decrypted.
+	// Traverse messages are not covered.
+	PacketIn func(packet *PacketRaw, senderPublicKey *btcec.PublicKey, connection *Connection)
+
+	// PacketOut is a low-level filter for outgoing packets before they are encrypted.
+	// IPv4 broadcast, IPv6 multicast, and Traverse messages are not covered.
+	PacketOut func(packet *PacketRaw, receiverPublicKey *btcec.PublicKey, connection *Connection)
+
+	// MessageIn is a high-level filter for decoded incoming messages. message is of type nil, MessageAnnouncement, MessageResponse, or MessageTraverse
+	MessageIn func(peer *PeerInfo, raw *MessageRaw, message interface{})
+
+	// MessageOutAnnouncement is a high-level filter for outgoing announcements. Peer is nil on first contact.
+	// Broadcast and Multicast messages are not covered.
+	MessageOutAnnouncement func(receiverPublicKey *btcec.PublicKey, peer *PeerInfo, packet *PacketRaw, findSelf bool, findPeer []KeyHash, findValue []KeyHash, files []InfoStore)
+
+	// MessageOutResponse is a high-level filter for outgoing responses.
+	MessageOutResponse func(peer *PeerInfo, packet *PacketRaw, hash2Peers []Hash2Peer, filesEmbed []EmbeddedFileData, hashesNotFound [][]byte)
+
+	// MessageOutTraverse is a high-level filter for outgoing traverse messages.
+	MessageOutTraverse func(peer *PeerInfo, packet *PacketRaw, embeddedPacket *PacketRaw, receiverEnd *btcec.PublicKey)
+
+	// MessageOutPing is a high-level filter for outgoing pings.
+	MessageOutPing func(peer *PeerInfo, packet *PacketRaw, connection *Connection)
+
+	// MessageOutPong is a high-level filter for outgoing pongs.
+	MessageOutPong func(peer *PeerInfo, packet *PacketRaw)
 }
 
 func initFilters() {
@@ -54,6 +82,32 @@ func initFilters() {
 	if Filters.IncomingRequest == nil {
 		Filters.IncomingRequest = func(peer *PeerInfo, Action int, Key []byte, Info interface{}) {}
 	}
+	if Filters.PacketIn == nil {
+		Filters.PacketIn = func(packet *PacketRaw, senderPublicKey *btcec.PublicKey, c *Connection) {}
+	}
+	if Filters.PacketOut == nil {
+		Filters.PacketOut = func(packet *PacketRaw, receiverPublicKey *btcec.PublicKey, c *Connection) {}
+	}
+	if Filters.MessageIn == nil {
+		Filters.MessageIn = func(peer *PeerInfo, raw *MessageRaw, message interface{}) {}
+	}
+	if Filters.MessageOutAnnouncement == nil {
+		Filters.MessageOutAnnouncement = func(receiverPublicKey *btcec.PublicKey, peer *PeerInfo, packet *PacketRaw, findSelf bool, findPeer []KeyHash, findValue []KeyHash, files []InfoStore) {
+		}
+	}
+	if Filters.MessageOutResponse == nil {
+		Filters.MessageOutResponse = func(peer *PeerInfo, packet *PacketRaw, hash2Peers []Hash2Peer, filesEmbed []EmbeddedFileData, hashesNotFound [][]byte) {
+		}
+	}
+	if Filters.MessageOutTraverse == nil {
+		Filters.MessageOutTraverse = func(peer *PeerInfo, packet *PacketRaw, embeddedPacket *PacketRaw, receiverEnd *btcec.PublicKey) {}
+	}
+	if Filters.MessageOutPing == nil {
+		Filters.MessageOutPing = func(peer *PeerInfo, packet *PacketRaw, connection *Connection) {}
+	}
+	if Filters.MessageOutPong == nil {
+		Filters.MessageOutPong = func(peer *PeerInfo, packet *PacketRaw) {}
+	}
 }
 
 // DefaultLogError is the default error logging function

Message Encoding.go

@@ -974,7 +974,10 @@ func msgEncodeTraverseSetAddress(raw []byte, connectionIPv4, connectionIPv6 *Con
 
 // pingConnection sends a ping to the target peer via the specified connection
 func (peer *PeerInfo) pingConnection(connection *Connection) {
-	err := peer.sendConnection(&PacketRaw{Command: CommandPing, Sequence: peer.msgNewSequence(nil).sequence}, connection)
+	raw := &PacketRaw{Command: CommandPing, Sequence: peer.msgNewSequence(nil).sequence}
+	Filters.MessageOutPing(peer, raw, connection)
+
+	err := peer.sendConnection(raw, connection)
 	connection.LastPingOut = time.Now()
 
 	if (connection.Status == ConnectionActive || connection.Status == ConnectionRedundant) && IsNetworkErrorFatal(err) {
@@ -993,7 +996,9 @@ func (peer *PeerInfo) sendAnnouncement(sendUA, findSelf bool, findPeer []KeyHash
 
 	for _, packet := range packets {
 		packet.sequence = peer.msgNewSequence(sequenceData)
-		packet.err = peer.send(&PacketRaw{Command: CommandAnnouncement, Payload: packet.raw, Sequence: packet.sequence.sequence})
+		raw := &PacketRaw{Command: CommandAnnouncement, Payload: packet.raw, Sequence: packet.sequence.sequence}
+		Filters.MessageOutAnnouncement(peer.PublicKey, peer, raw, findSelf, findPeer, findValue, files)
+		packet.err = peer.send(raw)
 	}
 
 	return
@@ -1004,7 +1009,9 @@ func (peer *PeerInfo) sendResponse(sequence uint32, sendUA bool, hash2Peers []Ha
 	packets, err := msgEncodeResponse(sendUA, hash2Peers, filesEmbed, hashesNotFound)
 
 	for _, packet := range packets {
-		peer.send(&PacketRaw{Command: CommandResponse, Payload: packet, Sequence: sequence})
+		raw := &PacketRaw{Command: CommandResponse, Payload: packet, Sequence: sequence}
+		Filters.MessageOutResponse(peer, raw, hash2Peers, filesEmbed, hashesNotFound)
+		peer.send(raw)
 	}
 
 	return err
@@ -1026,5 +1033,9 @@ func (peer *PeerInfo) sendTraverse(packet *PacketRaw, receiverEnd *btcec.PublicK
 		return err
 	}
 
-	return peer.send(&PacketRaw{Command: CommandTraverse, Payload: packetRaw})
+	raw := &PacketRaw{Command: CommandTraverse, Payload: packetRaw}
+
+	Filters.MessageOutTraverse(peer, raw, packet, receiverEnd)
+
+	return peer.send(raw)
 }

Network.go

@@ -162,6 +162,8 @@ func packetWorker(packets <-chan networkWire) {
 
 		connection := &Connection{Network: packet.network, Address: packet.sender, Status: ConnectionActive}
 
+		Filters.PacketIn(decoded, senderPublicKey, connection)
+
 		// A peer structure will always be returned, even if the peer won't be added to the peer list.
 		peer, added := PeerlistAdd(senderPublicKey, connection)
 		if !added {
@@ -187,6 +189,8 @@ func packetWorker(packets <-chan networkWire) {
 				peer.BlockchainHeight = announce.BlockchainHeight
 				peer.BlockchainVersion = announce.BlockchainVersion
 
+				Filters.MessageIn(peer, raw, announce)
+
 				peer.cmdAnouncement(announce)
 			}
 
@@ -210,6 +214,8 @@ func packetWorker(packets <-chan networkWire) {
 				peer.BlockchainHeight = response.BlockchainHeight
 				peer.BlockchainVersion = response.BlockchainVersion
 
+				Filters.MessageIn(peer, raw, response)
+
 				peer.cmdResponse(response)
 			}
 
@@ -222,10 +228,13 @@ func packetWorker(packets <-chan networkWire) {
 				peer.BlockchainHeight = announce.BlockchainHeight
 				peer.BlockchainVersion = announce.BlockchainVersion
 
+				Filters.MessageIn(peer, raw, announce)
+
 				peer.cmdLocalDiscovery(announce)
 			}
 
 		case CommandPing: // Ping
+			Filters.MessageIn(peer, raw, nil)
 			peer.cmdPing(raw)
 
 		case CommandPong: // Ping
@@ -237,13 +246,17 @@ func packetWorker(packets <-chan networkWire) {
 				connection.RoundTripTime = rtt
 			}
 
+			Filters.MessageIn(peer, raw, nil)
+
 			peer.cmdPong(raw)
 
 		case CommandChat: // Chat [debug]
+			Filters.MessageIn(peer, raw, nil)
 			peer.cmdChat(raw)
 
 		case CommandTraverse:
 			if traverse, _ := msgDecodeTraverse(raw); traverse != nil {
+				Filters.MessageIn(peer, raw, traverse)
 				if traverse.TargetPeer.IsEqual(peerPublicKey) && traverse.AuthorizedRelayPeer.IsEqual(peer.PublicKey) {
 					peer.cmdTraverseReceive(traverse)
 				} else if traverse.AuthorizedRelayPeer.IsEqual(peerPublicKey) {
@@ -252,6 +265,7 @@ func packetWorker(packets <-chan networkWire) {
 			}
 
 		default: // Unknown command
+			Filters.MessageIn(peer, raw, nil)
 
 		}
 

dht/DHT Lite.go

@@ -39,9 +39,6 @@ type DHT struct {
 	// FilterSearchStatus is called with updates of searches in the DHT
 	FilterSearchStatus func(client *SearchClient, function, format string, v ...interface{})
 
-	// The maximum time to wait for a response to any message in Store, Get, FindNode
-	TMsgTimeout time.Duration
-
 	// TimeoutSearch is the maximum time a search may take.
 	TimeoutSearch time.Duration
 
@@ -54,7 +51,6 @@ func NewDHT(self *Node, bits, bucketSize, alpha int) *DHT {
 	return &DHT{
 		ht:                 newHashTable(self, bits, bucketSize),
 		alpha:              alpha,
-		TMsgTimeout:        2 * time.Second,
 		FilterSearchStatus: func(client *SearchClient, function, format string, v ...interface{}) {},
 		TimeoutSearch:      10 * time.Second,
 		TimeoutIR:          6 * time.Second,