diff --git a/Bootstrap.go b/Bootstrap.go index 89b4654..3d974b1 100644 --- a/Bootstrap.go +++ b/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 } diff --git a/Commands.go b/Commands.go index 93fa618..54a0566 100644 --- a/Commands.go +++ b/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 diff --git a/Config.go b/Config.go index 1280b8c..2ebc425 100644 --- a/Config.go +++ b/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 diff --git a/Connection.go b/Connection.go index 687660e..1c451f2 100644 --- a/Connection.go +++ b/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 diff --git a/Filter.go b/Filter.go index a87551b..e9aec53 100644 --- a/Filter.go +++ b/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 diff --git a/Message Encoding.go b/Message Encoding.go index 2d49338..2799bab 100644 --- a/Message Encoding.go +++ b/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) } diff --git a/Network.go b/Network.go index 77d8377..54347d9 100644 --- a/Network.go +++ b/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) } diff --git a/dht/DHT Lite.go b/dht/DHT Lite.go index 1749779..5ad7a63 100644 --- a/dht/DHT Lite.go +++ b/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,