From 82033e124a58080be1fb6ac7e9b588e85b903076 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Wed, 19 May 2021 14:56:39 +0200 Subject: [PATCH] Finalize Traverse message implementation. Close #23 Remove createVirtualAnnouncement function, no longer needed. --- Bootstrap.go | 2 +- Command Traverse.go | 106 ++++++++++++++++---------------------------- Commands.go | 4 +- Connection.go | 30 ++++++++----- Message Encoding.go | 11 ++++- Peer ID.go | 34 +++++++++----- 6 files changed, 95 insertions(+), 92 deletions(-) diff --git a/Bootstrap.go b/Bootstrap.go index 337837d..3b3928a 100644 --- a/Bootstrap.go +++ b/Bootstrap.go @@ -207,7 +207,7 @@ func contactArbitraryPeer(publicKey *btcec.PublicKey, address *net.UDPAddr, rece return false } - sendAllNetworks(publicKey, &PacketRaw{Command: CommandAnnouncement, Payload: packets[0].raw}, address, receiverPortInternal, &bootstrapFindSelf{}) + sendAllNetworks(publicKey, &PacketRaw{Command: CommandAnnouncement, Payload: packets[0].raw}, address, receiverPortInternal, nil, &bootstrapFindSelf{}) return true } diff --git a/Command Traverse.go b/Command Traverse.go index fdc153a..2e6b26e 100644 --- a/Command Traverse.go +++ b/Command Traverse.go @@ -7,11 +7,8 @@ Author: Peter Kleissner package core import ( - "errors" - "net" + "math/rand" "time" - - "github.com/btcsuite/btcd/btcec" ) // cmdTraverseForward handles an incoming traverse message that should be forwarded to another peer @@ -56,17 +53,35 @@ func (peer *PeerInfo) cmdTraverseReceive(msg *MessageTraverse) { return } - // already an active connection established? nothing todo. - peerOriginalSender := PeerlistLookup(msg.SignerPublicKey) - if peerOriginalSender != nil { - // could process the packet? - //if connections := peerOriginalSender.GetConnections(true); len(connections) > 0 { - // rawPacketsIncoming <- networkWire{network: connections[0].Network, sender: addressOriginalSender, raw: msg.EmbeddedPacketRaw, receiverPublicKey: peerPublicKey, unicast: true} - //} + // Already an active connection established? The relayed message should not be needed in this case. + // This could be changed in the future if it turns out that there are 1-way connection issues. + if peerTarget := PeerlistLookup(msg.SignerPublicKey); peerTarget != nil { + return + } + + // parse IP addresses of the original sender + var addresses []*peerAddress + + if !msg.IPv4.IsUnspecified() { + port := msg.PortIPv4 + if msg.PortIPv4ReportedExternal > 0 { + port = msg.PortIPv4ReportedExternal + } + addresses = append(addresses, &peerAddress{IP: msg.IPv4, Port: port, PortInternal: 0}) + } + if !msg.IPv6.IsUnspecified() { + port := msg.PortIPv6 + if msg.PortIPv6ReportedExternal > 0 { + port = msg.PortIPv6ReportedExternal + } + addresses = append(addresses, &peerAddress{IP: msg.IPv6, Port: port, PortInternal: 0}) + } + if len(addresses) == 0 { return } // ---- fork packetWorker to decode and validate embedded packet --- + // Due to missing connection and other embedded details in the message (such as ports), the packet is not just simply queued to rawPacketsIncoming. decoded, senderPublicKey, err := PacketDecrypt(msg.EmbeddedPacketRaw, peerPublicKey) if err != nil { return @@ -81,65 +96,22 @@ func (peer *PeerInfo) cmdTraverseReceive(msg *MessageTraverse) { return } - // -------- - virtualMessage := &MessageRaw{PacketRaw: *decoded} - announce, err := msgDecodeAnnouncement(virtualMessage) - if err != nil { - return - } + // process the packet and create a virtual peer + raw := &MessageRaw{SenderPublicKey: senderPublicKey, PacketRaw: *decoded, connection: nil} + peerV := &PeerInfo{PublicKey: senderPublicKey, connectionActive: nil, connectionLatest: nil, NodeID: publicKey2NodeID(senderPublicKey), messageSequence: rand.Uint32(), isVirtual: true, targetAddresses: addresses} - // Proper handling of announcement todo, virtual announcement function - var hashesNotFound [][]byte - if announce.Actions&(1< 0 { - hashesNotFound = append(hashesNotFound, peer.NodeID) - } - if announce.Actions&(1< 0 && len(announce.FindPeerKeys) > 0 { - for _, findPeer := range announce.FindPeerKeys { - hashesNotFound = append(hashesNotFound, findPeer.Hash) - } - } - if announce.Actions&(1< 0 { - for _, findHash := range announce.FindDataKeys { - hashesNotFound = append(hashesNotFound, findHash.Hash) - } - } + // process it! + switch decoded.Command { + case CommandAnnouncement: // Announce + if announce, _ := msgDecodeAnnouncement(raw); announce != nil { + if len(announce.UserAgent) > 0 { + peerV.UserAgent = announce.UserAgent + } + peerV.Features = announce.Features - // TODO - //peer.sendResponse(announce.Sequence, true, nil, nil, hashesNotFound) - //packets, err := msgEncodeResponse(true, nil, nil, hashesNotFound) - //sendAllNetworks() - - if !msg.IPv4.IsUnspecified() { - addressOriginalSenderIPv4 := &net.UDPAddr{IP: msg.IPv4, Port: int(msg.PortIPv4)} - if msg.PortIPv4ReportedExternal > 0 { - addressOriginalSenderIPv4.Port = int(msg.PortIPv4ReportedExternal) + peerV.cmdAnouncement(announce) } - contactArbitraryPeer(msg.SignerPublicKey, addressOriginalSenderIPv4, 0) - } - - if !msg.IPv6.IsUnspecified() { - addressOriginalSenderIPv6 := &net.UDPAddr{IP: msg.IPv6, Port: int(msg.PortIPv6)} - if msg.PortIPv4ReportedExternal > 0 { - addressOriginalSenderIPv6.Port = int(msg.PortIPv6ReportedExternal) - } - - contactArbitraryPeer(msg.SignerPublicKey, addressOriginalSenderIPv6, 0) + default: } } - -// createVirtualAnnouncement is temporary code and will be improved. -func createVirtualAnnouncement(network *Network, receiverPublicKey *btcec.PublicKey, sequenceData interface{}) (raw []byte, err error) { - packets := msgEncodeAnnouncement(true, ShouldSendFindSelf(), nil, nil, nil) - if len(packets) == 0 || packets[0].err != nil { - return nil, errors.New("error creating virtual packet") - } - - packet := &PacketRaw{Command: CommandAnnouncement, Payload: packets[0].raw} - packet.setSelfReportedPorts(network) - - packet.Sequence = msgArbitrarySequence(receiverPublicKey, sequenceData).sequence - packet.Protocol = ProtocolVersion - - return PacketEncrypt(peerPrivateKey, receiverPublicKey, packet) -} diff --git a/Commands.go b/Commands.go index 509461a..a1b3d61 100644 --- a/Commands.go +++ b/Commands.go @@ -80,7 +80,7 @@ func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) { if stored && len(data) > 0 { filesEmbed = append(filesEmbed, EmbeddedFileData{ID: findHash, Data: data}) } else if stored { - selfRecord := selfPeerRecord(msg.connection.Network) + selfRecord := selfPeerRecord() hash2Peers = append(hash2Peers, Hash2Peer{ID: findHash, Storing: []PeerRecord{selfRecord}}) } else { hashesNotFound = append(hashesNotFound, findHash.Hash) @@ -150,7 +150,7 @@ func (peer *PeerInfo) cmdResponse(msg *MessageResponse) { } for _, hash2Peer := range msg.Hash2Peers { - info.QueueResult(&dht.NodeMessage{SenderID: peer.NodeID, Closest: records2Nodes(hash2Peer.Closest, msg.connection.Network, peer), Storing: records2Nodes(hash2Peer.Storing, msg.connection.Network, peer)}) + info.QueueResult(&dht.NodeMessage{SenderID: peer.NodeID, Closest: records2Nodes(hash2Peer.Closest, peer), Storing: records2Nodes(hash2Peer.Storing, peer)}) if hash2Peer.IsLast { info.Done() diff --git a/Connection.go b/Connection.go index 351a348..9ce1507 100644 --- a/Connection.go +++ b/Connection.go @@ -46,7 +46,7 @@ func (c *Connection) Equal(other *Connection) bool { // IsLocal checks if the connection is a local network one (LAN) func (c *Connection) IsLocal() bool { - return IsIPLocal(c.Address.IP) + return c != nil && IsIPLocal(c.Address.IP) } // IsIPv4 checks if the connection is using IPv4 @@ -300,11 +300,9 @@ func (c *Connection) send(packet *PacketRaw, receiverPublicKey *btcec.PublicKey, err = c.Network.send(c.Address.IP, c.Address.Port, raw) - // Send Traverse message if the peer is behind a NAT and this is the first message - if err == nil && isFirstPacket && c.IsBehindNAT() && c.traversePeer != nil { - if raw, err := createVirtualAnnouncement(c.Network, receiverPublicKey, &bootstrapFindSelf{}); err == nil { - c.traversePeer.sendTraverse(raw, receiverPublicKey) - } + // Send Traverse message if the peer is behind a NAT and this is the first message. Only for Announcement. + if err == nil && isFirstPacket && c.IsBehindNAT() && c.traversePeer != nil && packet.Command == CommandAnnouncement { + c.traversePeer.sendTraverse(packet, receiverPublicKey) } return err @@ -312,6 +310,12 @@ func (c *Connection) send(packet *PacketRaw, receiverPublicKey *btcec.PublicKey, // send sends a raw packet to the peer. Only uses active connections. func (peer *PeerInfo) send(packet *PacketRaw) (err error) { + if peer.isVirtual { // special case for peers that were not contacted before + for _, address := range peer.targetAddresses { + sendAllNetworks(peer.PublicKey, packet, &net.UDPAddr{IP: address.IP, Port: int(address.Port)}, address.PortInternal, peer.traversePeer, nil) + } + return + } if len(peer.connectionActive) == 0 { return errors.New("no valid connection to peer") } @@ -364,9 +368,7 @@ func (peer *PeerInfo) sendConnection(packet *PacketRaw, connection *Connection) // sendAllNetworks sends a raw packet via all networks. It assigns a new sequence for each sent packet. // receiverPortInternal is important for NAT detection and sending the traverse message. -func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remote *net.UDPAddr, receiverPortInternal uint16, sequenceData interface{}) (err error) { - successCount := 0 - +func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remote *net.UDPAddr, receiverPortInternal uint16, traversePeer *PeerInfo, sequenceData interface{}) (err error) { networksMutex.RLock() defer networksMutex.RUnlock() @@ -375,14 +377,20 @@ func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remo networksTarget = networks6 } + successCount := 0 + isFirstPacket := true + for _, network := range networksTarget { // Do not mix link-local unicast targets with non link-local networks (only when iface is known, i.e. not catch all local) if network.iface != nil && remote.IP.IsLinkLocalUnicast() != network.address.IP.IsLinkLocalUnicast() { continue } - packet.Sequence = msgArbitrarySequence(receiverPublicKey, sequenceData).sequence - err = (&Connection{Network: network, Address: remote, PortInternal: receiverPortInternal}).send(packet, receiverPublicKey, true) + if sequenceData != nil { + packet.Sequence = msgArbitrarySequence(receiverPublicKey, sequenceData).sequence + } + err = (&Connection{Network: network, Address: remote, PortInternal: receiverPortInternal, traversePeer: traversePeer}).send(packet, receiverPublicKey, isFirstPacket) + isFirstPacket = false if err == nil { successCount++ diff --git a/Message Encoding.go b/Message Encoding.go index b50780d..1f8df30 100644 --- a/Message Encoding.go +++ b/Message Encoding.go @@ -988,7 +988,16 @@ func (peer *PeerInfo) sendResponse(sequence uint32, sendUA bool, hash2Peers []Ha } // sendTraverse sends a traverse message -func (peer *PeerInfo) sendTraverse(embeddedPacketRaw []byte, receiverEnd *btcec.PublicKey) (err error) { +func (peer *PeerInfo) sendTraverse(packet *PacketRaw, receiverEnd *btcec.PublicKey) (err error) { + packet.Protocol = ProtocolVersion + // self-reported ports are not set, as this isn't sent via a specific network but a relay + //packet.setSelfReportedPorts(c.Network) + + embeddedPacketRaw, err := PacketEncrypt(peerPrivateKey, receiverEnd, packet) + if err != nil { + return err + } + packetRaw, err := msgEncodeTraverse(peerPrivateKey, embeddedPacketRaw, receiverEnd, peer.PublicKey) if err != nil { return err diff --git a/Peer ID.go b/Peer ID.go index cd19c03..6ff7c11 100644 --- a/Peer ID.go +++ b/Peer ID.go @@ -91,12 +91,21 @@ type PeerInfo struct { IsRootPeer bool // Whether the peer is a trusted root peer. UserAgent string // User Agent reported by remote peer. Empty if no Announcement/Response message was yet received. Features uint8 // Feature bit array. 0 = IPv4_LISTEN, 1 = IPv6_LISTEN + isVirtual bool // Whether it is a virtual peer for establishing a connection. + targetAddresses []*peerAddress // Virtual peer: Addresses to send any replies. + traversePeer *PeerInfo // Virtual peer: Same field as in connection. // statistics StatsPacketSent uint64 // Count of packets sent StatsPacketReceived uint64 // Count of packets received } +type peerAddress struct { + IP net.IP + Port uint16 + PortInternal uint16 +} + // peerList keeps track of all peers var peerList map[[btcec.PubKeyBytesLenCompressed]byte]*PeerInfo var peerlistMutex sync.RWMutex @@ -191,25 +200,30 @@ func publicKey2NodeID(publicKey *btcec.PublicKey) (nodeID []byte) { // records2Nodes translates infoPeer structures to nodes. If the reported nodes are not in the peer table, it will create temporary PeerInfo structures. // LastContact is passed on in the Node.LastSeen field. -// It requires the network parameter which must be the same as caller/supplier. This ensures that peer details do not "jump" between physical network adapters. -func records2Nodes(records []PeerRecord, network *Network, peerSource *PeerInfo) (nodes []*dht.Node) { +func records2Nodes(records []PeerRecord, peerSource *PeerInfo) (nodes []*dht.Node) { for _, record := range records { if record.IsBadQuality() { continue } var peer *PeerInfo - if peer = PeerlistLookup(record.PublicKey); peer == nil { + if record.PublicKey.IsEqual(peerSource.PublicKey) { + // Special case if peer that stores info = sender. In that case IP:Port in the record would be empty anyway. + peer = peerSource + } else if peer = PeerlistLookup(record.PublicKey); peer == nil { // Create temporary peer which is not added to the global list and not added to Kademlia. + var addresses []*peerAddress + port := record.Port if record.PortReportedExternal > 0 { // Use the external port if available port = record.PortReportedExternal } + addresses = append(addresses, &peerAddress{IP: record.IP, Port: port, PortInternal: record.PortReportedInternal}) + // traversePeer is set to the peer who provided the node information. - connection := &Connection{Network: network, Address: &net.UDPAddr{IP: record.IP, Port: int(port)}, Status: ConnectionActive, PortInternal: record.PortReportedInternal, PortExternal: record.PortReportedExternal, traversePeer: peerSource} - peer = &PeerInfo{PublicKey: record.PublicKey, connectionActive: []*Connection{connection}, connectionLatest: connection, NodeID: publicKey2NodeID(record.PublicKey), messageSequence: rand.Uint32()} + peer = &PeerInfo{PublicKey: record.PublicKey, connectionActive: nil, connectionLatest: nil, NodeID: publicKey2NodeID(record.PublicKey), messageSequence: rand.Uint32(), isVirtual: true, targetAddresses: addresses, traversePeer: peerSource} } nodes = append(nodes, &dht.Node{ID: peer.NodeID, LastSeen: record.LastContactT, Info: peer}) @@ -219,12 +233,12 @@ func records2Nodes(records []PeerRecord, network *Network, peerSource *PeerInfo) } // selfPeerRecord returns self as peer record -func selfPeerRecord(network *Network) (result PeerRecord) { +func selfPeerRecord() (result PeerRecord) { return PeerRecord{ - PublicKey: peerPublicKey, - NodeID: nodeID, - IP: network.address.IP, - Port: uint16(network.address.Port), + PublicKey: peerPublicKey, + NodeID: nodeID, + //IP: network.address.IP, + //Port: uint16(network.address.Port), LastContact: 0, } }