TUN-6813: Only proxy ICMP packets when warp-routing is enabled

This commit is contained in:
cthuang 2022-09-30 10:43:39 +01:00
parent eacc8c648d
commit 49438f30f5
7 changed files with 145 additions and 22 deletions

View File

@ -39,6 +39,7 @@ type Orchestrator interface {
UpdateConfig(version int32, config []byte) *pogs.UpdateConfigurationResponse UpdateConfig(version int32, config []byte) *pogs.UpdateConfigurationResponse
GetConfigJSON() ([]byte, error) GetConfigJSON() ([]byte, error)
GetOriginProxy() (OriginProxy, error) GetOriginProxy() (OriginProxy, error)
WarpRoutingEnabled() (enabled bool)
} }
type NamedTunnelProperties struct { type NamedTunnelProperties struct {

View File

@ -56,6 +56,10 @@ func (mcr *mockOrchestrator) GetOriginProxy() (OriginProxy, error) {
return mcr.originProxy, nil return mcr.originProxy, nil
} }
func (mcr *mockOrchestrator) WarpRoutingEnabled() (enabled bool) {
return true
}
type mockOriginProxy struct{} type mockOriginProxy struct{}
func (moc *mockOriginProxy) ProxyHTTP( func (moc *mockOriginProxy) ProxyHTTP(

View File

@ -75,7 +75,7 @@ func NewQUICConnection(
sessionDemuxChan := make(chan *packet.Session, demuxChanCapacity) sessionDemuxChan := make(chan *packet.Session, demuxChanCapacity)
datagramMuxer := quicpogs.NewDatagramMuxerV2(session, logger, sessionDemuxChan) datagramMuxer := quicpogs.NewDatagramMuxerV2(session, logger, sessionDemuxChan)
sessionManager := datagramsession.NewManager(logger, datagramMuxer.SendToSession, sessionDemuxChan) sessionManager := datagramsession.NewManager(logger, datagramMuxer.SendToSession, sessionDemuxChan)
packetRouter := packet.NewRouter(packetRouterConfig, datagramMuxer, &returnPipe{muxer: datagramMuxer}, logger) packetRouter := packet.NewRouter(packetRouterConfig, datagramMuxer, &returnPipe{muxer: datagramMuxer}, logger, orchestrator.WarpRoutingEnabled)
return &QUICConnection{ return &QUICConnection{
session: session, session: session,

View File

@ -27,10 +27,12 @@ type Orchestrator struct {
// Used by UpdateConfig to make sure one update at a time // Used by UpdateConfig to make sure one update at a time
lock sync.RWMutex lock sync.RWMutex
// Underlying value is proxy.Proxy, can be read without the lock, but still needs the lock to update // Underlying value is proxy.Proxy, can be read without the lock, but still needs the lock to update
proxy atomic.Value proxy atomic.Value
config *Config // TODO: TUN-6815 Use atomic.Bool once we upgrade to go 1.19. 1 Means enabled and 0 means disabled
tags []tunnelpogs.Tag warpRoutingEnabled uint32
log *zerolog.Logger config *Config
tags []tunnelpogs.Tag
log *zerolog.Logger
// orchestrator must not handle any more updates after shutdownC is closed // orchestrator must not handle any more updates after shutdownC is closed
shutdownC <-chan struct{} shutdownC <-chan struct{}
@ -122,6 +124,11 @@ func (o *Orchestrator) updateIngress(ingressRules ingress.Ingress, warpRouting i
o.proxy.Store(newProxy) o.proxy.Store(newProxy)
o.config.Ingress = &ingressRules o.config.Ingress = &ingressRules
o.config.WarpRouting = warpRouting o.config.WarpRouting = warpRouting
if warpRouting.Enabled {
atomic.StoreUint32(&o.warpRoutingEnabled, 1)
} else {
atomic.StoreUint32(&o.warpRoutingEnabled, 0)
}
// If proxyShutdownC is nil, there is no previous running proxy // If proxyShutdownC is nil, there is no previous running proxy
if o.proxyShutdownC != nil { if o.proxyShutdownC != nil {
@ -190,6 +197,14 @@ func (o *Orchestrator) GetOriginProxy() (connection.OriginProxy, error) {
return proxy, nil return proxy, nil
} }
// TODO: TUN-6815 consider storing WarpRouting.Enabled as atomic.Bool once we upgrade to go 1.19
func (o *Orchestrator) WarpRoutingEnabled() (enabled bool) {
if atomic.LoadUint32(&o.warpRoutingEnabled) == 0 {
return false
}
return true
}
func (o *Orchestrator) waitToCloseLastProxy() { func (o *Orchestrator) waitToCloseLastProxy() {
<-o.shutdownC <-o.shutdownC
o.lock.Lock() o.lock.Lock()

View File

@ -55,6 +55,7 @@ func TestUpdateConfiguration(t *testing.T) {
initOriginProxy, err := orchestrator.GetOriginProxy() initOriginProxy, err := orchestrator.GetOriginProxy()
require.NoError(t, err) require.NoError(t, err)
require.IsType(t, &proxy.Proxy{}, initOriginProxy) require.IsType(t, &proxy.Proxy{}, initOriginProxy)
require.False(t, orchestrator.WarpRoutingEnabled())
configJSONV2 := []byte(` configJSONV2 := []byte(`
{ {
@ -122,6 +123,7 @@ func TestUpdateConfiguration(t *testing.T) {
require.Equal(t, false, configV2.Ingress.Rules[2].Config.NoTLSVerify) require.Equal(t, false, configV2.Ingress.Rules[2].Config.NoTLSVerify)
require.Equal(t, true, configV2.Ingress.Rules[2].Config.NoHappyEyeballs) require.Equal(t, true, configV2.Ingress.Rules[2].Config.NoHappyEyeballs)
require.True(t, configV2.WarpRouting.Enabled) require.True(t, configV2.WarpRouting.Enabled)
require.Equal(t, configV2.WarpRouting.Enabled, orchestrator.WarpRoutingEnabled())
require.Equal(t, configV2.WarpRouting.ConnectTimeout.Duration, 10*time.Second) require.Equal(t, configV2.WarpRouting.ConnectTimeout.Duration, 10*time.Second)
originProxyV2, err := orchestrator.GetOriginProxy() originProxyV2, err := orchestrator.GetOriginProxy()
@ -166,6 +168,7 @@ func TestUpdateConfiguration(t *testing.T) {
require.True(t, configV10.Ingress.Rules[0].Matches("blogs.tunnel.io", "/2022/02/10")) require.True(t, configV10.Ingress.Rules[0].Matches("blogs.tunnel.io", "/2022/02/10"))
require.Equal(t, ingress.HelloWorldService, configV10.Ingress.Rules[0].Service.String()) require.Equal(t, ingress.HelloWorldService, configV10.Ingress.Rules[0].Service.String())
require.False(t, configV10.WarpRouting.Enabled) require.False(t, configV10.WarpRouting.Enabled)
require.Equal(t, configV10.WarpRouting.Enabled, orchestrator.WarpRoutingEnabled())
originProxyV10, err := orchestrator.GetOriginProxy() originProxyV10, err := orchestrator.GetOriginProxy()
require.NoError(t, err) require.NoError(t, err)

View File

@ -23,10 +23,11 @@ type Upstream interface {
// Router routes packets between Upstream and ICMPRouter. Currently it rejects all other type of ICMP packets // Router routes packets between Upstream and ICMPRouter. Currently it rejects all other type of ICMP packets
type Router struct { type Router struct {
upstream Upstream upstream Upstream
returnPipe FunnelUniPipe returnPipe FunnelUniPipe
globalConfig *GlobalRouterConfig globalConfig *GlobalRouterConfig
logger *zerolog.Logger logger *zerolog.Logger
checkRouterEnabledFunc func() bool
} }
// GlobalRouterConfig is the configuration shared by all instance of Router. // GlobalRouterConfig is the configuration shared by all instance of Router.
@ -37,12 +38,13 @@ type GlobalRouterConfig struct {
Zone string Zone string
} }
func NewRouter(globalConfig *GlobalRouterConfig, upstream Upstream, returnPipe FunnelUniPipe, logger *zerolog.Logger) *Router { func NewRouter(globalConfig *GlobalRouterConfig, upstream Upstream, returnPipe FunnelUniPipe, logger *zerolog.Logger, checkRouterEnabledFunc func() bool) *Router {
return &Router{ return &Router{
upstream: upstream, upstream: upstream,
returnPipe: returnPipe, returnPipe: returnPipe,
globalConfig: globalConfig, globalConfig: globalConfig,
logger: logger, logger: logger,
checkRouterEnabledFunc: checkRouterEnabledFunc,
} }
} }
@ -54,10 +56,16 @@ func (r *Router) Serve(ctx context.Context) error {
if err != nil { if err != nil {
return err return err
} }
// Drop packets if ICMPRouter wasn't created // Drop packets if ICMPRouter wasn't created
if r.globalConfig == nil { if r.globalConfig == nil {
continue continue
} }
if enabled := r.checkRouterEnabledFunc(); !enabled {
continue
}
icmpPacket, err := icmpDecoder.Decode(rawPacket) icmpPacket, err := icmpDecoder.Decode(rawPacket)
if err != nil { if err != nil {
r.logger.Err(err).Msg("Failed to decode ICMP packet from quic datagram") r.logger.Err(err).Msg("Failed to decode ICMP packet from quic datagram")

View File

@ -5,7 +5,9 @@ import (
"context" "context"
"fmt" "fmt"
"net/netip" "net/netip"
"sync/atomic"
"testing" "testing"
"time"
"github.com/google/gopacket/layers" "github.com/google/gopacket/layers"
"github.com/rs/zerolog" "github.com/rs/zerolog"
@ -16,7 +18,12 @@ import (
) )
var ( var (
noopLogger = zerolog.Nop() noopLogger = zerolog.Nop()
packetConfig = &GlobalRouterConfig{
ICMPRouter: &mockICMPRouter{},
IPv4Src: netip.MustParseAddr("172.16.0.1"),
IPv6Src: netip.MustParseAddr("fd51:2391:523:f4ee::1"),
}
) )
func TestRouterReturnTTLExceed(t *testing.T) { func TestRouterReturnTTLExceed(t *testing.T) {
@ -26,12 +33,9 @@ func TestRouterReturnTTLExceed(t *testing.T) {
returnPipe := &mockFunnelUniPipe{ returnPipe := &mockFunnelUniPipe{
uniPipe: make(chan RawPacket), uniPipe: make(chan RawPacket),
} }
packetConfig := &GlobalRouterConfig{ routerEnabled := &routerEnabledChecker{}
ICMPRouter: &mockICMPRouter{}, routerEnabled.set(true)
IPv4Src: netip.MustParseAddr("172.16.0.1"), router := NewRouter(packetConfig, upstream, returnPipe, &noopLogger, routerEnabled.isEnabled)
IPv6Src: netip.MustParseAddr("fd51:2391:523:f4ee::1"),
}
router := NewRouter(packetConfig, upstream, returnPipe, &noopLogger)
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
routerStopped := make(chan struct{}) routerStopped := make(chan struct{})
go func() { go func() {
@ -80,12 +84,71 @@ func TestRouterReturnTTLExceed(t *testing.T) {
<-routerStopped <-routerStopped
} }
func TestRouterCheckEnabled(t *testing.T) {
upstream := &mockUpstream{
source: make(chan RawPacket),
}
returnPipe := &mockFunnelUniPipe{
uniPipe: make(chan RawPacket),
}
routerEnabled := &routerEnabledChecker{}
router := NewRouter(packetConfig, upstream, returnPipe, &noopLogger, routerEnabled.isEnabled)
ctx, cancel := context.WithCancel(context.Background())
routerStopped := make(chan struct{})
go func() {
router.Serve(ctx)
close(routerStopped)
}()
pk := ICMP{
IP: &IP{
Src: netip.MustParseAddr("192.168.1.1"),
Dst: netip.MustParseAddr("10.0.0.1"),
Protocol: layers.IPProtocolICMPv4,
TTL: 1,
},
Message: &icmp.Message{
Type: ipv4.ICMPTypeEcho,
Code: 0,
Body: &icmp.Echo{
ID: 12481,
Seq: 8036,
Data: []byte(t.Name()),
},
},
}
// router is disabled
require.NoError(t, upstream.send(&pk))
select {
case <-time.After(time.Millisecond * 10):
case <-returnPipe.uniPipe:
t.Error("Unexpected reply when router is disabled")
}
routerEnabled.set(true)
// router is enabled, expects reply
require.NoError(t, upstream.send(&pk))
<-returnPipe.uniPipe
routerEnabled.set(false)
// router is disabled
require.NoError(t, upstream.send(&pk))
select {
case <-time.After(time.Millisecond * 10):
case <-returnPipe.uniPipe:
t.Error("Unexpected reply when router is disabled")
}
cancel()
<-routerStopped
}
func assertTTLExceed(t *testing.T, originalPacket *ICMP, expectedSrc netip.Addr, upstream *mockUpstream, returnPipe *mockFunnelUniPipe) { func assertTTLExceed(t *testing.T, originalPacket *ICMP, expectedSrc netip.Addr, upstream *mockUpstream, returnPipe *mockFunnelUniPipe) {
encoder := NewEncoder() encoder := NewEncoder()
rawPacket, err := encoder.Encode(originalPacket) rawPacket, err := encoder.Encode(originalPacket)
require.NoError(t, err) require.NoError(t, err)
upstream.source <- rawPacket upstream.source <- rawPacket
resp := <-returnPipe.uniPipe resp := <-returnPipe.uniPipe
decoder := NewICMPDecoder() decoder := NewICMPDecoder()
decoded, err := decoder.Decode(resp) decoded, err := decoder.Decode(resp)
@ -111,6 +174,16 @@ type mockUpstream struct {
source chan RawPacket source chan RawPacket
} }
func (ms *mockUpstream) send(pk Packet) error {
encoder := NewEncoder()
rawPacket, err := encoder.Encode(pk)
if err != nil {
return err
}
ms.source <- rawPacket
return nil
}
func (ms *mockUpstream) ReceivePacket(ctx context.Context) (RawPacket, error) { func (ms *mockUpstream) ReceivePacket(ctx context.Context) (RawPacket, error) {
select { select {
case <-ctx.Done(): case <-ctx.Done():
@ -129,3 +202,22 @@ func (mir mockICMPRouter) Serve(ctx context.Context) error {
func (mir mockICMPRouter) Request(pk *ICMP, responder FunnelUniPipe) error { func (mir mockICMPRouter) Request(pk *ICMP, responder FunnelUniPipe) error {
return fmt.Errorf("Request not implemented by mockICMPRouter") return fmt.Errorf("Request not implemented by mockICMPRouter")
} }
type routerEnabledChecker struct {
enabled uint32
}
func (rec *routerEnabledChecker) isEnabled() bool {
if atomic.LoadUint32(&rec.enabled) == 0 {
return false
}
return true
}
func (rec *routerEnabledChecker) set(enabled bool) {
if enabled {
atomic.StoreUint32(&rec.enabled, 1)
} else {
atomic.StoreUint32(&rec.enabled, 0)
}
}