mirror of https://github.com/coder/coder.git
fix: change servertailnet to register the DERP dialer before setting DERP map (#12137)
I noticed a possible race where tailnet.Conn can try to dial the embedded region before we've set our custom dialer that send the DERP in-memory. This closes that race and adds a test case for servertailnet with no STUN and an embedded relay
This commit is contained in:
parent
1bb4aecf49
commit
2d0b9106c0
|
@ -52,19 +52,41 @@ func NewServerTailnet(
|
||||||
traceProvider trace.TracerProvider,
|
traceProvider trace.TracerProvider,
|
||||||
) (*ServerTailnet, error) {
|
) (*ServerTailnet, error) {
|
||||||
logger = logger.Named("servertailnet")
|
logger = logger.Named("servertailnet")
|
||||||
originalDerpMap := derpMapFn()
|
|
||||||
conn, err := tailnet.NewConn(&tailnet.Options{
|
conn, err := tailnet.NewConn(&tailnet.Options{
|
||||||
Addresses: []netip.Prefix{netip.PrefixFrom(tailnet.IP(), 128)},
|
Addresses: []netip.Prefix{netip.PrefixFrom(tailnet.IP(), 128)},
|
||||||
DERPMap: originalDerpMap,
|
|
||||||
DERPForceWebSockets: derpForceWebSockets,
|
DERPForceWebSockets: derpForceWebSockets,
|
||||||
Logger: logger,
|
Logger: logger,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, xerrors.Errorf("create tailnet conn: %w", err)
|
return nil, xerrors.Errorf("create tailnet conn: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
serverCtx, cancel := context.WithCancel(ctx)
|
serverCtx, cancel := context.WithCancel(ctx)
|
||||||
|
|
||||||
|
// This is set to allow local DERP traffic to be proxied through memory
|
||||||
|
// instead of needing to hit the external access URL. Don't use the ctx
|
||||||
|
// given in this callback, it's only valid while connecting.
|
||||||
|
if derpServer != nil {
|
||||||
|
conn.SetDERPRegionDialer(func(_ context.Context, region *tailcfg.DERPRegion) net.Conn {
|
||||||
|
if !region.EmbeddedRelay {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
logger.Debug(ctx, "connecting to embedded DERP via in-memory pipe")
|
||||||
|
left, right := net.Pipe()
|
||||||
|
go func() {
|
||||||
|
defer left.Close()
|
||||||
|
defer right.Close()
|
||||||
|
brw := bufio.NewReadWriter(bufio.NewReader(right), bufio.NewWriter(right))
|
||||||
|
derpServer.Accept(ctx, right, brw, "internal")
|
||||||
|
}()
|
||||||
|
return left
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
derpMapUpdaterClosed := make(chan struct{})
|
derpMapUpdaterClosed := make(chan struct{})
|
||||||
|
originalDerpMap := derpMapFn()
|
||||||
|
// it's important to set the DERPRegionDialer above _before_ we set the DERP map so that if
|
||||||
|
// there is an embedded relay, we use the local in-memory dialer.
|
||||||
|
conn.SetDERPMap(originalDerpMap)
|
||||||
go func() {
|
go func() {
|
||||||
defer close(derpMapUpdaterClosed)
|
defer close(derpMapUpdaterClosed)
|
||||||
|
|
||||||
|
@ -139,25 +161,6 @@ func NewServerTailnet(
|
||||||
// registering the callback also triggers send of the initial node
|
// registering the callback also triggers send of the initial node
|
||||||
tn.coordinatee.SetNodeCallback(tn.nodeCallback)
|
tn.coordinatee.SetNodeCallback(tn.nodeCallback)
|
||||||
|
|
||||||
// This is set to allow local DERP traffic to be proxied through memory
|
|
||||||
// instead of needing to hit the external access URL. Don't use the ctx
|
|
||||||
// given in this callback, it's only valid while connecting.
|
|
||||||
if derpServer != nil {
|
|
||||||
conn.SetDERPRegionDialer(func(_ context.Context, region *tailcfg.DERPRegion) net.Conn {
|
|
||||||
if !region.EmbeddedRelay {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
left, right := net.Pipe()
|
|
||||||
go func() {
|
|
||||||
defer left.Close()
|
|
||||||
defer right.Close()
|
|
||||||
brw := bufio.NewReadWriter(bufio.NewReader(right), bufio.NewWriter(right))
|
|
||||||
derpServer.Accept(ctx, right, brw, "internal")
|
|
||||||
}()
|
|
||||||
return left
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
go tn.watchAgentUpdates()
|
go tn.watchAgentUpdates()
|
||||||
go tn.expireOldAgents()
|
go tn.expireOldAgents()
|
||||||
return tn, nil
|
return tn, nil
|
||||||
|
|
|
@ -50,6 +50,24 @@ func TestServerTailnet_AgentConn_OK(t *testing.T) {
|
||||||
assert.True(t, conn.AwaitReachable(ctx))
|
assert.True(t, conn.AwaitReachable(ctx))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestServerTailnet_AgentConn_NoSTUN(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitMedium)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
// Connect through the ServerTailnet
|
||||||
|
agents, serverTailnet := setupServerTailnetAgent(t, 1,
|
||||||
|
tailnettest.DisableSTUN, tailnettest.DERPIsEmbedded)
|
||||||
|
a := agents[0]
|
||||||
|
|
||||||
|
conn, release, err := serverTailnet.AgentConn(ctx, a.id)
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer release()
|
||||||
|
|
||||||
|
assert.True(t, conn.AwaitReachable(ctx))
|
||||||
|
}
|
||||||
|
|
||||||
func TestServerTailnet_ReverseProxy(t *testing.T) {
|
func TestServerTailnet_ReverseProxy(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
@ -311,9 +329,9 @@ type agentWithID struct {
|
||||||
agent.Agent
|
agent.Agent
|
||||||
}
|
}
|
||||||
|
|
||||||
func setupServerTailnetAgent(t *testing.T, agentNum int) ([]agentWithID, *coderd.ServerTailnet) {
|
func setupServerTailnetAgent(t *testing.T, agentNum int, opts ...tailnettest.DERPAndStunOption) ([]agentWithID, *coderd.ServerTailnet) {
|
||||||
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
||||||
derpMap, derpServer := tailnettest.RunDERPAndSTUN(t)
|
derpMap, derpServer := tailnettest.RunDERPAndSTUN(t, opts...)
|
||||||
|
|
||||||
coord := tailnet.NewCoordinator(logger)
|
coord := tailnet.NewCoordinator(logger)
|
||||||
t.Cleanup(func() {
|
t.Cleanup(func() {
|
||||||
|
|
|
@ -204,7 +204,8 @@ func (c *configMaps) netMapLocked() *netmap.NetworkMap {
|
||||||
nm.Addresses = make([]netip.Prefix, len(c.addresses))
|
nm.Addresses = make([]netip.Prefix, len(c.addresses))
|
||||||
copy(nm.Addresses, c.addresses)
|
copy(nm.Addresses, c.addresses)
|
||||||
|
|
||||||
nm.DERPMap = c.derpMap.Clone()
|
// we don't need to set the DERPMap in the network map because we separately
|
||||||
|
// send the DERPMap directly via SetDERPMap
|
||||||
nm.Peers = c.peerConfigLocked()
|
nm.Peers = c.peerConfigLocked()
|
||||||
nm.SelfNode.Addresses = nm.Addresses
|
nm.SelfNode.Addresses = nm.Addresses
|
||||||
nm.SelfNode.AllowedIPs = nm.Addresses
|
nm.SelfNode.AllowedIPs = nm.Addresses
|
||||||
|
|
|
@ -116,9 +116,6 @@ func NewConn(options *Options) (conn *Conn, err error) {
|
||||||
if len(options.Addresses) == 0 {
|
if len(options.Addresses) == 0 {
|
||||||
return nil, xerrors.New("At least one IP range must be provided")
|
return nil, xerrors.New("At least one IP range must be provided")
|
||||||
}
|
}
|
||||||
if options.DERPMap == nil {
|
|
||||||
return nil, xerrors.New("DERPMap must be provided")
|
|
||||||
}
|
|
||||||
|
|
||||||
nodePrivateKey := key.NewNode()
|
nodePrivateKey := key.NewNode()
|
||||||
var nodeID tailcfg.NodeID
|
var nodeID tailcfg.NodeID
|
||||||
|
@ -219,7 +216,9 @@ func NewConn(options *Options) (conn *Conn, err error) {
|
||||||
magicConn.DiscoPublicKey(),
|
magicConn.DiscoPublicKey(),
|
||||||
)
|
)
|
||||||
cfgMaps.setAddresses(options.Addresses)
|
cfgMaps.setAddresses(options.Addresses)
|
||||||
cfgMaps.setDERPMap(options.DERPMap)
|
if options.DERPMap != nil {
|
||||||
|
cfgMaps.setDERPMap(options.DERPMap)
|
||||||
|
}
|
||||||
cfgMaps.setBlockEndpoints(options.BlockEndpoints)
|
cfgMaps.setBlockEndpoints(options.BlockEndpoints)
|
||||||
|
|
||||||
nodeUp := newNodeUpdater(
|
nodeUp := newNodeUpdater(
|
||||||
|
|
|
@ -32,22 +32,47 @@ import (
|
||||||
//go:generate mockgen -destination ./coordinatormock.go -package tailnettest github.com/coder/coder/v2/tailnet Coordinator
|
//go:generate mockgen -destination ./coordinatormock.go -package tailnettest github.com/coder/coder/v2/tailnet Coordinator
|
||||||
//go:generate mockgen -destination ./coordinateemock.go -package tailnettest github.com/coder/coder/v2/tailnet Coordinatee
|
//go:generate mockgen -destination ./coordinateemock.go -package tailnettest github.com/coder/coder/v2/tailnet Coordinatee
|
||||||
|
|
||||||
|
type derpAndSTUNCfg struct {
|
||||||
|
DisableSTUN bool
|
||||||
|
DERPIsEmbedded bool
|
||||||
|
}
|
||||||
|
|
||||||
|
type DERPAndStunOption func(cfg *derpAndSTUNCfg)
|
||||||
|
|
||||||
|
func DisableSTUN(cfg *derpAndSTUNCfg) {
|
||||||
|
cfg.DisableSTUN = true
|
||||||
|
}
|
||||||
|
|
||||||
|
func DERPIsEmbedded(cfg *derpAndSTUNCfg) {
|
||||||
|
cfg.DERPIsEmbedded = true
|
||||||
|
}
|
||||||
|
|
||||||
// RunDERPAndSTUN creates a DERP mapping for tests.
|
// RunDERPAndSTUN creates a DERP mapping for tests.
|
||||||
func RunDERPAndSTUN(t *testing.T) (*tailcfg.DERPMap, *derp.Server) {
|
func RunDERPAndSTUN(t *testing.T, opts ...DERPAndStunOption) (*tailcfg.DERPMap, *derp.Server) {
|
||||||
|
cfg := new(derpAndSTUNCfg)
|
||||||
|
for _, o := range opts {
|
||||||
|
o(cfg)
|
||||||
|
}
|
||||||
logf := tailnet.Logger(slogtest.Make(t, nil))
|
logf := tailnet.Logger(slogtest.Make(t, nil))
|
||||||
d := derp.NewServer(key.NewNode(), logf)
|
d := derp.NewServer(key.NewNode(), logf)
|
||||||
server := httptest.NewUnstartedServer(derphttp.Handler(d))
|
server := httptest.NewUnstartedServer(derphttp.Handler(d))
|
||||||
server.Config.ErrorLog = tslogger.StdLogger(logf)
|
server.Config.ErrorLog = tslogger.StdLogger(logf)
|
||||||
server.Config.TLSNextProto = make(map[string]func(*http.Server, *tls.Conn, http.Handler))
|
server.Config.TLSNextProto = make(map[string]func(*http.Server, *tls.Conn, http.Handler))
|
||||||
server.StartTLS()
|
server.StartTLS()
|
||||||
|
|
||||||
stunAddr, stunCleanup := stuntest.ServeWithPacketListener(t, nettype.Std{})
|
|
||||||
t.Cleanup(func() {
|
t.Cleanup(func() {
|
||||||
server.CloseClientConnections()
|
server.CloseClientConnections()
|
||||||
server.Close()
|
server.Close()
|
||||||
d.Close()
|
d.Close()
|
||||||
stunCleanup()
|
|
||||||
})
|
})
|
||||||
|
|
||||||
|
stunPort := -1
|
||||||
|
if !cfg.DisableSTUN {
|
||||||
|
stunAddr, stunCleanup := stuntest.ServeWithPacketListener(t, nettype.Std{})
|
||||||
|
t.Cleanup(func() {
|
||||||
|
stunCleanup()
|
||||||
|
})
|
||||||
|
stunPort = stunAddr.Port
|
||||||
|
}
|
||||||
tcpAddr, ok := server.Listener.Addr().(*net.TCPAddr)
|
tcpAddr, ok := server.Listener.Addr().(*net.TCPAddr)
|
||||||
if !ok {
|
if !ok {
|
||||||
t.FailNow()
|
t.FailNow()
|
||||||
|
@ -65,11 +90,12 @@ func RunDERPAndSTUN(t *testing.T) (*tailcfg.DERPMap, *derp.Server) {
|
||||||
RegionID: 1,
|
RegionID: 1,
|
||||||
IPv4: "127.0.0.1",
|
IPv4: "127.0.0.1",
|
||||||
IPv6: "none",
|
IPv6: "none",
|
||||||
STUNPort: stunAddr.Port,
|
STUNPort: stunPort,
|
||||||
DERPPort: tcpAddr.Port,
|
DERPPort: tcpAddr.Port,
|
||||||
InsecureForTests: true,
|
InsecureForTests: true,
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
EmbeddedRelay: cfg.DERPIsEmbedded,
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
}, d
|
}, d
|
||||||
|
|
Loading…
Reference in New Issue