From d148fddd1eb8eb7937e6a4fc24d4416074715957 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Wed, 19 May 2021 03:40:26 +0200 Subject: [PATCH] Simplify and streamline send code. Close #21 --- Bootstrap.go | 24 +++------ Command Traverse.go | 11 ++--- Connection.go | 118 ++++++++++++++------------------------------ Message Encoding.go | 13 +++++ Network.go | 20 ++++++++ 5 files changed, 83 insertions(+), 103 deletions(-) diff --git a/Bootstrap.go b/Bootstrap.go index d341bcd..337837d 100644 --- a/Bootstrap.go +++ b/Bootstrap.go @@ -96,7 +96,10 @@ func parseAddress(Address string) (remote *net.UDPAddr, err error) { // contact tries to contact the root peer on all networks func (peer *rootPeer) contact() { - contactArbitraryPeer(peer.publicKey, peer.addresses) + for _, address := range peer.addresses { + // Port internal is always set to 0 for root peers. It disables NAT detection and will not send out a Traverse message. + contactArbitraryPeer(peer.publicKey, address, 0) + } } // bootstrap connects to the initial set of peers. @@ -194,7 +197,7 @@ func autoMulticastBroadcast() { // contactArbitraryPeer reaches for the first time to an arbitrary peer. // It does not contact the peer if it is in the peer list, which means that a connection is already established. -func contactArbitraryPeer(publicKey *btcec.PublicKey, addresses []*net.UDPAddr) (contacted bool) { +func contactArbitraryPeer(publicKey *btcec.PublicKey, address *net.UDPAddr, receiverPortInternal uint16) (contacted bool) { if peer := PeerlistLookup(publicKey); peer != nil { return false } @@ -204,9 +207,7 @@ func contactArbitraryPeer(publicKey *btcec.PublicKey, addresses []*net.UDPAddr) return false } - for _, address := range addresses { - sendAllNetworks(publicKey, &PacketRaw{Command: CommandAnnouncement, Payload: packets[0].raw}, address, &bootstrapFindSelf{}) - } + sendAllNetworks(publicKey, &PacketRaw{Command: CommandAnnouncement, Payload: packets[0].raw}, address, receiverPortInternal, &bootstrapFindSelf{}) return true } @@ -236,19 +237,10 @@ func (peer *PeerInfo) cmdResponseBootstrapFindSelf(msg *MessageResponse, closest port = closePeer.PortReportedExternal } - // Initiate contact. Once a response comes back, the peer is actually added to the list. - if contactArbitraryPeer(closePeer.PublicKey, []*net.UDPAddr{{IP: closePeer.IP, Port: int(port)}}) { + // Initiate contact. Once a response comes back, the peer will be actually added to the peer list. + if contactArbitraryPeer(closePeer.PublicKey, &net.UDPAddr{IP: closePeer.IP, Port: int(port)}, closePeer.PortReportedInternal) { // Blacklist the target Peer ID, IP:Port for contact in the next 10 minutes. // TODO - - // If NAT is detected and the port is not forwarded, send a Traverse message. - // NAT detection is the same algorithm as connection.IsBehindNAT. - if closePeer.PortReportedExternal == 0 && closePeer.Port != closePeer.PortReportedInternal { - // TODO: PortExternal needs to be guaranteed. send() needs to be broken up. - if raw, err := createVirtualAnnouncement(msg.MessageRaw.connection.Network, closePeer.PublicKey, &bootstrapFindSelf{}); err == nil { - peer.sendTraverse(raw, closePeer.PublicKey) - } - } } } } diff --git a/Command Traverse.go b/Command Traverse.go index a081d9a..fdc153a 100644 --- a/Command Traverse.go +++ b/Command Traverse.go @@ -108,14 +108,14 @@ func (peer *PeerInfo) cmdTraverseReceive(msg *MessageTraverse) { //peer.sendResponse(announce.Sequence, true, nil, nil, hashesNotFound) //packets, err := msgEncodeResponse(true, nil, nil, hashesNotFound) //sendAllNetworks() - var addresses []*net.UDPAddr if !msg.IPv4.IsUnspecified() { addressOriginalSenderIPv4 := &net.UDPAddr{IP: msg.IPv4, Port: int(msg.PortIPv4)} if msg.PortIPv4ReportedExternal > 0 { addressOriginalSenderIPv4.Port = int(msg.PortIPv4ReportedExternal) } - addresses = append(addresses, addressOriginalSenderIPv4) + + contactArbitraryPeer(msg.SignerPublicKey, addressOriginalSenderIPv4, 0) } if !msg.IPv6.IsUnspecified() { @@ -124,11 +124,8 @@ func (peer *PeerInfo) cmdTraverseReceive(msg *MessageTraverse) { addressOriginalSenderIPv6.Port = int(msg.PortIPv6ReportedExternal) } - addresses = append(addresses, addressOriginalSenderIPv6) + contactArbitraryPeer(msg.SignerPublicKey, addressOriginalSenderIPv6, 0) } - - // for now send a packet which should open up the NAT and establish a connection - contactArbitraryPeer(msg.SignerPublicKey, addresses) } // createVirtualAnnouncement is temporary code and will be improved. @@ -139,7 +136,7 @@ func createVirtualAnnouncement(network *Network, receiverPublicKey *btcec.Public } packet := &PacketRaw{Command: CommandAnnouncement, Payload: packets[0].raw} - setAnnouncementPorts(packet, network) + packet.setSelfReportedPorts(network) packet.Sequence = msgArbitrarySequence(receiverPublicKey, sequenceData).sequence packet.Protocol = ProtocolVersion diff --git a/Connection.go b/Connection.go index 13b4d48..351a348 100644 --- a/Connection.go +++ b/Connection.go @@ -7,7 +7,6 @@ Author: Peter Kleissner package core import ( - "encoding/binary" "errors" "net" "sync/atomic" @@ -283,30 +282,32 @@ func (peer *PeerInfo) IsBehindNAT() (result bool) { // ---- sending code ---- -// setAnnouncementPorts sets the fields Internal Port and External Port according to the connection details. -// This is important for the remote peer to make smart decisions whether this peer is behind a NAT/firewall and supports port forwarding/UPnP. -func setAnnouncementPorts(packet *PacketRaw, n *Network) { - if packet.Command != CommandAnnouncement && packet.Command != CommandResponse { // only for Announcement and Response messages - return +// send sends the packet to the peer on the connection +func (c *Connection) send(packet *PacketRaw, receiverPublicKey *btcec.PublicKey, isFirstPacket bool) (err error) { + if c == nil { + return errors.New("invalid connection") } - // The internal port is set to where the network listens on. - // Datacenter: This should usually be the same as the outgoing port. - // NAT: The internal port will be different than the outgoing one. - portI := uint16(n.address.Port) + packet.Protocol = ProtocolVersion + packet.setSelfReportedPorts(c.Network) - // External port: This is usually unknown, except in these 2 cases: - // UPnP: The port is forwarded automatically. - // Manual override in config: The user can specify a (global) incoming port that must be open on all listening IPs. - // This external port will be then passed onto other peers who will use it to connect. - portE := n.portExternal - - if config.PortForward > 0 { - portE = config.PortForward + raw, err := PacketEncrypt(peerPrivateKey, receiverPublicKey, packet) + if err != nil { + return err } - binary.LittleEndian.PutUint16(packet.Payload[15:17], portI) - binary.LittleEndian.PutUint16(packet.Payload[17:19], portE) + c.LastPacketOut = time.Now() + + 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) + } + } + + return err } // send sends a raw packet to the peer. Only uses active connections. @@ -315,8 +316,6 @@ func (peer *PeerInfo) send(packet *PacketRaw) (err error) { return errors.New("no valid connection to peer") } - packet.Protocol = ProtocolVersion - // For Traverse: check if no packet has been sent, and none received (i.e. initial contact). // If a packet was already received directly (note: not via incoming traversed message), a valid connection is already established. isFirstPacketOut := atomic.LoadUint64(&peer.StatsPacketSent) == 0 && atomic.LoadUint64(&peer.StatsPacketReceived) == 0 @@ -327,56 +326,28 @@ func (peer *PeerInfo) send(packet *PacketRaw) (err error) { // Send out the wire. Use connectionLatest if available. // Failover: If sending fails and there are other connections available, try those. Automatically update connectionLatest if one is successful. // Windows: This works great in case the adapter gets disabled, however, does not detect if the network cable is unplugged. - c := peer.connectionLatest - if c != nil { - setAnnouncementPorts(packet, c.Network) - - raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet) - if err != nil { - return err - } - - c.LastPacketOut = time.Now() - - if err = c.Network.send(c.Address.IP, c.Address.Port, raw); err == nil { - // Send Traverse message if the peer is behind NAT and this is the first message. - if isFirstPacketOut && c.IsBehindNAT() && c.traversePeer != nil { - if raw, err := createVirtualAnnouncement(c.Network, peer.PublicKey, &bootstrapFindSelf{}); err == nil { - c.traversePeer.sendTraverse(raw, peer.PublicKey) - } - } - + cLatest := peer.connectionLatest + if cLatest != nil { + if err := cLatest.send(packet, peer.PublicKey, isFirstPacketOut); err == nil { return nil - } - - // Invalid connection, immediately invalidate. Fallback to broadcast to all other active ones. - // Windows: A common error when the network adapter is disabled is "wsasendto: The requested address is not valid in its context". - if IsNetworkErrorFatal(err) { - peer.invalidateActiveConnection(c) + } else if IsNetworkErrorFatal(err) { + // Invalid connection, immediately invalidate. Fallback to broadcast to all other active ones. + // Windows: A common error when the network adapter is disabled is "wsasendto: The requested address is not valid in its context". + peer.invalidateActiveConnection(cLatest) } } - // If no latest connection available, broadcast on all available connections. + // If no latest connection available, broadcast on all other available connections. // This might be noisy, but if no latest connection is available it means the last established connection is already considered dead. // The receiver is responsible for incoming deduplication of packets. activeConnections := peer.GetConnections(true) for _, c := range activeConnections { - setAnnouncementPorts(packet, c.Network) - - raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet) - if err != nil { - return err + if c == cLatest { + continue } - c.LastPacketOut = time.Now() - - if err = c.Network.send(c.Address.IP, c.Address.Port, raw); err == nil { - // Send Traverse message if the peer is behind NAT and this is the first message. - if isFirstPacketOut && c.IsBehindNAT() && c.traversePeer != nil { - if raw, err := createVirtualAnnouncement(c.Network, peer.PublicKey, &bootstrapFindSelf{}); err == nil { - c.traversePeer.sendTraverse(raw, peer.PublicKey) - } - } + if err := c.send(packet, peer.PublicKey, isFirstPacketOut); err != nil && IsNetworkErrorFatal(err) { + peer.invalidateActiveConnection(c) } } @@ -385,23 +356,15 @@ func (peer *PeerInfo) send(packet *PacketRaw) (err error) { // sendConnection sends a packet to the peer using the specific connection func (peer *PeerInfo) sendConnection(packet *PacketRaw, connection *Connection) (err error) { - packet.Protocol = ProtocolVersion - raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet) - if err != nil { - return err - } - + isFirstPacketOut := atomic.LoadUint64(&peer.StatsPacketSent) == 0 && atomic.LoadUint64(&peer.StatsPacketReceived) == 0 atomic.AddUint64(&peer.StatsPacketSent, 1) - connection.LastPacketOut = time.Now() - return connection.Network.send(connection.Address.IP, connection.Address.Port, raw) + return connection.send(packet, peer.PublicKey, isFirstPacketOut) } // sendAllNetworks sends a raw packet via all networks. It assigns a new sequence for each sent packet. -func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remote *net.UDPAddr, sequenceData interface{}) (err error) { - var raw []byte - packet.Protocol = ProtocolVersion - +// 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 networksMutex.RLock() @@ -418,14 +381,9 @@ func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remo continue } - setAnnouncementPorts(packet, network) - packet.Sequence = msgArbitrarySequence(receiverPublicKey, sequenceData).sequence - if raw, err = PacketEncrypt(peerPrivateKey, receiverPublicKey, packet); err != nil { - return err - } + err = (&Connection{Network: network, Address: remote, PortInternal: receiverPortInternal}).send(packet, receiverPublicKey, true) - err = network.send(remote.IP, remote.Port, raw) if err == nil { successCount++ } diff --git a/Message Encoding.go b/Message Encoding.go index afd690b..b50780d 100644 --- a/Message Encoding.go +++ b/Message Encoding.go @@ -799,6 +799,19 @@ createPacketLoop: } } +// setSelfReportedPorts sets the fields Internal Port and External Port according to the connection details. +// This is important for the remote peer to make smart decisions whether this peer is behind a NAT/firewall and supports port forwarding/UPnP. +func (packet *PacketRaw) setSelfReportedPorts(n *Network) { + if packet.Command != CommandAnnouncement && packet.Command != CommandResponse { // only for Announcement and Response messages + return + } + + portI, portE := n.SelfReportedPorts() + + binary.LittleEndian.PutUint16(packet.Payload[15:17], portI) + binary.LittleEndian.PutUint16(packet.Payload[17:19], portE) +} + // ---- Traverse ---- const traversePayloadHeaderSize = 76 + 65 + 28 diff --git a/Network.go b/Network.go index a94ed08..9ee19d7 100644 --- a/Network.go +++ b/Network.go @@ -301,3 +301,23 @@ func (network *Network) Terminate() { removeListenAddress(network.address) } + +// SelfReportedPorts returns the internal and external ports as self-reported by the peer to others. +func (network *Network) SelfReportedPorts() (portI, portE uint16) { + // The internal port is set to where the network listens on. + // Datacenter: This should usually be the same as the outgoing port. + // NAT: The internal port will be different than the outgoing one. + portI = uint16(network.address.Port) + + // External port: This is usually unknown, except in these 2 cases: + // UPnP: The port is forwarded automatically. + // Manual override in config: The user can specify a (global) incoming port that must be open on all listening IPs. + // This external port will be then passed onto other peers who will use it to connect. + portE = network.portExternal + + if config.PortForward > 0 { + portE = config.PortForward + } + + return portI, portE +}