core/Connection.go

/*
File Username:  Connection.go
Copyright:  2021 Peernet s.r.o.
Author:     Peter Kleissner
*/

package core

import (
	"errors"
	"net"
	"sync/atomic"
	"time"

	"github.com/PeernetOfficial/core/btcec"
	"github.com/PeernetOfficial/core/protocol"
)

// 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 remote peer.
	PortInternal  uint16        // Internal listening port reported by remote peer. 0 if no Announcement/Response message was yet received.
	PortExternal  uint16        // External listening port reported by remote peer. 0 if not known by the peer.
	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.
	Expires       time.Time     // Inactive connections only: Expiry date. If it does not become active by that date, it will be considered expired and removed.
	Status        int           // 0 = Active established connection, 1 = Inactive, 2 = Removed, 3 = Redundant
	RoundTripTime time.Duration // Full round-trip time of last reply.
	Firewall      bool          // Whether the remote peer indicates a potential firewall. This means a Traverse message shall be sent to establish a connection.
	traversePeer  *PeerInfo     // Temporary peer that may act as proxy for a Traverse message used for the first packet. This is used to establish this Connection to a peer that is behind a NAT or firewall.
	backend       *Backend
}

// Connection status
const (
	ConnectionActive = iota
	ConnectionInactive
	ConnectionRemoved
	ConnectionRedundant // Same as active. Incoming packets are accepted. Outgoing use only for redundancy. Reduces ping overhead.
)

// 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)
}

// IsLocal checks if the connection is a local network one (LAN)
func (c *Connection) IsLocal() bool {
	return c != nil && IsIPLocal(c.Address.IP)
}

// IsIPv4 checks if the connection is using IPv4
func (c *Connection) IsIPv4() bool {
	return IsIPv4(c.Address.IP)
}

// IsIPv6 checks if the connection is using IPv6
func (c *Connection) IsIPv6() bool {
	return IsIPv6(c.Address.IP)
}

// IsBehindNAT checks if the remote peer on the connection is likely behind a NAT
func (c *Connection) IsBehindNAT() bool {
	return c.PortInternal > 0 && c.Address.Port != int(c.PortInternal)
}

// IsPortForward checks if the remote peer uses port forwarding on the connection
func (c *Connection) IsPortForward() bool {
	return c.PortExternal > 0
}

// IsVirtual returns true if the peer has not been connected yet. This is the case if another peer responds with peer details, and that peer shall be contacted.
func (peer *PeerInfo) IsVirtual() bool {
	return peer.isVirtual
}

// 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
}

// IsConnectionActive checks if the peer has an active connection that can be used to send and receive messages
func (peer *PeerInfo) IsConnectionActive() bool {
	peer.RLock()
	defer peer.RUnlock()

	return len(peer.connectionActive) > 0
}

// IsConnectable checks if the peer is connectable to the given IP parameters.
func (peer *PeerInfo) IsConnectable(allowLocal, allowIPv4, allowIPv6 bool) bool {
	peer.RLock()
	defer peer.RUnlock()

	// Only 1 active connection must be allowed for being connectable.
	for _, connection := range peer.connectionActive {
		// If the internal port is not known, which happens if no Announcement or Response was returned, do not share the peer details.
		// This can happen if only other messages such as Ping/Pong were received, or the protocol implementation is not compatible. The external port is also likely not available.
		// In this case sharing the peer would be bad, since the receiving peer could not use internal/external port to detemine the NAT status and port forwarding.
		if connection.PortInternal == 0 {
			continue
		}

		if IsIPv4(connection.Address.IP) && allowIPv4 || IsIPv6(connection.Address.IP) && allowIPv6 {
			if !(!allowLocal && connection.IsLocal()) {
				return true
			}
		}
	}

	return false
}

// GetConnection2Share returns a connection to share. Nil if none.
// allowLocal specifies whether it is OK to return local IPs.
func (peer *PeerInfo) GetConnection2Share(allowLocal, allowIPv4, allowIPv6 bool) (connection *Connection) {
	if !allowLocal && !allowIPv4 && !allowIPv6 {
		return nil
	}

	peer.RLock()
	defer peer.RUnlock()

	if peer.connectionLatest != nil && !(!allowLocal && peer.connectionLatest.IsLocal()) &&
		(IsIPv4(peer.connectionLatest.Address.IP) && allowIPv4 || IsIPv6(peer.connectionLatest.Address.IP) && allowIPv6) && peer.connectionLatest.PortInternal > 0 {
		return peer.connectionLatest
	}

	for _, connection := range peer.connectionActive {
		if (IsIPv4(connection.Address.IP) && allowIPv4 || IsIPv6(connection.Address.IP) && allowIPv6) && !(!allowLocal && connection.IsLocal()) && connection.PortInternal > 0 {
			return connection
		}
	}

	return nil
}

// 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 to the same IP
	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
			}

			connection.Status = ConnectionActive
			peer.setConnectionLatest(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
			connection.Status = ConnectionActive
			peer.connectionActive = append(peer.connectionActive, connection)
			peer.setConnectionLatest(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.setConnectionLatest(incoming)

	peer.Backend.Filters.NewPeerConnection(peer, incoming)

	return incoming
}

// setConnectionLatest updates the latest valid connection to use for sending. All other connections will be changed to redundant, which reduces ping overhead.
func (peer *PeerInfo) setConnectionLatest(latest *Connection) {
	if peer.connectionLatest == latest {
		return
	}

	peer.connectionLatest = latest

	for _, connection := range peer.connectionActive {
		if connection == latest {
			continue
		}
		connection.Status = ConnectionRedundant
	}
}

// invalidateActiveConnection invalidates an active connection
func (peer *PeerInfo) invalidateActiveConnection(input *Connection) {
	peer.Lock()
	defer peer.Unlock()

	// Change the status to inactive and start the expiration. If the connection does not become valid by that date, it will be removed.
	input.Status = ConnectionInactive
	input.Expires = time.Now().Add(connectionRemove * time.Second)

	// 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

			break
		}
	}
}

// removeInactiveConnection removes an inactive connection.
func (peer *PeerInfo) removeInactiveConnection(input *Connection) {
	peer.Lock()
	defer peer.Unlock()

	input.Status = ConnectionRemoved

	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
		}
	}
}

// GetRTT returns the round-trip time for the most recent active connection. 0 if not available.
func (peer *PeerInfo) GetRTT() (rtt time.Duration) {
	peer.Lock()
	defer peer.Unlock()

	if peer.connectionLatest != nil && peer.connectionLatest.RoundTripTime > 0 {
		return peer.connectionLatest.RoundTripTime
	}

	for _, connection := range peer.connectionActive {
		if connection.RoundTripTime > 0 {
			return connection.RoundTripTime
		}
	}

	return 0
}

// IsBehindNAT checks if the peer is behind NAT
func (peer *PeerInfo) IsBehindNAT() (result bool) {
	peer.Lock()
	defer peer.Unlock()

	// Default is no. Only if a public network reports different connected port vs internal one, NAT is assumed.
	// This also assumes that all 3rd party clients bind their connection to the outgoing port.
	// PortInternal is 0 if no Announcement or Response message was received.

	for _, connection := range peer.connectionActive {
		if connection.IsBehindNAT() {
			return true
		}
	}

	for _, connection := range peer.connectionInactive {
		if connection.IsBehindNAT() {
			return true
		}
	}

	return false
}

// IsPortForward checks if the peer uses port forwarding
func (peer *PeerInfo) IsPortForward() (result bool) {
	peer.Lock()
	defer peer.Unlock()

	for _, connection := range peer.connectionActive {
		if connection.IsPortForward() {
			return true
		}
	}

	for _, connection := range peer.connectionInactive {
		if connection.IsPortForward() {
			return true
		}
	}

	return false
}

// IsFirewallReported checks if the peer reported to be behind a firewall
func (peer *PeerInfo) IsFirewallReported() (result bool) {
	return peer.Features&(1<<protocol.FeatureFirewall) > 0
}

// ---- sending code ----

// send sends the packet to the peer on the connection
// isFirstPacket indicates whether this is the first packet to an uncontacted peer.
func (c *Connection) send(packet *protocol.PacketRaw, receiverPublicKey *btcec.PublicKey, isFirstPacket bool) (err error) {
	if c == nil {
		return errors.New("invalid connection")
	}

	packet.Protocol = protocol.ProtocolVersion
	packet.SetSelfReportedPorts(c.Network.SelfReportedPorts())

	c.backend.Filters.PacketOut(packet, receiverPublicKey, c)

	raw, err := protocol.PacketEncrypt(c.backend.PeerPrivateKey, receiverPublicKey, packet)
	if err != nil {
		return err
	}

	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 or firewall and this is the first message. Only for Announcement.
	if err == nil && isFirstPacket && (c.IsBehindNAT() || c.Firewall) && c.traversePeer != nil && packet.Command == protocol.CommandAnnouncement {
		err = c.traversePeer.sendTraverse(packet, receiverPublicKey)
	}

	return err
}

// send sends a raw packet to the peer. Only uses active connections.
func (peer *PeerInfo) send(packet *protocol.PacketRaw) (err error) {
	if peer.isVirtual { // special case for peers that were not contacted before
		for _, address := range peer.targetAddresses {
			peer.Backend.networks.sendAllNetworks(peer.PublicKey, packet, &net.UDPAddr{IP: address.IP, Port: int(address.Port)}, address.PortInternal, peer.Features&(1<<protocol.FeatureFirewall) > 0, peer.traversePeer, nil)
		}
		return
	}
	if len(peer.connectionActive) == 0 {
		return errors.New("no valid connection to peer")
	}

	// 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

	// always count as one sent packet even if sent via broadcast
	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.
	cLatest := peer.connectionLatest
	if cLatest != nil {
		if err := cLatest.send(packet, peer.PublicKey, isFirstPacketOut); err == nil {
			return nil
		} 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 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 {
		if c == cLatest {
			continue
		}

		if err := c.send(packet, peer.PublicKey, isFirstPacketOut); err != nil && IsNetworkErrorFatal(err) {
			peer.invalidateActiveConnection(c)
		}
	}

	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 *protocol.PacketRaw, connection *Connection) (err error) {
	isFirstPacketOut := atomic.LoadUint64(&peer.StatsPacketSent) == 0 && atomic.LoadUint64(&peer.StatsPacketReceived) == 0
	atomic.AddUint64(&peer.StatsPacketSent, 1)

	return connection.send(packet, peer.PublicKey, isFirstPacketOut)
}

// 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. Firewall indicates whether the remote peer was reported to be behind a firewall.
func (nets *Networks) sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *protocol.PacketRaw, remote *net.UDPAddr, receiverPortInternal uint16, receiverFirewall bool, traversePeer *PeerInfo, sequenceData interface{}) (err error) {
	nets.RLock()
	defer nets.RUnlock()

	networksTarget := nets.networks4
	if IsIPv6(remote.IP.To16()) {
		networksTarget = nets.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
		}

		if sequenceData != nil {
			packet.Sequence = nets.Sequences.ArbitrarySequence(receiverPublicKey, sequenceData).SequenceNumber
		}
		err = (&Connection{backend: nets.backend, Network: network, Address: remote, PortInternal: receiverPortInternal, traversePeer: traversePeer, Firewall: receiverFirewall}).send(packet, receiverPublicKey, isFirstPacket)
		isFirstPacket = false

		if err == nil {
			successCount++
		}
	}

	if successCount == 0 {
		return errors.New("no successful send")
	}

	return nil
}

// send sends a raw packet to the peer. Only uses active connections.
func (peer *PeerInfo) sendLite(raw []byte) (err error) {
	if peer.isVirtual { // special case for peers that were not contacted before
		return errors.New("cannot send lite packet to virtual peer")
	} else if len(peer.connectionActive) == 0 {
		return errors.New("no valid connection to peer")
	} else if atomic.LoadUint64(&peer.StatsPacketSent) == 0 && atomic.LoadUint64(&peer.StatsPacketReceived) == 0 {
		return errors.New("uncontacted peer") // A valid connection must have been established.
	}

	// always count as one sent packet even if sent via broadcast
	atomic.AddUint64(&peer.StatsPacketSent, 1)

	// Send out the wire. Use connectionLatest if available.
	cLatest := peer.connectionLatest
	if cLatest != nil {
		if err := cLatest.Network.send(cLatest.Address.IP, cLatest.Address.Port, raw); err == nil {
			return nil
		} 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 other available connections.
	for _, c := range peer.GetConnections(true) {
		if c == cLatest {
			continue
		}

		if err := c.Network.send(c.Address.IP, c.Address.Port, raw); err != nil && IsNetworkErrorFatal(err) {
			peer.invalidateActiveConnection(c)
		}
	}

	return nil // on broadcast no error is known and returned
}