From b53ab7c7b325aeb7e9792dbdd15dd9af1799b9e9 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Sun, 28 Feb 2021 20:59:39 +0100 Subject: [PATCH] Reworked connection code. Pings are now regularly sent out to every connection to check if it remains valid. Invalid connections are moved to a second list, and eventually dropped if they don't become active. --- Commands.go | 115 +++++++++++++++++++----- Connection.go | 239 ++++++++++++++++++++++++++++++++++++++++++++++++++ Network.go | 34 +++++-- Peer ID.go | 134 ++-------------------------- Peernet.go | 1 + 5 files changed, 365 insertions(+), 158 deletions(-) create mode 100644 Connection.go diff --git a/Commands.go b/Commands.go index 13084d1..1589696 100644 --- a/Commands.go +++ b/Commands.go @@ -8,66 +8,135 @@ package core import ( "fmt" - "net" + "time" "github.com/btcsuite/btcd/btcec" ) +// Commands between peers +const ( + // Peer List Management + CommandAnnouncement = 0 // Announcement + CommandResponse = 1 // Response + CommandPing = 2 // Keep-alive message (no payload). + CommandPong = 3 // Response to ping (no payload). + + // Blockchain + CommandGet = 4 // Request blocks for specified peer. + + // File Discovery + + // Debug + CommandChat = 10 // Chat message [debug] +) + // packet2 is a high-level message between peers type packet2 struct { PacketRaw SenderPublicKey *btcec.PublicKey // Sender Public Key, ECDSA (secp256k1) 257-bit - network *Network // network which received the packet - address *net.UDPAddr // address of the sender or receiver + connection *Connection // Connection that received the packet } -// announcement handles an incoming announcement -func (peer *PeerInfo) announcement(msg *packet2) { +// cmdAnouncement handles an incoming announcement +func (peer *PeerInfo) cmdAnouncement(msg *packet2) { if peer == nil { - peer, added := PeerlistAdd(msg.SenderPublicKey, &Connection{Network: msg.network, Address: msg.address}) - fmt.Printf("Incoming initial announcement from %s\n", msg.address.String()) + peer, added := PeerlistAdd(msg.SenderPublicKey, msg.connection) + fmt.Printf("Incoming initial announcement from %s\n", msg.connection.Address.String()) // send the Response if added { - peer.send(&PacketRaw{Command: 1}) + peer.send(&PacketRaw{Command: CommandResponse}) } return } - fmt.Printf("Incoming secondary announcement from %s\n", msg.address.String()) + fmt.Printf("Incoming secondary announcement from %s\n", msg.connection.Address.String()) // Announcement from existing peer means the peer most likely restarted - peer.send(&PacketRaw{Command: 1}) + peer.send(&PacketRaw{Command: CommandResponse}) } -// response handles the response to the announcement -func (peer *PeerInfo) response(msg *packet2) { +// cmdResponse handles the response to the announcement +func (peer *PeerInfo) cmdResponse(msg *packet2) { if peer == nil { - peer, _ = PeerlistAdd(msg.SenderPublicKey, &Connection{Network: msg.network, Address: msg.address}) - fmt.Printf("Incoming initial response from %s\n", msg.address.String()) + peer, _ = PeerlistAdd(msg.SenderPublicKey, msg.connection) + fmt.Printf("Incoming initial response from %s\n", msg.connection.Address.String()) return } - fmt.Printf("Incoming response from %s on %s\n", msg.address.String(), msg.network.address.String()) + fmt.Printf("Incoming response from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String()) } -// chat handles a chat message [debug] -func (peer *PeerInfo) chat(msg *packet2) { - fmt.Printf("Chat from '%s': %s\n", msg.address.String(), string(msg.PacketRaw.Payload)) +// cmdPing handles an incoming ping message +func (peer *PeerInfo) cmdPing(msg *packet2) { + peer.send(&PacketRaw{Command: CommandPong}) + //fmt.Printf("Incoming ping from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String()) } -// autoPingAll automatically pings all peers. This has multiple important reasons: -// * Keeping the UDP port open at NAT, if applicable -// * Making sure to recognize invalid connections and drop them. +// cmdPong handles an incoming pong message +func (peer *PeerInfo) cmdPong(msg *packet2) { + //fmt.Printf("Incoming pong from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String()) +} + +// cmdChat handles a chat message [debug] +func (peer *PeerInfo) cmdChat(msg *packet2) { + fmt.Printf("Chat from '%s': %s\n", msg.connection.Address.String(), string(msg.PacketRaw.Payload)) +} + +// pingTime is the time in seconds to send out ping messages +const pingTime = 10 + +// connectionInvalidate is the threshold in seconds to invalidate formerly active connections that no longer receive incoming packets. +const connectionInvalidate = 20 + +// connectionRemove is the threshold in seconds to remove inactive connections in case there is at least one active connection known. +const connectionRemove = 2 * 60 + +// autoPingAll sends out regular ping messages to all connections of all peers. This allows to detect invalid connections and eventually drop them. func autoPingAll() { + for { + time.Sleep(time.Second) + thresholdInvalidate := time.Now().Add(-connectionInvalidate * time.Second) + thresholdRemove := time.Now().Add(-(connectionRemove + connectionInvalidate) * time.Second) + thresholdPingOut := time.Now().Add(-pingTime * time.Second) - // TODO: If peer was previously unresponsive for X times, start pinging on all available ports. + for _, peer := range PeerlistGet() { + // first handle active connections + for _, connection := range peer.GetConnections(true) { + // Check if no incoming packet for the last X seconds. Regularly sent pings should result in incoming packets. + if connection.LastPacketIn.Before(thresholdInvalidate) { + peer.invalidateActiveConnection(connection) + } + + // Send ping if none was sent recently and no incoming packet was received recently. + if connection.LastPacketIn.Before(thresholdPingOut) && connection.LastPingOut.Before(thresholdPingOut) { + peer.sendConnection(&PacketRaw{Command: CommandPing}, connection) + connection.LastPingOut = time.Now() + } + } + + // handle inactive connections + for _, connection := range peer.GetConnections(false) { + // Remove connections that have been inactive for a long time; only if there is at least an active connection, or at least two other inactive ones. + if connection.LastPacketIn.Before(thresholdRemove) && (len(peer.connectionActive) >= 1 || len(peer.connectionInactive) > 2) { + peer.removeInactiveConnection(connection) + continue + } + + // if no ping was sent recently, send one now + if connection.LastPingOut.Before(thresholdPingOut) { + peer.sendConnection(&PacketRaw{Command: CommandPing}, connection) + connection.LastPingOut = time.Now() + } + } + } + } } // SendChatAll sends a text message to all peers func SendChatAll(text string) { for _, peer := range PeerlistGet() { - peer.send(&PacketRaw{Command: 10, Payload: []byte(text)}) + peer.send(&PacketRaw{Command: CommandChat, Payload: []byte(text)}) } } diff --git a/Connection.go b/Connection.go new file mode 100644 index 0000000..b25541f --- /dev/null +++ b/Connection.go @@ -0,0 +1,239 @@ +/* +File Name: Connection.go +Copyright: 2021 Peernet Foundation s.r.o. +Author: Peter Kleissner +*/ + +package core + +import ( + "errors" + "net" + "sync/atomic" + "time" + + "github.com/btcsuite/btcd/btcec" +) + +// Connection is an established connection between a remote IP address and a local network adapter. +// New connections may only be created in case of successful INCOMING packets. +type Connection struct { + Network *Network // network which received the packet + Address *net.UDPAddr // address of the sender or receiver + LastPacketIn time.Time // Last time an incoming packet was received. + LastPacketOut time.Time // Last time an outgoing packet was attempted to send. + LastPingOut time.Time // Last ping out. +} + +// Equal checks if the connection was established other the same network adapter using the same IP address. Port is intentionally not checked. +func (c *Connection) Equal(other *Connection) bool { + return c.Address.IP.Equal(other.Address.IP) && c.Network.address.IP.Equal(other.Network.address.IP) +} + +// GetConnections returns the list of connections +func (peer *PeerInfo) GetConnections(active bool) (connections []*Connection) { + peer.RLock() + defer peer.RUnlock() + + if active { + return peer.connectionActive + } + return peer.connectionInactive +} + +// registerConnection registers an incoming connection for an existing peer. If new, it will add to the list. If previously inactive, it will elevate. +func (peer *PeerInfo) registerConnection(incoming *Connection) (result *Connection) { + peer.Lock() + defer peer.Unlock() + + // first check if already an active connection + for _, connection := range peer.connectionActive { + if connection.Equal(incoming) { + // Connection already established. Verify port and update if necessary. + // Some NATs may rotate ports. Some mobile phone providers even rotate IPs which is not detected here. + if connection.Address.Port != incoming.Address.Port { + connection.Address.Port = incoming.Address.Port + } + + peer.connectionLatest = connection + return connection + } + } + + // if an inactive connection, elevate it to activated one + for n, connection := range peer.connectionInactive { + if connection.Equal(incoming) { + if connection.Address.Port != incoming.Address.Port { + connection.Address.Port = incoming.Address.Port + } + + // elevate by adding to active and mark as latest active + peer.connectionLatest = connection + peer.connectionActive = append(peer.connectionActive, connection) + + // remove from inactive + inactiveNew := peer.connectionInactive[:n] + if n < len(peer.connectionInactive)-1 { + inactiveNew = append(inactiveNew, peer.connectionInactive[n+1:]...) + } + peer.connectionInactive = inactiveNew + + return connection + } + } + + // otherwise it is a new connection! + peer.connectionActive = append(peer.connectionActive, incoming) + peer.connectionLatest = incoming + + return incoming +} + +// invalidateActiveConnection invalidates an active connection +func (peer *PeerInfo) invalidateActiveConnection(input *Connection) { + peer.Lock() + defer peer.Unlock() + + // remove from connectionLatest if selected so it won't be used by standard send function + if peer.connectionLatest == input { + peer.connectionLatest = nil + } + + for n, connection := range peer.connectionActive { + if connection == input { + // add to list of inactive connections + peer.connectionInactive = append(peer.connectionInactive, connection) + + // remove from active + activeNew := peer.connectionActive[:n] + if n < len(peer.connectionActive)-1 { + activeNew = append(activeNew, peer.connectionActive[n+1:]...) + } + peer.connectionActive = activeNew + + return + } + } +} + +// removeInactiveConnection removes an inactive connection. +func (peer *PeerInfo) removeInactiveConnection(input *Connection) { + peer.Lock() + defer peer.Unlock() + + for n, connection := range peer.connectionInactive { + if connection == input { + + // remove from inactive + inactiveNew := peer.connectionInactive[:n] + if n < len(peer.connectionInactive)-1 { + inactiveNew = append(inactiveNew, peer.connectionInactive[n+1:]...) + } + peer.connectionInactive = inactiveNew + + return + } + } +} + +// ---- sending code ---- + +// send sends a raw packet to the peer. Only uses active connections. +func (peer *PeerInfo) send(packet *PacketRaw) (err error) { + if len(peer.connectionActive) == 0 { + return errors.New("no valid connection to peer") + } + + packet.Protocol = 0 + + raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet) + if err != nil { + return err + } + + atomic.AddUint64(&peer.StatsPacketSent, 1) + + // 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 { + c.LastPacketOut = time.Now() + + if err = c.Network.send(c.Address.IP, c.Address.Port, raw); 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". + peer.invalidateActiveConnection(c) + } + + // If no latest connection available, broadcast on all 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 { + c.LastPacketOut = time.Now() + c.Network.send(c.Address.IP, c.Address.Port, raw) + } + + return nil // on broadcast no error is known and returned +} + +// sendConnection sends a packet to the peer using the specific connection +func (peer *PeerInfo) sendConnection(packet *PacketRaw, connection *Connection) (err error) { + packet.Protocol = 0 + raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet) + if err != nil { + return err + } + + atomic.AddUint64(&peer.StatsPacketSent, 1) + connection.LastPacketOut = time.Now() + + return connection.Network.send(connection.Address.IP, connection.Address.Port, raw) +} + +// sendAllNetworks sends a raw packet via all networks +func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remote *net.UDPAddr) (err error) { + packet.Protocol = 0 + raw, err := PacketEncrypt(peerPrivateKey, receiverPublicKey, packet) + if err != nil { + return err + } + + successCount := 0 + + if IsIPv6(remote.IP.To16()) { + for _, network := range networks6 { + // 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 + } + + err = network.send(remote.IP, remote.Port, raw) + if err == nil { + successCount++ + } + } + } else { + for _, network := range networks4 { + // 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 + } + + err = network.send(remote.IP, remote.Port, raw) + if err == nil { + successCount++ + } + } + } + + if successCount == 0 { + return errors.New("no successful send") + } + + return nil +} diff --git a/Network.go b/Network.go index 46a392c..dec146d 100644 --- a/Network.go +++ b/Network.go @@ -129,25 +129,35 @@ func packetWorker(packets <-chan networkWire) { continue } + connection := &Connection{Network: packet.network, Address: packet.sender} + peer := PeerlistLookup(senderPublicKey) if peer != nil { // Existing peers: Update statistics and network address if new atomic.AddUint64(&peer.StatsPacketReceived, 1) - peer.registerConnection(packet.sender, packet.network) + connection = peer.registerConnection(connection) } + connection.LastPacketIn = time.Now() + // process the packet - message := &packet2{SenderPublicKey: senderPublicKey, PacketRaw: *decoded, network: packet.network, address: packet.sender} + message := &packet2{SenderPublicKey: senderPublicKey, PacketRaw: *decoded, connection: connection} switch decoded.Command { - case 0: // Announce - peer.announcement(message) + case CommandAnnouncement: // Announce + peer.cmdAnouncement(message) - case 1: // Response - peer.response(message) + case CommandResponse: // Response + peer.cmdResponse(message) - case 10: // Chat [debug] - peer.chat(message) + case CommandPing: // Ping + peer.cmdPing(message) + + case CommandPong: // Ping + peer.cmdPong(message) + + case CommandChat: // Chat [debug] + peer.cmdChat(message) default: // Unknown command @@ -171,3 +181,11 @@ func GetNetworks(networkType int) (networks []*Network) { func (network *Network) GetListen() (listen *net.UDPAddr, multicastIPv6 net.IP, broadcastIPv4 []net.IP) { return network.address, network.multicastIP, network.broadcastIPv4 } + +// GetAdapterName returns the adapter name, if available +func (network *Network) GetAdapterName() string { + if network.iface != nil { + return network.iface.Name + } + return "[unknown adapter]" +} diff --git a/Peer ID.go b/Peer ID.go index e37433c..0a84542 100644 --- a/Peer ID.go +++ b/Peer ID.go @@ -8,12 +8,9 @@ package core import ( "encoding/hex" - "errors" "log" - "net" "os" "sync" - "sync/atomic" "github.com/btcsuite/btcd/btcec" ) @@ -66,18 +63,13 @@ func ExportPrivateKey() (privateKey *btcec.PrivateKey, publicKey *btcec.PublicKe return peerPrivateKey, peerPublicKey } -// Connection is an established connection between a remote IP address and a local network adapter. -// New connections may only be created in case of successful INCOMING packets. -type Connection struct { - Network *Network // network which received the packet - Address *net.UDPAddr // address of the sender or receiver -} - // PeerInfo stores information about a single remote peer type PeerInfo struct { - PublicKey *btcec.PublicKey // Public key - Connections []*Connection // list of established connections to the peer - connectionLatest *Connection // Latest valid connection. Allows to switch on the fly between connections; if one becomes invalid, try others. + PublicKey *btcec.PublicKey // Public key + connectionActive []*Connection // List of active established connections to the peer. + connectionInactive []*Connection // List of former connections that are no longer valid. They may be removed after a while. + connectionLatest *Connection // Latest valid connection. + sync.RWMutex // Mutex for access to list of connections. // statistics StatsPacketSent uint64 // Count of packets sent @@ -87,7 +79,7 @@ type PeerInfo struct { var peerList map[[btcec.PubKeyBytesLenCompressed]byte]*PeerInfo var peerlistMutex sync.RWMutex -// PeerlistAdd adds a new peer to the peer list. It does not validate the peer info. If the peer is already added, it does nothing. +// PeerlistAdd adds a new peer to the peer list. It does not validate the peer info. If the peer is already added, it does nothing. Connections must be live. func PeerlistAdd(PublicKey *btcec.PublicKey, connections ...*Connection) (peer *PeerInfo, added bool) { if len(connections) == 0 { return nil, false @@ -101,7 +93,7 @@ func PeerlistAdd(PublicKey *btcec.PublicKey, connections ...*Connection) (peer * return peer, false } - peer = &PeerInfo{PublicKey: PublicKey, Connections: connections, connectionLatest: connections[0]} + peer = &PeerInfo{PublicKey: PublicKey, connectionActive: connections, connectionLatest: connections[0]} peerList[publicKey2Compressed(peer.PublicKey)] = peer return peer, true @@ -149,115 +141,3 @@ func publicKey2Compressed(publicKey *btcec.PublicKey) [btcec.PubKeyBytesLenCompr copy(key[:], publicKey.SerializeCompressed()) return key } - -// send sends a raw packet to the peer -func (peer *PeerInfo) send(packet *PacketRaw) (err error) { - if len(peer.Connections) == 0 { - return errors.New("no valid connection to peer") - } - - packet.Protocol = 0 - - raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet) - if err != nil { - return err - } - - atomic.AddUint64(&peer.StatsPacketSent, 1) - - // 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 on if the adapter is disabled, however, does not detect if the network cable is unplugged. - if peer.connectionLatest != nil { - useConnection := peer.connectionLatest - if err = useConnection.Network.send(useConnection.Address.IP, useConnection.Address.Port, raw); err != nil && len(peer.Connections) > 1 { - err = peer.sendFailover(useConnection, raw) - } - return err - } - - if err = peer.Connections[0].Network.send(peer.Connections[0].Address.IP, peer.Connections[0].Address.Port, raw); err != nil && len(peer.Connections) > 1 { - err = peer.sendFailover(peer.Connections[0], raw) - } - return err -} - -// sendFailover tries to send the packet over any other different (than the failed) connection. -// If send works, it will switch over connectionLatest. -// Windows: A common error when the network adapter is disabled is "wsasendto: The requested address is not valid in its context". -func (peer *PeerInfo) sendFailover(failed *Connection, raw []byte) (err error) { - for _, connection := range peer.Connections { - if connection != failed { - err = connection.Network.send(connection.Address.IP, connection.Address.Port, raw) - if err == nil { - peer.connectionLatest = connection - return nil - } - } - } - return err -} - -// sendAllNetworks sends a raw packet via all networks -func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remote *net.UDPAddr) (err error) { - packet.Protocol = 0 - raw, err := PacketEncrypt(peerPrivateKey, receiverPublicKey, packet) - if err != nil { - return err - } - - successCount := 0 - - if IsIPv6(remote.IP.To16()) { - for _, network := range networks6 { - // 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 - } - - err = network.send(remote.IP, remote.Port, raw) - if err == nil { - successCount++ - } - } - } else { - for _, network := range networks4 { - // 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 - } - - err = network.send(remote.IP, remote.Port, raw) - if err == nil { - successCount++ - } - } - } - - if successCount == 0 { - return errors.New("no successful send") - } - - return nil -} - -// registerConnection registers an incoming connection for an existing peer. If new, it will add to the list. -// Currently there is only a connectionLatest indicating the latest valid connection. In the future other priority such as last packet time, speed, local vs remote IP might be used. -func (peer *PeerInfo) registerConnection(remote *net.UDPAddr, network *Network) { - for n := range peer.Connections { - if peer.Connections[n].Address.IP.Equal(remote.IP) && peer.Connections[n].Network.address.IP.Equal(network.address.IP) { - // Connection already established. Verify port and update if necessary. - // Some NATs may rotate ports. Some mobile phone providers even rotate IPs which is not detected here. - if peer.Connections[n].Address.Port != remote.Port { - peer.Connections[n].Address.Port = remote.Port - } - - peer.connectionLatest = peer.Connections[n] - return - } - } - - peer.Connections = append(peer.Connections, &Connection{Address: remote, Network: network}) - - peer.connectionLatest = peer.Connections[len(peer.Connections)-1] -} diff --git a/Peernet.go b/Peernet.go index 2964417..13fc43c 100644 --- a/Peernet.go +++ b/Peernet.go @@ -19,4 +19,5 @@ func Init() { func Connect() { go bootstrap() go autoMulticastBroadcast() + go autoPingAll() }