mirror of
https://github.com/tailscale/tailscale.git
synced 2024-11-30 05:25:35 +00:00
4338db28f7
Since link-local addresses are definitionally more likely to be a direct (lower-latency, more reliable) connection than a non-link-local private address, give those a bit of a boost when selecting endpoints. Updates #8097 Signed-off-by: Andrew Dunham <andrew@du.nham.ca> Change-Id: I93fdeb07de55ba39ba5fcee0834b579ca05c2a4e
1852 lines
59 KiB
Go
1852 lines
59 KiB
Go
// Copyright (c) Tailscale Inc & AUTHORS
|
|
// SPDX-License-Identifier: BSD-3-Clause
|
|
|
|
package magicsock
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/binary"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"math/rand"
|
|
"net"
|
|
"net/netip"
|
|
"reflect"
|
|
"runtime"
|
|
"slices"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"golang.org/x/crypto/poly1305"
|
|
xmaps "golang.org/x/exp/maps"
|
|
"golang.org/x/net/ipv4"
|
|
"golang.org/x/net/ipv6"
|
|
"tailscale.com/disco"
|
|
"tailscale.com/ipn/ipnstate"
|
|
"tailscale.com/net/stun"
|
|
"tailscale.com/net/tstun"
|
|
"tailscale.com/tailcfg"
|
|
"tailscale.com/tstime/mono"
|
|
"tailscale.com/types/key"
|
|
"tailscale.com/types/logger"
|
|
"tailscale.com/util/mak"
|
|
"tailscale.com/util/ringbuffer"
|
|
)
|
|
|
|
var mtuProbePingSizesV4 []int
|
|
var mtuProbePingSizesV6 []int
|
|
|
|
func init() {
|
|
for _, m := range tstun.WireMTUsToProbe {
|
|
mtuProbePingSizesV4 = append(mtuProbePingSizesV4, pktLenToPingSize(m, false))
|
|
mtuProbePingSizesV6 = append(mtuProbePingSizesV6, pktLenToPingSize(m, true))
|
|
}
|
|
}
|
|
|
|
// endpoint is a wireguard/conn.Endpoint. In wireguard-go and kernel WireGuard
|
|
// there is only one endpoint for a peer, but in Tailscale we distribute a
|
|
// number of possible endpoints for a peer which would include the all the
|
|
// likely addresses at which a peer may be reachable. This endpoint type holds
|
|
// the information required that when wireguard-go wants to send to a
|
|
// particular peer (essentially represented by this endpoint type), the send
|
|
// function can use the currently best known Tailscale endpoint to send packets
|
|
// to the peer.
|
|
type endpoint struct {
|
|
// atomically accessed; declared first for alignment reasons
|
|
lastRecvWG mono.Time // last time there were incoming packets from this peer destined for wireguard-go (e.g. not disco)
|
|
lastRecvUDPAny mono.Time // last time there were incoming UDP packets from this peer of any kind
|
|
numStopAndResetAtomic int64
|
|
debugUpdates *ringbuffer.RingBuffer[EndpointChange]
|
|
|
|
// These fields are initialized once and never modified.
|
|
c *Conn
|
|
nodeID tailcfg.NodeID
|
|
publicKey key.NodePublic // peer public key (for WireGuard + DERP)
|
|
publicKeyHex string // cached output of publicKey.UntypedHexString
|
|
fakeWGAddr netip.AddrPort // the UDP address we tell wireguard-go we're using
|
|
nodeAddr netip.Addr // the node's first tailscale address; used for logging & wireguard rate-limiting (Issue 6686)
|
|
|
|
disco atomic.Pointer[endpointDisco] // if the peer supports disco, the key and short string
|
|
|
|
// mu protects all following fields.
|
|
mu sync.Mutex // Lock ordering: Conn.mu, then endpoint.mu
|
|
|
|
heartBeatTimer *time.Timer // nil when idle
|
|
lastSendExt mono.Time // last time there were outgoing packets sent to this peer from an external trigger (e.g. wireguard-go or disco pingCLI)
|
|
lastSendAny mono.Time // last time there were outgoing packets sent this peer from any trigger, internal or external to magicsock
|
|
lastFullPing mono.Time // last time we pinged all disco or wireguard only endpoints
|
|
derpAddr netip.AddrPort // fallback/bootstrap path, if non-zero (non-zero for well-behaved clients)
|
|
|
|
bestAddr addrQuality // best non-DERP path; zero if none; mutate via setBestAddrLocked()
|
|
bestAddrAt mono.Time // time best address re-confirmed
|
|
trustBestAddrUntil mono.Time // time when bestAddr expires
|
|
sentPing map[stun.TxID]sentPing
|
|
endpointState map[netip.AddrPort]*endpointState
|
|
isCallMeMaybeEP map[netip.AddrPort]bool
|
|
|
|
// The following fields are related to the new "silent disco"
|
|
// implementation that's a WIP as of 2022-10-20.
|
|
// See #540 for background.
|
|
heartbeatDisabled bool
|
|
probeUDPLifetime *probeUDPLifetime // UDP path lifetime probing; nil if disabled
|
|
|
|
expired bool // whether the node has expired
|
|
isWireguardOnly bool // whether the endpoint is WireGuard only
|
|
}
|
|
|
|
func (de *endpoint) setBestAddrLocked(v addrQuality) {
|
|
if v.AddrPort != de.bestAddr.AddrPort {
|
|
de.probeUDPLifetime.resetCycleEndpointLocked()
|
|
}
|
|
de.bestAddr = v
|
|
}
|
|
|
|
const (
|
|
// udpLifetimeProbeCliffSlack is how much slack to use relative to a
|
|
// ProbeUDPLifetimeConfig.Cliffs duration in order to account for RTT,
|
|
// scheduling jitter, buffers, etc. If the cliff is 10s, we attempt to probe
|
|
// after 10s - 2s (8s) amount of inactivity.
|
|
udpLifetimeProbeCliffSlack = time.Second * 2
|
|
// udpLifetimeProbeSchedulingTolerance is how much of a difference can be
|
|
// tolerated between a UDP lifetime probe scheduling target and when it
|
|
// actually fired. This must be some fraction of udpLifetimeProbeCliffSlack.
|
|
udpLifetimeProbeSchedulingTolerance = udpLifetimeProbeCliffSlack / 8
|
|
)
|
|
|
|
// probeUDPLifetime represents the configuration and state tied to probing UDP
|
|
// path lifetime. A probe "cycle" involves pinging the UDP path at various
|
|
// timeout cliffs, which are pre-defined durations of interest commonly used by
|
|
// NATs/firewalls as default stateful session timeout values. Cliffs are probed
|
|
// in ascending order. A "cycle" completes when all cliffs have received a pong,
|
|
// or when a ping times out. Cycles may extend across endpoint session lifetimes
|
|
// if they are disrupted by user traffic.
|
|
type probeUDPLifetime struct {
|
|
// All fields are guarded by endpoint.mu. probeUDPLifetime methods are for
|
|
// convenience.
|
|
|
|
// config holds the probing configuration.
|
|
config ProbeUDPLifetimeConfig
|
|
|
|
// timer is nil when idle. A non-nil timer indicates we intend to probe a
|
|
// timeout cliff in the future.
|
|
timer *time.Timer
|
|
|
|
// bestAddr contains the endpoint.bestAddr.AddrPort at the time a cycle was
|
|
// scheduled to start. A probing cycle is 1:1 with the current
|
|
// endpoint.bestAddr.AddrPort in the interest of simplicity. When
|
|
// endpoint.bestAddr.AddrPort changes, any active probing cycle will reset.
|
|
bestAddr netip.AddrPort
|
|
// cycleStartedAt contains the time at which the first cliff
|
|
// (ProbeUDPLifetimeConfig.Cliffs[0]) was pinged for the current/last cycle.
|
|
cycleStartedAt time.Time
|
|
// cycleActive is true if a probing cycle is active, otherwise false.
|
|
cycleActive bool
|
|
// currentCliff represents the index into ProbeUDPLifetimeConfig.Cliffs for
|
|
// the cliff that we are waiting to ping, or waiting on a pong/timeout.
|
|
currentCliff int
|
|
// lastTxID is the ID for the last ping that was sent.
|
|
lastTxID stun.TxID
|
|
}
|
|
|
|
func (p *probeUDPLifetime) currentCliffDurationEndpointLocked() time.Duration {
|
|
if p == nil {
|
|
return 0
|
|
}
|
|
return p.config.Cliffs[p.currentCliff]
|
|
}
|
|
|
|
// cycleCompleteMaxCliffEndpointLocked records the max cliff (as an index of
|
|
// ProbeUDPLifetimeConfig.Cliffs) a probing cycle reached, i.e. received a pong
|
|
// for. A value < 0 indicates no cliff was reached. It is a no-op if the active
|
|
// configuration does not equal defaultProbeUDPLifetimeConfig.
|
|
func (p *probeUDPLifetime) cycleCompleteMaxCliffEndpointLocked(cliff int) {
|
|
if !p.config.Equals(defaultProbeUDPLifetimeConfig) {
|
|
return
|
|
}
|
|
switch {
|
|
case cliff < 0:
|
|
metricUDPLifetimeCycleCompleteNoCliffReached.Add(1)
|
|
case cliff == 0:
|
|
metricUDPLifetimeCycleCompleteAt10sCliff.Add(1)
|
|
case cliff == 1:
|
|
metricUDPLifetimeCycleCompleteAt30sCliff.Add(1)
|
|
case cliff == 2:
|
|
metricUDPLifetimeCycleCompleteAt60sCliff.Add(1)
|
|
}
|
|
}
|
|
|
|
// resetCycleEndpointLocked resets the state contained in p to reflect an
|
|
// inactive cycle.
|
|
func (p *probeUDPLifetime) resetCycleEndpointLocked() {
|
|
if p == nil {
|
|
return
|
|
}
|
|
if p.timer != nil {
|
|
p.timer.Stop()
|
|
p.timer = nil
|
|
}
|
|
p.cycleActive = false
|
|
p.currentCliff = 0
|
|
p.bestAddr = netip.AddrPort{}
|
|
}
|
|
|
|
// ProbeUDPLifetimeConfig represents the configuration for probing UDP path
|
|
// lifetime.
|
|
type ProbeUDPLifetimeConfig struct {
|
|
// The timeout cliffs to probe. Values are in ascending order. Ascending
|
|
// order is chosen over descending because we have limited opportunities to
|
|
// probe. With a descending order we are stuck waiting for a new UDP
|
|
// path/session if the first value times out. When that new path is
|
|
// established is anyone's guess.
|
|
Cliffs []time.Duration
|
|
// CycleCanStartEvery represents the min duration between cycles starting
|
|
// up.
|
|
CycleCanStartEvery time.Duration
|
|
}
|
|
|
|
var (
|
|
// defaultProbeUDPLifetimeConfig is the configuration that must be used
|
|
// for UDP path lifetime probing until it can be wholly disseminated (not
|
|
// just on/off) from upstream control components, and associated metrics
|
|
// (metricUDPLifetime*) have lifetime management.
|
|
//
|
|
// TODO(#10928): support dynamic config via tailcfg.PeerCapMap.
|
|
defaultProbeUDPLifetimeConfig = &ProbeUDPLifetimeConfig{
|
|
Cliffs: []time.Duration{
|
|
time.Second * 10,
|
|
time.Second * 30,
|
|
time.Second * 60,
|
|
},
|
|
CycleCanStartEvery: time.Hour * 24,
|
|
}
|
|
)
|
|
|
|
// Equals returns true if b equals p, otherwise false. If both sides are nil,
|
|
// Equals returns true. If only one side is nil, Equals returns false.
|
|
func (p *ProbeUDPLifetimeConfig) Equals(b *ProbeUDPLifetimeConfig) bool {
|
|
if p == b {
|
|
return true
|
|
}
|
|
if (p == nil && b != nil) || (b == nil && p != nil) {
|
|
return false
|
|
}
|
|
if !slices.Equal(p.Cliffs, b.Cliffs) {
|
|
return false
|
|
}
|
|
if p.CycleCanStartEvery != b.CycleCanStartEvery {
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
// Valid returns true if p is valid, otherwise false. p must be non-nil.
|
|
func (p *ProbeUDPLifetimeConfig) Valid() bool {
|
|
if len(p.Cliffs) < 1 {
|
|
// We need at least one cliff, otherwise there is nothing to probe.
|
|
return false
|
|
}
|
|
if p.CycleCanStartEvery < 1 {
|
|
// Probing must be constrained by a positive CycleCanStartEvery.
|
|
return false
|
|
}
|
|
for i, c := range p.Cliffs {
|
|
if c <= max(udpLifetimeProbeCliffSlack*2, heartbeatInterval) {
|
|
// A timeout cliff less than or equal to twice
|
|
// udpLifetimeProbeCliffSlack is invalid due to being effectively
|
|
// zero when the cliff slack is subtracted from the cliff value at
|
|
// scheduling time.
|
|
//
|
|
// A timeout cliff less or equal to the heartbeatInterval is also
|
|
// invalid, as we may attempt to schedule on the tail end of the
|
|
// last heartbeat tied to an active session.
|
|
//
|
|
// These values are constants, but max()'d in case they change in
|
|
// the future.
|
|
return false
|
|
}
|
|
if i == 0 {
|
|
continue
|
|
}
|
|
if c <= p.Cliffs[i-1] {
|
|
// Cliffs must be in ascending order.
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// setProbeUDPLifetimeOn enables or disables probing of UDP path lifetime based
|
|
// on v. In the case of enablement defaultProbeUDPLifetimeConfig is used as the
|
|
// desired configuration.
|
|
func (de *endpoint) setProbeUDPLifetimeOn(v bool) {
|
|
de.mu.Lock()
|
|
if v {
|
|
de.setProbeUDPLifetimeConfigLocked(defaultProbeUDPLifetimeConfig)
|
|
} else {
|
|
de.setProbeUDPLifetimeConfigLocked(nil)
|
|
}
|
|
de.mu.Unlock()
|
|
}
|
|
|
|
// setProbeUDPLifetimeConfigLocked sets the desired configuration for probing
|
|
// UDP path lifetime. Ownership of desired is passed to endpoint, it must not be
|
|
// mutated once this call is made. A nil value disables the feature. If desired
|
|
// is non-nil but desired.Valid() returns false this is a no-op.
|
|
func (de *endpoint) setProbeUDPLifetimeConfigLocked(desired *ProbeUDPLifetimeConfig) {
|
|
if de.isWireguardOnly {
|
|
return
|
|
}
|
|
if desired == nil {
|
|
if de.probeUDPLifetime == nil {
|
|
// noop, not currently configured or desired
|
|
return
|
|
}
|
|
de.probeUDPLifetime.resetCycleEndpointLocked()
|
|
de.probeUDPLifetime = nil
|
|
return
|
|
}
|
|
if !desired.Valid() {
|
|
return
|
|
}
|
|
if de.probeUDPLifetime != nil {
|
|
if de.probeUDPLifetime.config.Equals(desired) {
|
|
// noop, current config equals desired
|
|
return
|
|
}
|
|
de.probeUDPLifetime.resetCycleEndpointLocked()
|
|
} else {
|
|
de.probeUDPLifetime = &probeUDPLifetime{}
|
|
}
|
|
p := de.probeUDPLifetime
|
|
p.config = *desired
|
|
p.resetCycleEndpointLocked()
|
|
}
|
|
|
|
// endpointDisco is the current disco key and short string for an endpoint. This
|
|
// structure is immutable.
|
|
type endpointDisco struct {
|
|
key key.DiscoPublic // for discovery messages.
|
|
short string // ShortString of discoKey.
|
|
}
|
|
|
|
type sentPing struct {
|
|
to netip.AddrPort
|
|
at mono.Time
|
|
timer *time.Timer // timeout timer
|
|
purpose discoPingPurpose
|
|
size int // size of the disco message
|
|
resCB *pingResultAndCallback // or nil for internal use
|
|
}
|
|
|
|
// endpointState is some state and history for a specific endpoint of
|
|
// a endpoint. (The subject is the endpoint.endpointState
|
|
// map key)
|
|
type endpointState struct {
|
|
// all fields guarded by endpoint.mu
|
|
|
|
// lastPing is the last (outgoing) ping time.
|
|
lastPing mono.Time
|
|
|
|
// lastGotPing, if non-zero, means that this was an endpoint
|
|
// that we learned about at runtime (from an incoming ping)
|
|
// and that is not in the network map. If so, we keep the time
|
|
// updated and use it to discard old candidates.
|
|
lastGotPing time.Time
|
|
|
|
// lastGotPingTxID contains the TxID for the last incoming ping. This is
|
|
// used to de-dup incoming pings that we may see on both the raw disco
|
|
// socket on Linux, and UDP socket. We cannot rely solely on the raw socket
|
|
// disco handling due to https://github.com/tailscale/tailscale/issues/7078.
|
|
lastGotPingTxID stun.TxID
|
|
|
|
// callMeMaybeTime, if non-zero, is the time this endpoint
|
|
// was advertised last via a call-me-maybe disco message.
|
|
callMeMaybeTime time.Time
|
|
|
|
recentPongs []pongReply // ring buffer up to pongHistoryCount entries
|
|
recentPong uint16 // index into recentPongs of most recent; older before, wrapped
|
|
|
|
index int16 // index in nodecfg.Node.Endpoints; meaningless if lastGotPing non-zero
|
|
}
|
|
|
|
// clear removes all derived / probed state from an endpointState.
|
|
func (s *endpointState) clear() {
|
|
*s = endpointState{
|
|
index: s.index,
|
|
lastGotPing: s.lastGotPing,
|
|
}
|
|
}
|
|
|
|
// pongHistoryCount is how many pongReply values we keep per endpointState
|
|
const pongHistoryCount = 64
|
|
|
|
type pongReply struct {
|
|
latency time.Duration
|
|
pongAt mono.Time // when we received the pong
|
|
from netip.AddrPort // the pong's src (usually same as endpoint map key)
|
|
pongSrc netip.AddrPort // what they reported they heard
|
|
}
|
|
|
|
// EndpointChange is a structure containing information about changes made to a
|
|
// particular endpoint. This is not a stable interface and could change at any
|
|
// time.
|
|
type EndpointChange struct {
|
|
When time.Time // when the change occurred
|
|
What string // what this change is
|
|
From any `json:",omitempty"` // information about the previous state
|
|
To any `json:",omitempty"` // information about the new state
|
|
}
|
|
|
|
// shouldDeleteLocked reports whether we should delete this endpoint.
|
|
func (st *endpointState) shouldDeleteLocked() bool {
|
|
switch {
|
|
case !st.callMeMaybeTime.IsZero():
|
|
return false
|
|
case st.lastGotPing.IsZero():
|
|
// This was an endpoint from the network map. Is it still in the network map?
|
|
return st.index == indexSentinelDeleted
|
|
default:
|
|
// This was an endpoint discovered at runtime.
|
|
return time.Since(st.lastGotPing) > sessionActiveTimeout
|
|
}
|
|
}
|
|
|
|
// latencyLocked returns the most recent latency measurement, if any.
|
|
// endpoint.mu must be held.
|
|
func (st *endpointState) latencyLocked() (lat time.Duration, ok bool) {
|
|
if len(st.recentPongs) == 0 {
|
|
return 0, false
|
|
}
|
|
return st.recentPongs[st.recentPong].latency, true
|
|
}
|
|
|
|
// endpoint.mu must be held.
|
|
func (st *endpointState) addPongReplyLocked(r pongReply) {
|
|
if n := len(st.recentPongs); n < pongHistoryCount {
|
|
st.recentPong = uint16(n)
|
|
st.recentPongs = append(st.recentPongs, r)
|
|
return
|
|
}
|
|
i := st.recentPong + 1
|
|
if i == pongHistoryCount {
|
|
i = 0
|
|
}
|
|
st.recentPongs[i] = r
|
|
st.recentPong = i
|
|
}
|
|
|
|
func (de *endpoint) deleteEndpointLocked(why string, ep netip.AddrPort) {
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "deleteEndpointLocked-" + why,
|
|
From: ep,
|
|
})
|
|
delete(de.endpointState, ep)
|
|
if de.bestAddr.AddrPort == ep {
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "deleteEndpointLocked-bestAddr-" + why,
|
|
From: de.bestAddr,
|
|
})
|
|
de.setBestAddrLocked(addrQuality{})
|
|
}
|
|
}
|
|
|
|
// initFakeUDPAddr populates fakeWGAddr with a globally unique fake UDPAddr.
|
|
// The current implementation just uses the pointer value of de jammed into an IPv6
|
|
// address, but it could also be, say, a counter.
|
|
func (de *endpoint) initFakeUDPAddr() {
|
|
var addr [16]byte
|
|
addr[0] = 0xfd
|
|
addr[1] = 0x00
|
|
binary.BigEndian.PutUint64(addr[2:], uint64(reflect.ValueOf(de).Pointer()))
|
|
de.fakeWGAddr = netip.AddrPortFrom(netip.AddrFrom16(addr).Unmap(), 12345)
|
|
}
|
|
|
|
// noteRecvActivity records receive activity on de, and invokes
|
|
// Conn.noteRecvActivity no more than once every 10s.
|
|
func (de *endpoint) noteRecvActivity(ipp netip.AddrPort, now mono.Time) {
|
|
if de.isWireguardOnly {
|
|
de.mu.Lock()
|
|
de.bestAddr.AddrPort = ipp
|
|
de.bestAddrAt = now
|
|
de.trustBestAddrUntil = now.Add(5 * time.Second)
|
|
de.mu.Unlock()
|
|
} else {
|
|
// TODO(jwhited): subject to change as part of silent disco effort.
|
|
// Necessary when heartbeat is disabled for the endpoint, otherwise we
|
|
// kick off discovery disco pings every trustUDPAddrDuration and mirror
|
|
// to DERP.
|
|
de.mu.Lock()
|
|
if de.heartbeatDisabled && de.bestAddr.AddrPort == ipp {
|
|
de.trustBestAddrUntil = now.Add(trustUDPAddrDuration)
|
|
}
|
|
de.mu.Unlock()
|
|
}
|
|
|
|
elapsed := now.Sub(de.lastRecvWG.LoadAtomic())
|
|
if elapsed > 10*time.Second {
|
|
de.lastRecvWG.StoreAtomic(now)
|
|
|
|
if de.c.noteRecvActivity == nil {
|
|
return
|
|
}
|
|
de.c.noteRecvActivity(de.publicKey)
|
|
}
|
|
}
|
|
|
|
func (de *endpoint) discoShort() string {
|
|
var short string
|
|
if d := de.disco.Load(); d != nil {
|
|
short = d.short
|
|
}
|
|
return short
|
|
}
|
|
|
|
// String exists purely so wireguard-go internals can log.Printf("%v")
|
|
// its internal conn.Endpoints and we don't end up with data races
|
|
// from fmt (via log) reading mutex fields and such.
|
|
func (de *endpoint) String() string {
|
|
return fmt.Sprintf("magicsock.endpoint{%v, %v}", de.publicKey.ShortString(), de.discoShort())
|
|
}
|
|
|
|
func (de *endpoint) ClearSrc() {}
|
|
func (de *endpoint) SrcToString() string { panic("unused") } // unused by wireguard-go
|
|
func (de *endpoint) SrcIP() netip.Addr { panic("unused") } // unused by wireguard-go
|
|
func (de *endpoint) DstToString() string { return de.publicKeyHex }
|
|
func (de *endpoint) DstIP() netip.Addr { return de.nodeAddr } // see tailscale/tailscale#6686
|
|
func (de *endpoint) DstToBytes() []byte { return packIPPort(de.fakeWGAddr) }
|
|
|
|
// addrForSendLocked returns the address(es) that should be used for
|
|
// sending the next packet. Zero, one, or both of UDP address and DERP
|
|
// addr may be non-zero. If the endpoint is WireGuard only and does not have
|
|
// latency information, a bool is returned to indicate that the
|
|
// WireGuard latency discovery pings should be sent.
|
|
//
|
|
// de.mu must be held.
|
|
//
|
|
// TODO(val): Rewrite the addrFor*Locked() variations to share code.
|
|
func (de *endpoint) addrForSendLocked(now mono.Time) (udpAddr, derpAddr netip.AddrPort, sendWGPing bool) {
|
|
udpAddr = de.bestAddr.AddrPort
|
|
|
|
if udpAddr.IsValid() && !now.After(de.trustBestAddrUntil) {
|
|
return udpAddr, netip.AddrPort{}, false
|
|
}
|
|
|
|
if de.isWireguardOnly {
|
|
// If the endpoint is wireguard-only, we don't have a DERP
|
|
// address to send to, so we have to send to the UDP address.
|
|
udpAddr, shouldPing := de.addrForWireGuardSendLocked(now)
|
|
return udpAddr, netip.AddrPort{}, shouldPing
|
|
}
|
|
|
|
// We had a bestAddr but it expired so send both to it
|
|
// and DERP.
|
|
return udpAddr, de.derpAddr, false
|
|
}
|
|
|
|
// addrForWireGuardSendLocked returns the address that should be used for
|
|
// sending the next packet. If a packet has never or not recently been sent to
|
|
// the endpoint, then a randomly selected address for the endpoint is returned,
|
|
// as well as a bool indiciating that WireGuard discovery pings should be started.
|
|
// If the addresses have latency information available, then the address with the
|
|
// best latency is used.
|
|
//
|
|
// de.mu must be held.
|
|
func (de *endpoint) addrForWireGuardSendLocked(now mono.Time) (udpAddr netip.AddrPort, shouldPing bool) {
|
|
if len(de.endpointState) == 0 {
|
|
de.c.logf("magicsock: addrForSendWireguardLocked: [unexpected] no candidates available for endpoint")
|
|
return udpAddr, false
|
|
}
|
|
|
|
// lowestLatency is a high duration initially, so we
|
|
// can be sure we're going to have a duration lower than this
|
|
// for the first latency retrieved.
|
|
lowestLatency := time.Hour
|
|
var oldestPing mono.Time
|
|
for ipp, state := range de.endpointState {
|
|
if oldestPing.IsZero() {
|
|
oldestPing = state.lastPing
|
|
} else if state.lastPing.Before(oldestPing) {
|
|
oldestPing = state.lastPing
|
|
}
|
|
|
|
if latency, ok := state.latencyLocked(); ok {
|
|
if latency < lowestLatency || latency == lowestLatency && ipp.Addr().Is6() {
|
|
// If we have the same latency,IPv6 is prioritized.
|
|
// TODO(catzkorn): Consider a small increase in latency to use
|
|
// IPv6 in comparison to IPv4, when possible.
|
|
lowestLatency = latency
|
|
udpAddr = ipp
|
|
}
|
|
}
|
|
}
|
|
needPing := len(de.endpointState) > 1 && now.Sub(oldestPing) > wireguardPingInterval
|
|
|
|
if !udpAddr.IsValid() {
|
|
candidates := xmaps.Keys(de.endpointState)
|
|
|
|
// Randomly select an address to use until we retrieve latency information
|
|
// and give it a short trustBestAddrUntil time so we avoid flapping between
|
|
// addresses while waiting on latency information to be populated.
|
|
udpAddr = candidates[rand.Intn(len(candidates))]
|
|
}
|
|
|
|
de.bestAddr.AddrPort = udpAddr
|
|
// Only extend trustBestAddrUntil by one second to avoid packet
|
|
// reordering and/or CPU usage from random selection during the first
|
|
// second. We should receive a response due to a WireGuard handshake in
|
|
// less than one second in good cases, in which case this will be then
|
|
// extended to 15 seconds.
|
|
de.trustBestAddrUntil = now.Add(time.Second)
|
|
return udpAddr, needPing
|
|
}
|
|
|
|
// addrForPingSizeLocked returns the address(es) that should be used for sending
|
|
// the next ping. It will only return addrs with a large enough path MTU to
|
|
// permit a ping payload of size bytes to be delivered (DERP is always one such
|
|
// addr as it is a TCP connection). If it returns a zero-value udpAddr, then we
|
|
// should continue probing the MTU of all paths to this endpoint. Zero, one, or
|
|
// both of the returned UDP address and DERP address may be non-zero.
|
|
//
|
|
// de.mu must be held.
|
|
func (de *endpoint) addrForPingSizeLocked(now mono.Time, size int) (udpAddr, derpAddr netip.AddrPort) {
|
|
if size == 0 {
|
|
udpAddr, derpAddr, _ = de.addrForSendLocked(now)
|
|
return
|
|
}
|
|
|
|
udpAddr = de.bestAddr.AddrPort
|
|
pathMTU := de.bestAddr.wireMTU
|
|
requestedMTU := pingSizeToPktLen(size, udpAddr.Addr().Is6())
|
|
mtuOk := requestedMTU <= pathMTU
|
|
|
|
if udpAddr.IsValid() && mtuOk {
|
|
if !now.After(de.trustBestAddrUntil) {
|
|
return udpAddr, netip.AddrPort{}
|
|
}
|
|
// We had a bestAddr with large enough MTU but it expired, so
|
|
// send both to it and DERP.
|
|
return udpAddr, de.derpAddr
|
|
}
|
|
|
|
// The UDP address isn't valid or it doesn't have a path MTU big enough
|
|
// for the packet. Return a zero-value udpAddr to signal that we should
|
|
// keep probing the path MTU to all addresses for this endpoint, and a
|
|
// valid DERP addr to signal that we should also send via DERP.
|
|
return netip.AddrPort{}, de.derpAddr
|
|
}
|
|
|
|
// maybeProbeUDPLifetimeLocked returns an afterInactivityFor duration and true
|
|
// if de is a candidate for UDP path lifetime probing in the future, otherwise
|
|
// false.
|
|
func (de *endpoint) maybeProbeUDPLifetimeLocked() (afterInactivityFor time.Duration, maybe bool) {
|
|
p := de.probeUDPLifetime
|
|
if p == nil {
|
|
return afterInactivityFor, false
|
|
}
|
|
if !de.bestAddr.IsValid() {
|
|
return afterInactivityFor, false
|
|
}
|
|
epDisco := de.disco.Load()
|
|
if epDisco == nil {
|
|
// peer does not support disco
|
|
return afterInactivityFor, false
|
|
}
|
|
// We compare disco keys, which may have a shorter lifetime than node keys
|
|
// since disco keys reset on startup. This has the desired side effect of
|
|
// shuffling probing probability where the local node ends up with a large
|
|
// key value lexicographically relative to the other nodes it tends to
|
|
// communicate with. If de's disco key changes, the cycle will reset.
|
|
if de.c.discoPublic.Compare(epDisco.key) >= 0 {
|
|
// lower disco pub key node probes higher
|
|
return afterInactivityFor, false
|
|
}
|
|
if !p.cycleActive && time.Since(p.cycleStartedAt) < p.config.CycleCanStartEvery {
|
|
// This is conservative as it doesn't account for afterInactivityFor use
|
|
// by the caller, potentially delaying the start of the next cycle. We
|
|
// assume the cycle could start immediately following
|
|
// maybeProbeUDPLifetimeLocked(), regardless of the value of
|
|
// afterInactivityFor relative to latest packets in/out time.
|
|
return afterInactivityFor, false
|
|
}
|
|
afterInactivityFor = p.currentCliffDurationEndpointLocked() - udpLifetimeProbeCliffSlack
|
|
if afterInactivityFor < 0 {
|
|
// shouldn't happen
|
|
return afterInactivityFor, false
|
|
}
|
|
return afterInactivityFor, true
|
|
}
|
|
|
|
// heartbeatForLifetimeVia represents the scheduling source of
|
|
// endpoint.heartbeatForLifetime().
|
|
type heartbeatForLifetimeVia string
|
|
|
|
const (
|
|
heartbeatForLifetimeViaSessionInactive heartbeatForLifetimeVia = "session-inactive"
|
|
heartbeatForLifetimeViaPongRx heartbeatForLifetimeVia = "pong-rx"
|
|
heartbeatForLifetimeViaSelf heartbeatForLifetimeVia = "self"
|
|
)
|
|
|
|
// scheduleHeartbeatForLifetimeLocked schedules de.heartbeatForLifetime to fire
|
|
// in the future (after). The caller must describe themselves in the via arg.
|
|
func (de *endpoint) scheduleHeartbeatForLifetimeLocked(after time.Duration, via heartbeatForLifetimeVia) {
|
|
p := de.probeUDPLifetime
|
|
if p == nil {
|
|
return
|
|
}
|
|
de.c.dlogf("[v1] magicsock: disco: scheduling UDP lifetime probe for cliff=%v via=%v to %v (%v)",
|
|
p.currentCliffDurationEndpointLocked(), via, de.publicKey.ShortString(), de.discoShort())
|
|
p.bestAddr = de.bestAddr.AddrPort
|
|
p.timer = time.AfterFunc(after, de.heartbeatForLifetime)
|
|
if via == heartbeatForLifetimeViaSelf {
|
|
metricUDPLifetimeCliffsRescheduled.Add(1)
|
|
} else {
|
|
metricUDPLifetimeCliffsScheduled.Add(1)
|
|
}
|
|
}
|
|
|
|
// heartbeatForLifetime sends a disco ping recorded locally with a purpose of
|
|
// pingHeartbeatForUDPLifetime to de if de.bestAddr has remained stable, and it
|
|
// has been inactive for a duration that is within the error bounds for current
|
|
// lifetime probing cliff. Alternatively it may reschedule itself into the
|
|
// future, which is one of three scheduling sources. The other scheduling
|
|
// sources are de.heartbeat() and de.probeUDPLifetimeCliffDoneLocked().
|
|
func (de *endpoint) heartbeatForLifetime() {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
p := de.probeUDPLifetime
|
|
if p == nil || p.timer == nil {
|
|
// We raced with a code path trying to p.timer.Stop() us. Give up early
|
|
// in the interest of simplicity. If p.timer.Stop() happened in
|
|
// de.heartbeat() presumably because of recent packets in/out we *could*
|
|
// still probe here, and it would be meaningful, but the time logic
|
|
// below would reschedule as-is.
|
|
return
|
|
}
|
|
p.timer = nil
|
|
if !p.bestAddr.IsValid() || de.bestAddr.AddrPort != p.bestAddr {
|
|
// best path changed
|
|
p.resetCycleEndpointLocked()
|
|
return
|
|
}
|
|
afterInactivityFor, ok := de.maybeProbeUDPLifetimeLocked()
|
|
if !ok {
|
|
p.resetCycleEndpointLocked()
|
|
return
|
|
}
|
|
inactiveFor := mono.Now().Sub(max(de.lastRecvUDPAny.LoadAtomic(), de.lastSendAny))
|
|
delta := afterInactivityFor - inactiveFor
|
|
if delta.Abs() > udpLifetimeProbeSchedulingTolerance {
|
|
if delta < 0 {
|
|
// We missed our opportunity. We can resume this cliff at the tail
|
|
// end of another session.
|
|
metricUDPLifetimeCliffsMissed.Add(1)
|
|
return
|
|
} else {
|
|
// We need to wait longer before sending a ping. This can happen for
|
|
// a number of reasons, which are described in more detail in
|
|
// de.heartbeat().
|
|
de.scheduleHeartbeatForLifetimeLocked(delta, heartbeatForLifetimeViaSelf)
|
|
return
|
|
}
|
|
}
|
|
if p.currentCliff == 0 {
|
|
p.cycleStartedAt = time.Now()
|
|
p.cycleActive = true
|
|
}
|
|
de.c.dlogf("[v1] magicsock: disco: sending disco ping for UDP lifetime probe cliff=%v to %v (%v)",
|
|
p.currentCliffDurationEndpointLocked(), de.publicKey.ShortString(), de.discoShort())
|
|
de.startDiscoPingLocked(de.bestAddr.AddrPort, mono.Now(), pingHeartbeatForUDPLifetime, 0, nil)
|
|
}
|
|
|
|
// heartbeat is called every heartbeatInterval to keep the best UDP path alive,
|
|
// kick off discovery of other paths, or schedule the probing of UDP path
|
|
// lifetime on the tail end of an active session.
|
|
func (de *endpoint) heartbeat() {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
if de.probeUDPLifetime != nil && de.probeUDPLifetime.timer != nil {
|
|
de.probeUDPLifetime.timer.Stop()
|
|
de.probeUDPLifetime.timer = nil
|
|
}
|
|
de.heartBeatTimer = nil
|
|
|
|
if de.heartbeatDisabled {
|
|
// If control override to disable heartBeatTimer set, return early.
|
|
return
|
|
}
|
|
|
|
if de.lastSendExt.IsZero() {
|
|
// Shouldn't happen.
|
|
return
|
|
}
|
|
|
|
now := mono.Now()
|
|
if now.Sub(de.lastSendExt) > sessionActiveTimeout {
|
|
// Session's idle. Stop heartbeating.
|
|
de.c.dlogf("[v1] magicsock: disco: ending heartbeats for idle session to %v (%v)", de.publicKey.ShortString(), de.discoShort())
|
|
if afterInactivityFor, ok := de.maybeProbeUDPLifetimeLocked(); ok {
|
|
// This is the best place to best effort schedule a probe of UDP
|
|
// path lifetime in the future as it loosely translates to "UDP path
|
|
// is inactive".
|
|
//
|
|
// Note: wireguard-go schedules a WireGuard keepalive packet (by
|
|
// default, not tied to persistent keepalive feature) 10 seconds in
|
|
// the future after receiving an authenticated data packet. It's
|
|
// typically only sent by one side based on how the WireGuard state
|
|
// machine controls the timer. So, if we are on the receiving end of
|
|
// that keepalive, de.lastSendExt won't move, assuming there is no
|
|
// other user-generated traffic. This is one reason why we perform
|
|
// a more granular check of the last packets in/out time, below, as
|
|
// a WireGuard keepalive may have fallen somewhere within the
|
|
// sessionActiveTimeout window. heartbeatForLifetime will also
|
|
// perform a similar check, and reschedule as necessary.
|
|
inactiveFor := now.Sub(max(de.lastSendAny, de.lastRecvUDPAny.LoadAtomic()))
|
|
after := afterInactivityFor - inactiveFor
|
|
if after < 0 {
|
|
// shouldn't happen
|
|
return
|
|
}
|
|
de.scheduleHeartbeatForLifetimeLocked(after, heartbeatForLifetimeViaSessionInactive)
|
|
}
|
|
return
|
|
}
|
|
|
|
udpAddr, _, _ := de.addrForSendLocked(now)
|
|
if udpAddr.IsValid() {
|
|
// We have a preferred path. Ping that every 2 seconds.
|
|
de.startDiscoPingLocked(udpAddr, now, pingHeartbeat, 0, nil)
|
|
}
|
|
|
|
if de.wantFullPingLocked(now) {
|
|
de.sendDiscoPingsLocked(now, true)
|
|
}
|
|
|
|
de.heartBeatTimer = time.AfterFunc(heartbeatInterval, de.heartbeat)
|
|
}
|
|
|
|
// setHeartbeatDisabled sets heartbeatDisabled to the provided value.
|
|
func (de *endpoint) setHeartbeatDisabled(v bool) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
de.heartbeatDisabled = v
|
|
}
|
|
|
|
// wantFullPingLocked reports whether we should ping to all our peers looking for
|
|
// a better path.
|
|
//
|
|
// de.mu must be held.
|
|
func (de *endpoint) wantFullPingLocked(now mono.Time) bool {
|
|
if runtime.GOOS == "js" {
|
|
return false
|
|
}
|
|
if !de.bestAddr.IsValid() || de.lastFullPing.IsZero() {
|
|
return true
|
|
}
|
|
if now.After(de.trustBestAddrUntil) {
|
|
return true
|
|
}
|
|
if de.bestAddr.latency <= goodEnoughLatency {
|
|
return false
|
|
}
|
|
if now.Sub(de.lastFullPing) >= upgradeInterval {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (de *endpoint) noteTxActivityExtTriggerLocked(now mono.Time) {
|
|
de.lastSendExt = now
|
|
if de.heartBeatTimer == nil && !de.heartbeatDisabled {
|
|
de.heartBeatTimer = time.AfterFunc(heartbeatInterval, de.heartbeat)
|
|
}
|
|
}
|
|
|
|
// MaxDiscoPingSize is the largest useful ping message size that we
|
|
// can send - the maximum packet size minus the IPv4 and UDP headers.
|
|
var MaxDiscoPingSize = tstun.MaxPacketSize - 20 - 8
|
|
|
|
type pingResultAndCallback struct {
|
|
taken atomic.Bool // first CompareAndSwamp from false to true takes ownership of res
|
|
res *ipnstate.PingResult
|
|
cb func(*ipnstate.PingResult)
|
|
}
|
|
|
|
func (p *pingResultAndCallback) reply() bool {
|
|
return p != nil && p.taken.CompareAndSwap(false, true)
|
|
}
|
|
|
|
// discoPing starts a disco-level ping for the "tailscale ping" command (or other
|
|
// callers, such as c2n). res is value to call cb with, already partially
|
|
// filled. cb must be called at most once. Once called, ownership of res passes to cb.
|
|
func (de *endpoint) discoPing(res *ipnstate.PingResult, size int, cb func(*ipnstate.PingResult)) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
if de.expired {
|
|
res.Err = errExpired.Error()
|
|
cb(res)
|
|
return
|
|
}
|
|
if size > MaxDiscoPingSize {
|
|
res.Err = errPingTooBig.Error()
|
|
cb(res)
|
|
return
|
|
}
|
|
|
|
resCB := &pingResultAndCallback{res: res, cb: cb}
|
|
|
|
now := mono.Now()
|
|
udpAddr, derpAddr := de.addrForPingSizeLocked(now, size)
|
|
|
|
if derpAddr.IsValid() {
|
|
de.startDiscoPingLocked(derpAddr, now, pingCLI, size, resCB)
|
|
}
|
|
if udpAddr.IsValid() && now.Before(de.trustBestAddrUntil) {
|
|
// Already have an active session, so just ping the address we're using.
|
|
// Otherwise "tailscale ping" results to a node on the local network
|
|
// can look like they're bouncing between, say 10.0.0.0/9 and the peer's
|
|
// IPv6 address, both 1ms away, and it's random who replies first.
|
|
de.startDiscoPingLocked(udpAddr, now, pingCLI, size, resCB)
|
|
} else {
|
|
for ep := range de.endpointState {
|
|
de.startDiscoPingLocked(ep, now, pingCLI, size, resCB)
|
|
}
|
|
}
|
|
}
|
|
|
|
var (
|
|
errExpired = errors.New("peer's node key has expired")
|
|
errNoUDPOrDERP = errors.New("no UDP or DERP addr")
|
|
errPingTooBig = errors.New("ping size too big")
|
|
)
|
|
|
|
func (de *endpoint) send(buffs [][]byte) error {
|
|
de.mu.Lock()
|
|
if de.expired {
|
|
de.mu.Unlock()
|
|
return errExpired
|
|
}
|
|
|
|
now := mono.Now()
|
|
udpAddr, derpAddr, startWGPing := de.addrForSendLocked(now)
|
|
|
|
if de.isWireguardOnly {
|
|
if startWGPing {
|
|
de.sendWireGuardOnlyPingsLocked(now)
|
|
}
|
|
} else if !udpAddr.IsValid() || now.After(de.trustBestAddrUntil) {
|
|
de.sendDiscoPingsLocked(now, true)
|
|
}
|
|
de.noteTxActivityExtTriggerLocked(now)
|
|
de.lastSendAny = now
|
|
de.mu.Unlock()
|
|
|
|
if !udpAddr.IsValid() && !derpAddr.IsValid() {
|
|
return errNoUDPOrDERP
|
|
}
|
|
var err error
|
|
if udpAddr.IsValid() {
|
|
_, err = de.c.sendUDPBatch(udpAddr, buffs)
|
|
|
|
// If the error is known to indicate that the endpoint is no longer
|
|
// usable, clear the endpoint statistics so that the next send will
|
|
// re-evaluate the best endpoint.
|
|
if err != nil && isBadEndpointErr(err) {
|
|
de.noteBadEndpoint(udpAddr)
|
|
}
|
|
|
|
// TODO(raggi): needs updating for accuracy, as in error conditions we may have partial sends.
|
|
if stats := de.c.stats.Load(); err == nil && stats != nil {
|
|
var txBytes int
|
|
for _, b := range buffs {
|
|
txBytes += len(b)
|
|
}
|
|
stats.UpdateTxPhysical(de.nodeAddr, udpAddr, txBytes)
|
|
}
|
|
}
|
|
if derpAddr.IsValid() {
|
|
allOk := true
|
|
for _, buff := range buffs {
|
|
ok, _ := de.c.sendAddr(derpAddr, de.publicKey, buff)
|
|
if stats := de.c.stats.Load(); stats != nil {
|
|
stats.UpdateTxPhysical(de.nodeAddr, derpAddr, len(buff))
|
|
}
|
|
if !ok {
|
|
allOk = false
|
|
}
|
|
}
|
|
if allOk {
|
|
return nil
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
|
|
// probeUDPLifetimeCliffDoneLocked is called when a disco
|
|
// pingHeartbeatForUDPLifetime is being cleaned up. result contains the reason
|
|
// for the cleanup, txid contains the ping's txid.
|
|
// probeUDPLifetimeCliffDoneLocked may schedule another
|
|
// pingHeartbeatForUDPLifetime in the future if there is another cliff remaining
|
|
// for the current probing cycle.
|
|
func (de *endpoint) probeUDPLifetimeCliffDoneLocked(result discoPingResult, txid stun.TxID) {
|
|
p := de.probeUDPLifetime
|
|
if p == nil || !p.cycleActive || de.probeUDPLifetime.timer != nil || txid != p.lastTxID {
|
|
// Probing may have been disabled while heartbeats were in flight. This
|
|
// can also be a duplicate or late arriving result.
|
|
return
|
|
}
|
|
metricUDPLifetimeCliffsCompleted.Add(1)
|
|
if result != discoPongReceived || p.currentCliff >= len(p.config.Cliffs)-1 {
|
|
maxCliffIndex := p.currentCliff
|
|
if result != discoPongReceived {
|
|
maxCliffIndex = p.currentCliff - 1
|
|
}
|
|
var maxCliffDuration time.Duration
|
|
if maxCliffIndex >= 0 {
|
|
maxCliffDuration = p.config.Cliffs[maxCliffIndex]
|
|
}
|
|
p.cycleCompleteMaxCliffEndpointLocked(maxCliffIndex)
|
|
de.c.dlogf("[v1] magicsock: disco: UDP lifetime probe cycle completed max cliff=%v for %v (%v)",
|
|
maxCliffDuration, de.publicKey.ShortString(), de.discoShort())
|
|
metricUDPLifetimeCyclesCompleted.Add(1)
|
|
p.resetCycleEndpointLocked()
|
|
} else {
|
|
p.currentCliff++
|
|
if after, ok := de.maybeProbeUDPLifetimeLocked(); ok {
|
|
de.scheduleHeartbeatForLifetimeLocked(after, heartbeatForLifetimeViaPongRx)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (de *endpoint) discoPingTimeout(txid stun.TxID) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
sp, ok := de.sentPing[txid]
|
|
if !ok {
|
|
return
|
|
}
|
|
if debugDisco() || !de.bestAddr.IsValid() || mono.Now().After(de.trustBestAddrUntil) {
|
|
de.c.dlogf("[v1] magicsock: disco: timeout waiting for pong %x from %v (%v, %v)", txid[:6], sp.to, de.publicKey.ShortString(), de.discoShort())
|
|
}
|
|
de.removeSentDiscoPingLocked(txid, sp, discoPingTimedOut)
|
|
}
|
|
|
|
// forgetDiscoPing is called when a ping fails to send.
|
|
func (de *endpoint) forgetDiscoPing(txid stun.TxID) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
if sp, ok := de.sentPing[txid]; ok {
|
|
de.removeSentDiscoPingLocked(txid, sp, discoPingFailed)
|
|
}
|
|
}
|
|
|
|
// discoPingResult represents the result of an attempted disco ping send
|
|
// operation.
|
|
type discoPingResult int
|
|
|
|
const (
|
|
discoPingResultUnknown discoPingResult = iota
|
|
discoPingFailed
|
|
discoPingTimedOut
|
|
discoPongReceived
|
|
)
|
|
|
|
func (de *endpoint) removeSentDiscoPingLocked(txid stun.TxID, sp sentPing, result discoPingResult) {
|
|
// Stop the timer for the case where sendPing failed to write to UDP.
|
|
// In the case of a timer already having fired, this is a no-op:
|
|
sp.timer.Stop()
|
|
if sp.purpose == pingHeartbeatForUDPLifetime {
|
|
de.probeUDPLifetimeCliffDoneLocked(result, txid)
|
|
}
|
|
delete(de.sentPing, txid)
|
|
}
|
|
|
|
// discoPingSize is the size of a complete disco ping packet, without any padding.
|
|
const discoPingSize = len(disco.Magic) + key.DiscoPublicRawLen + disco.NonceLen +
|
|
poly1305.TagSize + disco.MessageHeaderLen + disco.PingLen
|
|
|
|
// sendDiscoPing sends a ping with the provided txid to ep using de's discoKey. size
|
|
// is the desired disco message size, including all disco headers but excluding IP/UDP
|
|
// headers.
|
|
//
|
|
// The caller (startDiscoPingLocked) should've already recorded the ping in
|
|
// sentPing and set up the timer.
|
|
//
|
|
// The caller should use de.discoKey as the discoKey argument.
|
|
// It is passed in so that sendDiscoPing doesn't need to lock de.mu.
|
|
func (de *endpoint) sendDiscoPing(ep netip.AddrPort, discoKey key.DiscoPublic, txid stun.TxID, size int, logLevel discoLogLevel) {
|
|
size = min(size, MaxDiscoPingSize)
|
|
padding := max(size-discoPingSize, 0)
|
|
|
|
sent, _ := de.c.sendDiscoMessage(ep, de.publicKey, discoKey, &disco.Ping{
|
|
TxID: [12]byte(txid),
|
|
NodeKey: de.c.publicKeyAtomic.Load(),
|
|
Padding: padding,
|
|
}, logLevel)
|
|
if !sent {
|
|
de.forgetDiscoPing(txid)
|
|
return
|
|
}
|
|
|
|
if size != 0 {
|
|
metricSentDiscoPeerMTUProbes.Add(1)
|
|
metricSentDiscoPeerMTUProbeBytes.Add(int64(pingSizeToPktLen(size, ep.Addr().Is6())))
|
|
}
|
|
}
|
|
|
|
// discoPingPurpose is the reason why a discovery ping message was sent.
|
|
type discoPingPurpose int
|
|
|
|
//go:generate go run tailscale.com/cmd/addlicense -file discopingpurpose_string.go go run golang.org/x/tools/cmd/stringer -type=discoPingPurpose -trimprefix=ping
|
|
const (
|
|
// pingDiscovery means that purpose of a ping was to see if a
|
|
// path was valid.
|
|
pingDiscovery discoPingPurpose = iota
|
|
|
|
// pingHeartbeat means that purpose of a ping was whether a
|
|
// peer was still there.
|
|
pingHeartbeat
|
|
|
|
// pingCLI means that the user is running "tailscale ping"
|
|
// from the CLI. These types of pings can go over DERP.
|
|
pingCLI
|
|
|
|
// pingHeartbeatForUDPLifetime means that the purpose of a ping was to
|
|
// discover whether the UDP path was still active through any and all
|
|
// stateful middleboxes involved.
|
|
pingHeartbeatForUDPLifetime
|
|
)
|
|
|
|
// startDiscoPingLocked sends a disco ping to ep in a separate goroutine. resCB,
|
|
// if non-nil, means that a caller external to the magicsock package internals
|
|
// is interested in the result (such as a CLI "tailscale ping" or a c2n ping
|
|
// request, etc)
|
|
func (de *endpoint) startDiscoPingLocked(ep netip.AddrPort, now mono.Time, purpose discoPingPurpose, size int, resCB *pingResultAndCallback) {
|
|
if runtime.GOOS == "js" {
|
|
return
|
|
}
|
|
epDisco := de.disco.Load()
|
|
if epDisco == nil {
|
|
return
|
|
}
|
|
if purpose != pingCLI {
|
|
st, ok := de.endpointState[ep]
|
|
if !ok {
|
|
// Shouldn't happen. But don't ping an endpoint that's
|
|
// not active for us.
|
|
de.c.logf("magicsock: disco: [unexpected] attempt to ping no longer live endpoint %v", ep)
|
|
return
|
|
}
|
|
st.lastPing = now
|
|
}
|
|
|
|
// If we are doing a discovery ping or a CLI ping with no specified size
|
|
// to a non DERP address, then probe the MTU. Otherwise just send the
|
|
// one specified ping.
|
|
|
|
// Default to sending a single ping of the specified size
|
|
sizes := []int{size}
|
|
if de.c.PeerMTUEnabled() {
|
|
isDerp := ep.Addr() == tailcfg.DerpMagicIPAddr
|
|
if !isDerp && ((purpose == pingDiscovery) || (purpose == pingCLI && size == 0)) {
|
|
de.c.dlogf("[v1] magicsock: starting MTU probe")
|
|
sizes = mtuProbePingSizesV4
|
|
if ep.Addr().Is6() {
|
|
sizes = mtuProbePingSizesV6
|
|
}
|
|
}
|
|
}
|
|
|
|
logLevel := discoLog
|
|
if purpose == pingHeartbeat {
|
|
logLevel = discoVerboseLog
|
|
}
|
|
if purpose == pingCLI {
|
|
de.noteTxActivityExtTriggerLocked(now)
|
|
}
|
|
de.lastSendAny = now
|
|
for _, s := range sizes {
|
|
txid := stun.NewTxID()
|
|
de.sentPing[txid] = sentPing{
|
|
to: ep,
|
|
at: now,
|
|
timer: time.AfterFunc(pingTimeoutDuration, func() { de.discoPingTimeout(txid) }),
|
|
purpose: purpose,
|
|
resCB: resCB,
|
|
size: s,
|
|
}
|
|
if purpose == pingHeartbeatForUDPLifetime && de.probeUDPLifetime != nil {
|
|
de.probeUDPLifetime.lastTxID = txid
|
|
}
|
|
go de.sendDiscoPing(ep, epDisco.key, txid, s, logLevel)
|
|
}
|
|
|
|
}
|
|
|
|
// sendDiscoPingsLocked starts pinging all of ep's endpoints.
|
|
func (de *endpoint) sendDiscoPingsLocked(now mono.Time, sendCallMeMaybe bool) {
|
|
de.lastFullPing = now
|
|
var sentAny bool
|
|
for ep, st := range de.endpointState {
|
|
if st.shouldDeleteLocked() {
|
|
de.deleteEndpointLocked("sendPingsLocked", ep)
|
|
continue
|
|
}
|
|
if runtime.GOOS == "js" {
|
|
continue
|
|
}
|
|
if !st.lastPing.IsZero() && now.Sub(st.lastPing) < discoPingInterval {
|
|
continue
|
|
}
|
|
|
|
firstPing := !sentAny
|
|
sentAny = true
|
|
|
|
if firstPing && sendCallMeMaybe {
|
|
de.c.dlogf("[v1] magicsock: disco: send, starting discovery for %v (%v)", de.publicKey.ShortString(), de.discoShort())
|
|
}
|
|
|
|
de.startDiscoPingLocked(ep, now, pingDiscovery, 0, nil)
|
|
}
|
|
derpAddr := de.derpAddr
|
|
if sentAny && sendCallMeMaybe && derpAddr.IsValid() {
|
|
// Have our magicsock.Conn figure out its STUN endpoint (if
|
|
// it doesn't know already) and then send a CallMeMaybe
|
|
// message to our peer via DERP informing them that we've
|
|
// sent so our firewall ports are probably open and now
|
|
// would be a good time for them to connect.
|
|
go de.c.enqueueCallMeMaybe(derpAddr, de)
|
|
}
|
|
}
|
|
|
|
// sendWireGuardOnlyPingsLocked evaluates all available addresses for
|
|
// a WireGuard only endpoint and initates an ICMP ping for useable
|
|
// addresses.
|
|
func (de *endpoint) sendWireGuardOnlyPingsLocked(now mono.Time) {
|
|
if runtime.GOOS == "js" {
|
|
return
|
|
}
|
|
|
|
// Normally we only send pings at a low rate as the decision to start
|
|
// sending a ping sets bestAddrAtUntil with a reasonable time to keep trying
|
|
// that address, however, if that code changed we may want to be sure that
|
|
// we don't ever send excessive pings to avoid impact to the client/user.
|
|
if !now.After(de.lastFullPing.Add(10 * time.Second)) {
|
|
return
|
|
}
|
|
de.lastFullPing = now
|
|
|
|
for ipp := range de.endpointState {
|
|
if ipp.Addr().Is4() && de.c.noV4.Load() {
|
|
continue
|
|
}
|
|
if ipp.Addr().Is6() && de.c.noV6.Load() {
|
|
continue
|
|
}
|
|
|
|
go de.sendWireGuardOnlyPing(ipp, now)
|
|
}
|
|
}
|
|
|
|
// sendWireGuardOnlyPing sends a ICMP ping to a WireGuard only address to
|
|
// discover the latency.
|
|
func (de *endpoint) sendWireGuardOnlyPing(ipp netip.AddrPort, now mono.Time) {
|
|
ctx, cancel := context.WithTimeout(de.c.connCtx, 5*time.Second)
|
|
defer cancel()
|
|
|
|
de.setLastPing(ipp, now)
|
|
|
|
addr := &net.IPAddr{
|
|
IP: net.IP(ipp.Addr().AsSlice()),
|
|
Zone: ipp.Addr().Zone(),
|
|
}
|
|
|
|
p := de.c.getPinger()
|
|
if p == nil {
|
|
de.c.logf("[v2] magicsock: sendWireGuardOnlyPingLocked: pinger is nil")
|
|
return
|
|
}
|
|
|
|
latency, err := p.Send(ctx, addr, nil)
|
|
if err != nil {
|
|
de.c.logf("[v2] magicsock: sendWireGuardOnlyPingLocked: %s", err)
|
|
return
|
|
}
|
|
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
state, ok := de.endpointState[ipp]
|
|
if !ok {
|
|
return
|
|
}
|
|
state.addPongReplyLocked(pongReply{
|
|
latency: latency,
|
|
pongAt: now,
|
|
from: ipp,
|
|
pongSrc: netip.AddrPort{}, // We don't know this.
|
|
})
|
|
}
|
|
|
|
// setLastPing sets lastPing on the endpointState to now.
|
|
func (de *endpoint) setLastPing(ipp netip.AddrPort, now mono.Time) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
state, ok := de.endpointState[ipp]
|
|
if !ok {
|
|
return
|
|
}
|
|
state.lastPing = now
|
|
}
|
|
|
|
// updateFromNode updates the endpoint based on a tailcfg.Node from a NetMap
|
|
// update.
|
|
func (de *endpoint) updateFromNode(n tailcfg.NodeView, heartbeatDisabled bool, probeUDPLifetimeEnabled bool) {
|
|
if !n.Valid() {
|
|
panic("nil node when updating endpoint")
|
|
}
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
de.heartbeatDisabled = heartbeatDisabled
|
|
if probeUDPLifetimeEnabled {
|
|
de.setProbeUDPLifetimeConfigLocked(defaultProbeUDPLifetimeConfig)
|
|
} else {
|
|
de.setProbeUDPLifetimeConfigLocked(nil)
|
|
}
|
|
de.expired = n.Expired()
|
|
|
|
epDisco := de.disco.Load()
|
|
var discoKey key.DiscoPublic
|
|
if epDisco != nil {
|
|
discoKey = epDisco.key
|
|
}
|
|
|
|
if discoKey != n.DiscoKey() {
|
|
de.c.logf("[v1] magicsock: disco: node %s changed from %s to %s", de.publicKey.ShortString(), discoKey, n.DiscoKey())
|
|
de.disco.Store(&endpointDisco{
|
|
key: n.DiscoKey(),
|
|
short: n.DiscoKey().ShortString(),
|
|
})
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "updateFromNode-resetLocked",
|
|
})
|
|
de.resetLocked()
|
|
}
|
|
if n.DERP() == "" {
|
|
if de.derpAddr.IsValid() {
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "updateFromNode-remove-DERP",
|
|
From: de.derpAddr,
|
|
})
|
|
}
|
|
de.derpAddr = netip.AddrPort{}
|
|
} else {
|
|
newDerp, _ := netip.ParseAddrPort(n.DERP())
|
|
if de.derpAddr != newDerp {
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "updateFromNode-DERP",
|
|
From: de.derpAddr,
|
|
To: newDerp,
|
|
})
|
|
}
|
|
de.derpAddr = newDerp
|
|
}
|
|
|
|
de.setEndpointsLocked(n.Endpoints())
|
|
}
|
|
|
|
func (de *endpoint) setEndpointsLocked(eps interface {
|
|
Len() int
|
|
At(i int) netip.AddrPort
|
|
}) {
|
|
for _, st := range de.endpointState {
|
|
st.index = indexSentinelDeleted // assume deleted until updated in next loop
|
|
}
|
|
|
|
var newIpps []netip.AddrPort
|
|
for i := range eps.Len() {
|
|
if i > math.MaxInt16 {
|
|
// Seems unlikely.
|
|
break
|
|
}
|
|
ipp := eps.At(i)
|
|
if !ipp.IsValid() {
|
|
de.c.logf("magicsock: bogus netmap endpoint from %v", eps)
|
|
continue
|
|
}
|
|
if st, ok := de.endpointState[ipp]; ok {
|
|
st.index = int16(i)
|
|
} else {
|
|
de.endpointState[ipp] = &endpointState{index: int16(i)}
|
|
newIpps = append(newIpps, ipp)
|
|
}
|
|
}
|
|
if len(newIpps) > 0 {
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "updateFromNode-new-Endpoints",
|
|
To: newIpps,
|
|
})
|
|
}
|
|
|
|
// Now delete anything unless it's still in the network map or
|
|
// was a recently discovered endpoint.
|
|
for ep, st := range de.endpointState {
|
|
if st.shouldDeleteLocked() {
|
|
de.deleteEndpointLocked("updateFromNode", ep)
|
|
}
|
|
}
|
|
}
|
|
|
|
// addCandidateEndpoint adds ep as an endpoint to which we should send
|
|
// future pings. If there is an existing endpointState for ep, and forRxPingTxID
|
|
// matches the last received ping TxID, this function reports true, otherwise
|
|
// false.
|
|
//
|
|
// This is called once we've already verified that we got a valid
|
|
// discovery message from de via ep.
|
|
func (de *endpoint) addCandidateEndpoint(ep netip.AddrPort, forRxPingTxID stun.TxID) (duplicatePing bool) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
if st, ok := de.endpointState[ep]; ok {
|
|
duplicatePing = forRxPingTxID == st.lastGotPingTxID
|
|
if !duplicatePing {
|
|
st.lastGotPingTxID = forRxPingTxID
|
|
}
|
|
if st.lastGotPing.IsZero() {
|
|
// Already-known endpoint from the network map.
|
|
return duplicatePing
|
|
}
|
|
st.lastGotPing = time.Now()
|
|
return duplicatePing
|
|
}
|
|
|
|
// Newly discovered endpoint. Exciting!
|
|
de.c.dlogf("[v1] magicsock: disco: adding %v as candidate endpoint for %v (%s)", ep, de.discoShort(), de.publicKey.ShortString())
|
|
de.endpointState[ep] = &endpointState{
|
|
lastGotPing: time.Now(),
|
|
lastGotPingTxID: forRxPingTxID,
|
|
}
|
|
|
|
// If for some reason this gets very large, do some cleanup.
|
|
if size := len(de.endpointState); size > 100 {
|
|
for ep, st := range de.endpointState {
|
|
if st.shouldDeleteLocked() {
|
|
de.deleteEndpointLocked("addCandidateEndpoint", ep)
|
|
}
|
|
}
|
|
size2 := len(de.endpointState)
|
|
de.c.dlogf("[v1] magicsock: disco: addCandidateEndpoint pruned %v candidate set from %v to %v entries", size, size2)
|
|
}
|
|
return false
|
|
}
|
|
|
|
// clearBestAddrLocked clears the bestAddr and related fields such that future
|
|
// packets will re-evaluate the best address to send to next.
|
|
//
|
|
// de.mu must be held.
|
|
func (de *endpoint) clearBestAddrLocked() {
|
|
de.setBestAddrLocked(addrQuality{})
|
|
de.bestAddrAt = 0
|
|
de.trustBestAddrUntil = 0
|
|
}
|
|
|
|
// noteBadEndpoint marks ipp as a bad endpoint that would need to be
|
|
// re-evaluated before future use, this should be called for example if a send
|
|
// to ipp fails due to a host unreachable error or similar.
|
|
func (de *endpoint) noteBadEndpoint(ipp netip.AddrPort) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
de.clearBestAddrLocked()
|
|
|
|
if st, ok := de.endpointState[ipp]; ok {
|
|
st.clear()
|
|
}
|
|
}
|
|
|
|
// noteConnectivityChange is called when connectivity changes enough
|
|
// that we should question our earlier assumptions about which paths
|
|
// work.
|
|
func (de *endpoint) noteConnectivityChange() {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
de.clearBestAddrLocked()
|
|
|
|
for k := range de.endpointState {
|
|
de.endpointState[k].clear()
|
|
}
|
|
}
|
|
|
|
// pingSizeToPktLen calculates the minimum path MTU that would permit
|
|
// a disco ping message of length size to reach its target at
|
|
// addr. size is the length of the entire disco message including
|
|
// disco headers. If size is zero, assume it is the safe wire MTU.
|
|
func pingSizeToPktLen(size int, is6 bool) tstun.WireMTU {
|
|
if size == 0 {
|
|
return tstun.SafeWireMTU()
|
|
}
|
|
headerLen := ipv4.HeaderLen
|
|
if is6 {
|
|
headerLen = ipv6.HeaderLen
|
|
}
|
|
headerLen += 8 // UDP header length
|
|
return tstun.WireMTU(size + headerLen)
|
|
}
|
|
|
|
// pktLenToPingSize calculates the ping payload size that would
|
|
// create a disco ping message whose on-the-wire length is exactly mtu
|
|
// bytes long. If mtu is zero or less than the minimum ping size, then
|
|
// no MTU probe is desired and return zero for an unpadded ping.
|
|
func pktLenToPingSize(mtu tstun.WireMTU, is6 bool) int {
|
|
if mtu == 0 {
|
|
return 0
|
|
}
|
|
headerLen := ipv4.HeaderLen
|
|
if is6 {
|
|
headerLen = ipv6.HeaderLen
|
|
}
|
|
headerLen += 8 // UDP header length
|
|
if mtu < tstun.WireMTU(headerLen) {
|
|
return 0
|
|
}
|
|
return int(mtu) - headerLen
|
|
}
|
|
|
|
// handlePongConnLocked handles a Pong message (a reply to an earlier ping).
|
|
// It should be called with the Conn.mu held.
|
|
//
|
|
// It reports whether m.TxID corresponds to a ping that this endpoint sent.
|
|
func (de *endpoint) handlePongConnLocked(m *disco.Pong, di *discoInfo, src netip.AddrPort) (knownTxID bool) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
isDerp := src.Addr() == tailcfg.DerpMagicIPAddr
|
|
|
|
sp, ok := de.sentPing[m.TxID]
|
|
if !ok {
|
|
// This is not a pong for a ping we sent.
|
|
return false
|
|
}
|
|
knownTxID = true // for naked returns below
|
|
de.removeSentDiscoPingLocked(m.TxID, sp, discoPongReceived)
|
|
|
|
pktLen := int(pingSizeToPktLen(sp.size, sp.to.Addr().Is6()))
|
|
if sp.size != 0 {
|
|
m := getPeerMTUsProbedMetric(tstun.WireMTU(pktLen))
|
|
m.Add(1)
|
|
if metricMaxPeerMTUProbed.Value() < int64(pktLen) {
|
|
metricMaxPeerMTUProbed.Set(int64(pktLen))
|
|
}
|
|
}
|
|
|
|
now := mono.Now()
|
|
latency := now.Sub(sp.at)
|
|
|
|
if !isDerp {
|
|
st, ok := de.endpointState[sp.to]
|
|
if !ok {
|
|
// This is no longer an endpoint we care about.
|
|
return
|
|
}
|
|
|
|
de.c.peerMap.setNodeKeyForIPPort(src, de.publicKey)
|
|
|
|
st.addPongReplyLocked(pongReply{
|
|
latency: latency,
|
|
pongAt: now,
|
|
from: src,
|
|
pongSrc: m.Src,
|
|
})
|
|
}
|
|
|
|
if sp.purpose != pingHeartbeat && sp.purpose != pingHeartbeatForUDPLifetime {
|
|
de.c.dlogf("[v1] magicsock: disco: %v<-%v (%v, %v) got pong tx=%x latency=%v pktlen=%v pong.src=%v%v", de.c.discoShort, de.discoShort(), de.publicKey.ShortString(), src, m.TxID[:6], latency.Round(time.Millisecond), pktLen, m.Src, logger.ArgWriter(func(bw *bufio.Writer) {
|
|
if sp.to != src {
|
|
fmt.Fprintf(bw, " ping.to=%v", sp.to)
|
|
}
|
|
}))
|
|
}
|
|
|
|
// Currently only CLI ping uses this callback.
|
|
if sp.resCB.reply() {
|
|
if sp.purpose == pingCLI {
|
|
de.c.populateCLIPingResponseLocked(sp.resCB.res, latency, sp.to)
|
|
}
|
|
go sp.resCB.cb(sp.resCB.res)
|
|
}
|
|
|
|
// Promote this pong response to our current best address if it's lower latency.
|
|
// TODO(bradfitz): decide how latency vs. preference order affects decision
|
|
if !isDerp {
|
|
thisPong := addrQuality{sp.to, latency, tstun.WireMTU(pingSizeToPktLen(sp.size, sp.to.Addr().Is6()))}
|
|
if betterAddr(thisPong, de.bestAddr) {
|
|
de.c.logf("magicsock: disco: node %v %v now using %v mtu=%v tx=%x", de.publicKey.ShortString(), de.discoShort(), sp.to, thisPong.wireMTU, m.TxID[:6])
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "handlePingLocked-bestAddr-update",
|
|
From: de.bestAddr,
|
|
To: thisPong,
|
|
})
|
|
de.setBestAddrLocked(thisPong)
|
|
}
|
|
if de.bestAddr.AddrPort == thisPong.AddrPort {
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "handlePingLocked-bestAddr-latency",
|
|
From: de.bestAddr,
|
|
To: thisPong,
|
|
})
|
|
de.bestAddr.latency = latency
|
|
de.bestAddrAt = now
|
|
de.trustBestAddrUntil = now.Add(trustUDPAddrDuration)
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
// addrQuality is an IPPort with an associated latency and path mtu.
|
|
type addrQuality struct {
|
|
netip.AddrPort
|
|
latency time.Duration
|
|
wireMTU tstun.WireMTU
|
|
}
|
|
|
|
func (a addrQuality) String() string {
|
|
return fmt.Sprintf("%v@%v+%v", a.AddrPort, a.latency, a.wireMTU)
|
|
}
|
|
|
|
// betterAddr reports whether a is a better addr to use than b.
|
|
func betterAddr(a, b addrQuality) bool {
|
|
if a.AddrPort == b.AddrPort {
|
|
if a.wireMTU > b.wireMTU {
|
|
// TODO(val): Think harder about the case of lower
|
|
// latency and smaller or unknown MTU, and higher
|
|
// latency but larger MTU. Probably in most cases the
|
|
// largest MTU will also be the lowest latency but we
|
|
// can't depend on that.
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
if !b.IsValid() {
|
|
return true
|
|
}
|
|
if !a.IsValid() {
|
|
return false
|
|
}
|
|
|
|
// Each address starts with a set of points (from 0 to 100) that
|
|
// represents how much faster they are than the highest-latency
|
|
// endpoint. For example, if a has latency 200ms and b has latency
|
|
// 190ms, then a starts with 0 points and b starts with 5 points since
|
|
// it's 5% faster.
|
|
var aPoints, bPoints int
|
|
if a.latency > b.latency && a.latency > 0 {
|
|
bPoints = int(100 - ((b.latency * 100) / a.latency))
|
|
} else if b.latency > 0 {
|
|
aPoints = int(100 - ((a.latency * 100) / b.latency))
|
|
}
|
|
|
|
// Prefer private IPs over public IPs as long as the latencies are
|
|
// roughly equivalent, since it's less likely that a user will have to
|
|
// pay for the bandwidth in a cloud environment.
|
|
//
|
|
// Additionally, prefer any loopback address strongly over non-loopback
|
|
// addresses, and prefer link-local unicast addresses over other types
|
|
// of private IP addresses since it's definitionally more likely that
|
|
// they'll be on the same network segment than a general private IP.
|
|
if a.Addr().IsLoopback() {
|
|
aPoints += 50
|
|
} else if a.Addr().IsLinkLocalUnicast() {
|
|
aPoints += 30
|
|
} else if a.Addr().IsPrivate() {
|
|
aPoints += 20
|
|
}
|
|
if b.Addr().IsLoopback() {
|
|
bPoints += 50
|
|
} else if b.Addr().IsLinkLocalUnicast() {
|
|
bPoints += 30
|
|
} else if b.Addr().IsPrivate() {
|
|
bPoints += 20
|
|
}
|
|
|
|
// Prefer IPv6 for being a bit more robust, as long as
|
|
// the latencies are roughly equivalent.
|
|
if a.Addr().Is6() {
|
|
aPoints += 10
|
|
}
|
|
if b.Addr().Is6() {
|
|
bPoints += 10
|
|
}
|
|
|
|
// Don't change anything if the latency improvement is less than 1%; we
|
|
// want a bit of "stickiness" (a.k.a. hysteresis) to avoid flapping if
|
|
// there's two roughly-equivalent endpoints.
|
|
//
|
|
// Points are essentially the percentage improvement of latency vs. the
|
|
// slower endpoint; absent any boosts from private IPs, IPv6, etc., a
|
|
// will be a better address than b by a fraction of 1% or less if
|
|
// aPoints <= 1 and bPoints == 0.
|
|
if aPoints <= 1 && bPoints == 0 {
|
|
return false
|
|
}
|
|
|
|
return aPoints > bPoints
|
|
}
|
|
|
|
// handleCallMeMaybe handles a CallMeMaybe discovery message via
|
|
// DERP. The contract for use of this message is that the peer has
|
|
// already sent to us via UDP, so their stateful firewall should be
|
|
// open. Now we can Ping back and make it through.
|
|
func (de *endpoint) handleCallMeMaybe(m *disco.CallMeMaybe) {
|
|
if runtime.GOOS == "js" {
|
|
// Nothing to do on js/wasm if we can't send UDP packets anyway.
|
|
return
|
|
}
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
now := time.Now()
|
|
for ep := range de.isCallMeMaybeEP {
|
|
de.isCallMeMaybeEP[ep] = false // mark for deletion
|
|
}
|
|
var newEPs []netip.AddrPort
|
|
for _, ep := range m.MyNumber {
|
|
if ep.Addr().Is6() && ep.Addr().IsLinkLocalUnicast() {
|
|
// We send these out, but ignore them for now.
|
|
// TODO: teach the ping code to ping on all interfaces
|
|
// for these.
|
|
continue
|
|
}
|
|
mak.Set(&de.isCallMeMaybeEP, ep, true)
|
|
if es, ok := de.endpointState[ep]; ok {
|
|
es.callMeMaybeTime = now
|
|
} else {
|
|
de.endpointState[ep] = &endpointState{callMeMaybeTime: now}
|
|
newEPs = append(newEPs, ep)
|
|
}
|
|
}
|
|
if len(newEPs) > 0 {
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "handleCallMeMaybe-new-endpoints",
|
|
To: newEPs,
|
|
})
|
|
|
|
de.c.dlogf("[v1] magicsock: disco: call-me-maybe from %v %v added new endpoints: %v",
|
|
de.publicKey.ShortString(), de.discoShort(),
|
|
logger.ArgWriter(func(w *bufio.Writer) {
|
|
for i, ep := range newEPs {
|
|
if i > 0 {
|
|
w.WriteString(", ")
|
|
}
|
|
w.WriteString(ep.String())
|
|
}
|
|
}))
|
|
}
|
|
|
|
// Delete any prior CallMeMaybe endpoints that weren't included
|
|
// in this message.
|
|
for ep, want := range de.isCallMeMaybeEP {
|
|
if !want {
|
|
delete(de.isCallMeMaybeEP, ep)
|
|
de.deleteEndpointLocked("handleCallMeMaybe", ep)
|
|
}
|
|
}
|
|
|
|
// Zero out all the lastPing times to force sendPingsLocked to send new ones,
|
|
// even if it's been less than 5 seconds ago.
|
|
for _, st := range de.endpointState {
|
|
st.lastPing = 0
|
|
}
|
|
de.sendDiscoPingsLocked(mono.Now(), false)
|
|
}
|
|
|
|
func (de *endpoint) populatePeerStatus(ps *ipnstate.PeerStatus) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
ps.Relay = de.c.derpRegionCodeOfIDLocked(int(de.derpAddr.Port()))
|
|
|
|
if de.lastSendExt.IsZero() {
|
|
return
|
|
}
|
|
|
|
now := mono.Now()
|
|
ps.LastWrite = de.lastSendExt.WallTime()
|
|
ps.Active = now.Sub(de.lastSendExt) < sessionActiveTimeout
|
|
|
|
if udpAddr, derpAddr, _ := de.addrForSendLocked(now); udpAddr.IsValid() && !derpAddr.IsValid() {
|
|
ps.CurAddr = udpAddr.String()
|
|
}
|
|
}
|
|
|
|
// stopAndReset stops timers associated with de and resets its state back to zero.
|
|
// It's called when a discovery endpoint is no longer present in the
|
|
// NetworkMap, or when magicsock is transitioning from running to
|
|
// stopped state (via SetPrivateKey(zero))
|
|
func (de *endpoint) stopAndReset() {
|
|
atomic.AddInt64(&de.numStopAndResetAtomic, 1)
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
|
|
if closing := de.c.closing.Load(); !closing {
|
|
if de.isWireguardOnly {
|
|
de.c.logf("[v1] magicsock: doing cleanup for wireguard key %s", de.publicKey.ShortString())
|
|
} else {
|
|
de.c.logf("[v1] magicsock: doing cleanup for discovery key %s", de.discoShort())
|
|
}
|
|
}
|
|
|
|
de.debugUpdates.Add(EndpointChange{
|
|
When: time.Now(),
|
|
What: "stopAndReset-resetLocked",
|
|
})
|
|
de.resetLocked()
|
|
if de.heartBeatTimer != nil {
|
|
de.heartBeatTimer.Stop()
|
|
de.heartBeatTimer = nil
|
|
}
|
|
}
|
|
|
|
// resetLocked clears all the endpoint's p2p state, reverting it to a
|
|
// DERP-only endpoint. It does not stop the endpoint's heartbeat
|
|
// timer, if one is running.
|
|
func (de *endpoint) resetLocked() {
|
|
de.lastSendExt = 0
|
|
de.lastFullPing = 0
|
|
de.clearBestAddrLocked()
|
|
for _, es := range de.endpointState {
|
|
es.lastPing = 0
|
|
}
|
|
if !de.isWireguardOnly {
|
|
for txid, sp := range de.sentPing {
|
|
de.removeSentDiscoPingLocked(txid, sp, discoPingResultUnknown)
|
|
}
|
|
}
|
|
de.probeUDPLifetime.resetCycleEndpointLocked()
|
|
}
|
|
|
|
func (de *endpoint) numStopAndReset() int64 {
|
|
return atomic.LoadInt64(&de.numStopAndResetAtomic)
|
|
}
|
|
|
|
func (de *endpoint) setDERPHome(regionID uint16) {
|
|
de.mu.Lock()
|
|
defer de.mu.Unlock()
|
|
de.derpAddr = netip.AddrPortFrom(tailcfg.DerpMagicIPAddr, uint16(regionID))
|
|
}
|