Something went wrong. Try again.
Cloudflare Tunnel client with NetBSD support
Something went wrong. Try again.
37 kB · 1368 lines
Go
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369package proxy
import ( "bufio" "bytes" "context" "flag" "fmt" "io" "net" "net/http" "net/http/httptest" "strings" "sync" "testing" "time"
"github.com/gobwas/ws/wsutil" gorillaWS "github.com/gorilla/websocket" "github.com/rs/zerolog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/urfave/cli/v2" "go.uber.org/mock/gomock" "golang.org/x/sync/errgroup"
"github.com/cloudflare/cloudflared/mocks"
cfdflow "github.com/cloudflare/cloudflared/flow"
"github.com/cloudflare/cloudflared/cfio" "github.com/cloudflare/cloudflared/config" "github.com/cloudflare/cloudflared/connection" "github.com/cloudflare/cloudflared/hello" "github.com/cloudflare/cloudflared/ingress" "github.com/cloudflare/cloudflared/quicktunnelauth" "github.com/cloudflare/cloudflared/tracing" "github.com/cloudflare/cloudflared/tunnelrpc/pogs")
var ( testTags = []pogs.Tag{{Name: "Name", Value: "value"}} testDefaultDialer = ingress.NewDialer(ingress.WarpRoutingConfig{ ConnectTimeout: config.CustomDuration{Duration: 1 * time.Second}, TCPKeepAlive: config.CustomDuration{Duration: 15 * time.Second}, MaxActiveFlows: 0, }))
type mockHTTPRespWriter struct { *httptest.ResponseRecorder writtenHeaders http.Header}
// nonInterruptingRequestBody intentionally violates http.Request.Body's Close// contract. It isolates the HTTP/2 response-close fallback: proxy cancellation// must keep response cleanup from waiting forever even when a body wrapper// fails to unblock its pending Read.type nonInterruptingRequestBody struct { readStarted chan struct{} readDone chan struct{} releaseRead chan struct{} closeCalled chan struct{} readOnce sync.Once closeOnce sync.Once}
func newNonInterruptingRequestBody() *nonInterruptingRequestBody { return &nonInterruptingRequestBody{ readStarted: make(chan struct{}), readDone: make(chan struct{}), releaseRead: make(chan struct{}), closeCalled: make(chan struct{}), }}
func (b *nonInterruptingRequestBody) Read([]byte) (int, error) { b.readOnce.Do(func() { close(b.readStarted) }) <-b.releaseRead close(b.readDone) return 0, io.EOF}
func (b *nonInterruptingRequestBody) Close() error { b.closeOnce.Do(func() { close(b.closeCalled) }) return nil}
func newMockHTTPRespWriter() *mockHTTPRespWriter { return &mockHTTPRespWriter{ ResponseRecorder: httptest.NewRecorder(), }}
func (w *mockHTTPRespWriter) WriteResponse() error { return nil}
func (w *mockHTTPRespWriter) WriteRespHeaders(status int, header http.Header) error { w.writtenHeaders = header.Clone() w.WriteHeader(status) for header, val := range header { w.Header()[header] = val } return nil}
func (w *mockHTTPRespWriter) AddTrailer(trailerName, trailerValue string) { // do nothing}
func (w *mockHTTPRespWriter) Read(data []byte) (int, error) { return 0, fmt.Errorf("mockHTTPRespWriter doesn't implement io.Reader")}
func (m *mockHTTPRespWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { panic("Hijack not implemented")}
type mockHTTPRequestAuthorizer struct { called bool}
func (h *mockHTTPRequestAuthorizer) AuthorizeHTTP(w http.ResponseWriter, _ *http.Request) (connection.HTTPRequestAuthorizationDecision, string, error) { h.called = true w.WriteHeader(http.StatusTeapot) return connection.HTTPRequestAuthorizationHandled, "test_handled", nil}
type mockWSRespWriter struct { *mockHTTPRespWriter writeNotification chan []byte reader io.Reader}
func newMockWSRespWriter(reader io.Reader) *mockWSRespWriter { return &mockWSRespWriter{ newMockHTTPRespWriter(), make(chan []byte), reader, }}
func (w *mockWSRespWriter) Write(data []byte) (int, error) { w.writeNotification <- data return len(data), nil}
func (w *mockWSRespWriter) respBody() io.ReadWriter { data := <-w.writeNotification return bytes.NewBuffer(data)}
func (w *mockWSRespWriter) Close() error { close(w.writeNotification) return nil}
func (w *mockWSRespWriter) Read(data []byte) (int, error) { return w.reader.Read(data)}
func (w *mockWSRespWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { panic("Hijack not implemented")}
type mockSSERespWriter struct { *mockHTTPRespWriter writeNotification chan []byte}
func newMockSSERespWriter() *mockSSERespWriter { return &mockSSERespWriter{ newMockHTTPRespWriter(), make(chan []byte), }}
func (w *mockSSERespWriter) Write(data []byte) (int, error) { newData := make([]byte, len(data)) copy(newData, data)
w.writeNotification <- newData return len(data), nil}
func (w *mockSSERespWriter) WriteString(str string) (int, error) { return w.Write([]byte(str))}
func (w *mockSSERespWriter) ReadBytes() []byte { return <-w.writeNotification}
func TestHTTPRequestAuthorizerRunsBeforeIngressSelection(t *testing.T) { t.Parallel()
log := zerolog.Nop() authorizer := &mockHTTPRequestAuthorizer{} originProxy := NewOriginProxyWithHTTPRequestAuthorizer( ingress.Ingress{}, nil, testTags, cfdflow.NewLimiter(0), authorizer, &log, ) responseWriter := newMockHTTPRespWriter() request, err := http.NewRequest(http.MethodGet, "https://test-tunnel.trycloudflare.com/", nil) require.NoError(t, err)
err = originProxy.ProxyHTTP( responseWriter, tracing.NewTracedHTTPRequest(request, 0, &log), false, ) require.NoError(t, err) assert.True(t, authorizer.called) assert.Equal(t, http.StatusTeapot, responseWriter.Code) for _, tag := range testTags { assert.Empty(t, request.Header.Get(TagHeaderNamePrefix+tag.Name)) }}
// HTTP/2 ServeHTTP and QUIC dispatchRequest both route HTTP and WebSocket// traffic through ProxyHTTP, so this table exercises their shared boundary.func TestQuickTunnelAuthorizationBeforeOriginSelection(t *testing.T) { t.Parallel()
const stateCookieName = "__Secure-cloudflared-qt-auth-state-test" tests := []struct { name string protected bool authenticated bool method string isWebsocket bool expectedOrigin bool expectedStatus int }{ { name: "public HTTP request", method: http.MethodGet, expectedOrigin: true, expectedStatus: http.StatusNoContent, }, { name: "authenticated HTTP request", protected: true, authenticated: true, method: http.MethodPost, expectedOrigin: true, expectedStatus: http.StatusNoContent, }, { name: "authenticated WebSocket request", protected: true, authenticated: true, method: http.MethodGet, isWebsocket: true, expectedOrigin: true, expectedStatus: http.StatusNoContent, }, { name: "unauthenticated GET request", protected: true, method: http.MethodGet, expectedStatus: http.StatusFound, }, { name: "unauthenticated HEAD request", protected: true, method: http.MethodHead, expectedStatus: http.StatusFound, }, { name: "unauthenticated WebSocket request", protected: true, method: http.MethodGet, isWebsocket: true, expectedStatus: http.StatusFound, }, { name: "unauthenticated unsafe method", protected: true, method: http.MethodPost, expectedStatus: http.StatusUnauthorized, }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { t.Parallel()
originRequestHeaders := make(chan http.Header, 1) originServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { originRequestHeaders <- r.Header.Clone() w.Header().Set("Cache-Control", "public") w.Header().Set("Referrer-Policy", "unsafe-url") w.Header().Add("Set-Cookie", "__Host-cloudflared-qt-auth-session=origin-value; Path=/; Secure; HttpOnly; SameSite=Lax") w.Header().Add("Set-Cookie", stateCookieName+"=origin-value; Path=/; Secure; HttpOnly; SameSite=Lax") w.Header().Add("Set-Cookie", "origin-session=origin-value; Path=/; Secure; HttpOnly; SameSite=Lax") w.WriteHeader(http.StatusNoContent) })) t.Cleanup(originServer.Close)
log := zerolog.Nop() ctx, cancel := context.WithCancel(t.Context()) t.Cleanup(cancel) ingressRules := createSingleIngressConfig(t, originServer.URL) require.NoError(t, ingressRules.StartOrigins(&log, ctx.Done()))
var authorizer connection.HTTPRequestAuthorizer var sessionCookie *http.Cookie if test.protected { authorizer, sessionCookie = newTestQuickTunnelAuthorization(t, test.authenticated) } originProxy := NewOriginProxyWithHTTPRequestAuthorizer( ingressRules, nil, testTags, cfdflow.NewLimiter(0), authorizer, &log, ) responseWriter := newMockHTTPRespWriter() request := httptest.NewRequest( test.method, "https://test-tunnel.trycloudflare.com/dashboard", nil, ) if sessionCookie != nil { request.AddCookie(sessionCookie) request.AddCookie(&http.Cookie{ Name: stateCookieName, Value: "state-value", Secure: true, HttpOnly: true, SameSite: http.SameSiteLaxMode, }) }
err := originProxy.ProxyHTTP( responseWriter, tracing.NewTracedHTTPRequest(request, 0, &log), test.isWebsocket, ) require.NoError(t, err) assert.Equal(t, test.expectedStatus, responseWriter.Code)
select { case headers := <-originRequestHeaders: assert.True(t, test.expectedOrigin) for _, tag := range testTags { assert.Equal(t, tag.Value, headers.Get(TagHeaderNamePrefix+tag.Name)) } if sessionCookie != nil { assert.NotContains(t, headers.Get("Cookie"), sessionCookie.Name) assert.NotContains(t, headers.Get("Cookie"), stateCookieName) } if test.isWebsocket { assert.Equal(t, "websocket", headers.Get("Upgrade")) } default: assert.False(t, test.expectedOrigin) }
responseHeaders := responseWriter.Header() if test.expectedOrigin { responseHeaders = responseWriter.writtenHeaders } if test.protected { assert.Equal(t, "private, no-store", responseHeaders.Get("Cache-Control")) assert.Equal(t, "no-referrer", responseHeaders.Get("Referrer-Policy")) if test.expectedOrigin { assert.Equal(t, []string{ "origin-session=origin-value; Path=/; Secure; HttpOnly; SameSite=Lax", }, responseHeaders.Values("Set-Cookie")) } } else { assert.Equal(t, "public", responseHeaders.Get("Cache-Control")) assert.Equal(t, "unsafe-url", responseHeaders.Get("Referrer-Policy")) assert.Len(t, responseHeaders.Values("Set-Cookie"), 3) } }) }}
// Both HTTP/2 and QUIC route TCP requests through ProxyTCP.func TestQuickTunnelAuthorizationRejectsTCPTransport(t *testing.T) { t.Parallel()
authorizer, _ := newTestQuickTunnelAuthorization(t, false) var logOutput bytes.Buffer log := zerolog.New(&logOutput) originProxy := NewOriginProxyWithHTTPRequestAuthorizer( ingress.Ingress{}, nil, nil, nil, authorizer, &log, )
err := originProxy.ProxyTCP(t.Context(), nil, &connection.TCPRequest{}) require.ErrorIs(t, err, errHTTPAuthorizationUnsupportedTransport) assert.Contains(t, logOutput.String(), `"authOutcome":"unsupported_tcp_transport"`)}
func newTestQuickTunnelAuthorization( t *testing.T, authenticated bool,) (connection.HTTPRequestAuthorizer, *http.Cookie) { t.Helper()
stateManager, err := quicktunnelauth.NewQuickTunnelAuthStateManager("test-tunnel.trycloudflare.com") require.NoError(t, err) assertionValidator, err := quicktunnelauth.NewQuickTunnelAuthAssertionValidator() require.NoError(t, err) t.Cleanup(assertionValidator.Close) sessionManager, err := quicktunnelauth.NewQuickTunnelAuthSessionManager() require.NoError(t, err) recipientPolicy, err := quicktunnelauth.NewQuickTunnelAuthRecipientPolicy([]string{"visitor@example.com"}) require.NoError(t, err) handler, err := quicktunnelauth.NewQuickTunnelAuthHandlerWithAuthorization( stateManager, assertionValidator, sessionManager, recipientPolicy, ) require.NoError(t, err)
if !authenticated { return handler, nil } sessionCookie, err := sessionManager.IssueSession(time.Now().Add(time.Hour)) require.NoError(t, err) return handler, sessionCookie}
func TestProxySingleOrigin(t *testing.T) { log := zerolog.Nop()
ctx, cancel := context.WithCancel(t.Context())
flagSet := flag.NewFlagSet(t.Name(), flag.PanicOnError) flagSet.Bool("hello-world", true, "")
cliCtx := cli.NewContext(cli.NewApp(), flagSet, nil) err := cliCtx.Set("hello-world", "true") require.NoError(t, err)
ingressRule, err := ingress.ParseIngressFromConfigAndCLI(&config.Configuration{}, cliCtx, &log) require.NoError(t, err)
require.NoError(t, ingressRule.StartOrigins(&log, ctx.Done()))
originDialer := ingress.NewOriginDialer(ingress.OriginConfig{ DefaultDialer: testDefaultDialer, TCPWriteTimeout: 1 * time.Second, }, &log)
proxy := NewOriginProxy(ingressRule, originDialer, testTags, cfdflow.NewLimiter(0), &log) t.Run("testProxyHTTP", testProxyHTTP(proxy)) t.Run("testProxyWebsocket", testProxyWebsocket(proxy)) t.Run("testProxySSE", testProxySSE(proxy)) cancel()}
func TestProxyHTTPRequestCancelsBeforeClosingHTTP2Response(t *testing.T) { t.Parallel()
body := newNonInterruptingRequestBody() origin := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { // Do not respond until the HTTP/2 request writer is blocked in Body.Read. <-body.readStarted w.WriteHeader(http.StatusOK) })) origin.EnableHTTP2 = true origin.StartTLS() t.Cleanup(origin.Close)
req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, origin.URL, body) require.NoError(t, err)
responseWriter := newMockHTTPRespWriter() log := zerolog.Nop() proxy := &Proxy{} proxyResult := make(chan error, 1) go func() { proxyResult <- proxy.proxyHTTPRequest( responseWriter, tracing.NewTracedHTTPRequest(req, 0, &log), origin.Client().Transport, false, false, &log, ) }()
select { case err := <-proxyResult: require.NoError(t, err) case <-time.After(5 * time.Second): t.Fatal("proxy remained blocked while closing the HTTP/2 response body") } require.Equal(t, http.StatusOK, responseWriter.Code)
select { case <-body.closeCalled: case <-time.After(time.Second): t.Fatal("HTTP/2 transport did not close the request body") }
select { case <-body.readDone: t.Fatal("non-interrupting request body unexpectedly stopped reading") default: // The proxy returned through the canceled context rather than waiting // for this deliberately broken request body. }
close(body.releaseRead) select { case <-body.readDone: case <-time.After(time.Second): t.Fatal("request body read did not finish after test cleanup") }}
func testProxyHTTP(proxy connection.OriginProxy) func(t *testing.T) { return func(t *testing.T) { responseWriter := newMockHTTPRespWriter() req, err := http.NewRequest(http.MethodGet, "http://localhost:8080", nil) require.NoError(t, err)
log := zerolog.Nop() err = proxy.ProxyHTTP(responseWriter, tracing.NewTracedHTTPRequest(req, 0, &log), false) require.NoError(t, err) for _, tag := range testTags { assert.Equal(t, tag.Value, req.Header.Get(TagHeaderNamePrefix+tag.Name)) }
assert.Equal(t, http.StatusOK, responseWriter.Code) }}
func testProxyWebsocket(proxy connection.OriginProxy) func(t *testing.T) { return func(t *testing.T) { // WSRoute is a websocket echo handler const testTimeout = 5 * time.Second * 1000 ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() readPipe, writePipe := io.Pipe() req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://localhost:8080%s", hello.WSRoute), readPipe) req.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==") req.Header.Set("Connection", "Upgrade") req.Header.Set("Upgrade", "websocket") responseWriter := newMockWSRespWriter(nil)
finished := make(chan struct{})
errGroup, ctx := errgroup.WithContext(ctx) errGroup.Go(func() error { log := zerolog.Nop() err = proxy.ProxyHTTP(responseWriter, tracing.NewTracedHTTPRequest(req, 0, &log), true) require.NoError(t, err)
require.Equal(t, http.StatusSwitchingProtocols, responseWriter.Code) return nil })
errGroup.Go(func() error { select { case <-finished: case <-ctx.Done(): } if ctx.Err() == context.DeadlineExceeded { t.Errorf("Test timed out") _ = readPipe.Close() _ = writePipe.Close() _ = responseWriter.Close() } return nil })
msg := []byte("test websocket") err = wsutil.WriteClientText(writePipe, msg) require.NoError(t, err)
// ReadServerText reads next data message from rw, considering that caller represents proxy side. returnedMsg, err := wsutil.ReadServerText(responseWriter.respBody()) require.NoError(t, err) require.Equal(t, msg, returnedMsg)
err = wsutil.WriteClientBinary(writePipe, msg) require.NoError(t, err)
returnedMsg, err = wsutil.ReadServerBinary(responseWriter.respBody()) require.NoError(t, err) require.Equal(t, msg, returnedMsg)
_ = readPipe.Close() _ = writePipe.Close() _ = responseWriter.Close()
close(finished) _ = errGroup.Wait() }}
func testProxySSE(proxy connection.OriginProxy) func(t *testing.T) { return func(t *testing.T) { var ( pushCount = 50 pushFreq = time.Millisecond * 10 ) responseWriter := newMockSSERespWriter() ctx, cancel := context.WithCancel(t.Context()) req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://localhost:8080%s?freq=%s", hello.SSERoute, pushFreq), nil) require.NoError(t, err)
var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() log := zerolog.Nop() err = proxy.ProxyHTTP(responseWriter, tracing.NewTracedHTTPRequest(req, 0, &log), false) require.Equal(t, "context canceled", err.Error())
require.Equal(t, http.StatusOK, responseWriter.Code) }()
for i := 0; i < pushCount; i++ { line := responseWriter.ReadBytes() expect := fmt.Sprintf("%d\n\n", i) require.Equal(t, []byte(expect), line, "Expect to read %v, got %v", expect, line) }
cancel() wg.Wait() }}
// Regression test to guarantee that we always write the contents downstream even if EOF is reached without// hitting the delimiterfunc TestProxySSEAllData(t *testing.T) { eyeballReader := io.NopCloser(strings.NewReader("data\r\r")) responseWriter := newMockSSERespWriter()
// responseWriter uses an unbuffered channel, so we call in a different go-routine go func() { _, _ = cfio.Copy(responseWriter, eyeballReader) }()
result := string(<-responseWriter.writeNotification) require.Equal(t, "data\r\r", result)}
func TestProxyMultipleOrigins(t *testing.T) { api := httptest.NewServer(mockAPI{}) defer api.Close()
unvalidatedIngress := []config.UnvalidatedIngressRule{ { Hostname: "api.example.com", Service: api.URL, }, { Hostname: "hello.example.com", Service: "hello-world", }, { Hostname: "health.example.com", Path: "/health", Service: "http_status:200", }, { Hostname: "*", Service: "http_status:404", }, }
tests := []MultipleIngressTest{ { url: "http://api.example.com", expectedStatus: http.StatusCreated, expectedBody: []byte("Created"), }, { url: fmt.Sprintf("http://hello.example.com%s", hello.HealthRoute), expectedStatus: http.StatusOK, expectedBody: []byte("ok"), }, { url: "http://health.example.com/health", expectedStatus: http.StatusOK, }, { url: "http://health.example.com/", expectedStatus: http.StatusNotFound, }, { url: "http://not-found.example.com", expectedStatus: http.StatusNotFound, }, }
runIngressTestScenarios(t, unvalidatedIngress, tests)}
type MultipleIngressTest struct { url string expectedStatus int expectedBody []byte}
func runIngressTestScenarios(t *testing.T, unvalidatedIngress []config.UnvalidatedIngressRule, tests []MultipleIngressTest) { ingressRule, err := ingress.ParseIngress(&config.Configuration{ TunnelID: t.Name(), Ingress: unvalidatedIngress, }) require.NoError(t, err)
log := zerolog.Nop()
ctx, cancel := context.WithCancel(t.Context()) require.NoError(t, ingressRule.StartOrigins(&log, ctx.Done()))
originDialer := ingress.NewOriginDialer(ingress.OriginConfig{ DefaultDialer: testDefaultDialer, TCPWriteTimeout: 1 * time.Second, }, &log)
proxy := NewOriginProxy(ingressRule, originDialer, testTags, cfdflow.NewLimiter(0), &log)
for _, test := range tests { responseWriter := newMockHTTPRespWriter() req, err := http.NewRequest(http.MethodGet, test.url, nil) require.NoError(t, err)
err = proxy.ProxyHTTP(responseWriter, tracing.NewTracedHTTPRequest(req, 0, &log), false) require.NoError(t, err)
assert.Equal(t, test.expectedStatus, responseWriter.Code) if test.expectedBody != nil { assert.Equal(t, test.expectedBody, responseWriter.Body.Bytes()) } else { assert.Equal(t, 0, responseWriter.Body.Len()) } } cancel()}
type mockAPI struct{}
func (ma mockAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusCreated) _, _ = w.Write([]byte("Created"))}
type errorOriginTransport struct{}
func (errorOriginTransport) RoundTrip(*http.Request) (*http.Response, error) { return nil, fmt.Errorf("Proxy error")}
func TestProxyError(t *testing.T) { ing := ingress.Ingress{ Rules: []ingress.Rule{ { Hostname: "*", Path: nil, Service: ingress.MockOriginHTTPService{ Transport: errorOriginTransport{}, }, }, }, }
log := zerolog.Nop()
originDialer := ingress.NewOriginDialer(ingress.OriginConfig{ DefaultDialer: testDefaultDialer, TCPWriteTimeout: 1 * time.Second, }, &log)
proxy := NewOriginProxy(ing, originDialer, testTags, cfdflow.NewLimiter(0), &log)
responseWriter := newMockHTTPRespWriter() req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1", nil) require.NoError(t, err)
require.Error(t, proxy.ProxyHTTP(responseWriter, tracing.NewTracedHTTPRequest(req, 0, &log), false))}
type replayer struct { sync.RWMutex rw *bytes.Buffer}
func (r *replayer) Read(p []byte) (int, error) { r.RLock() defer r.RUnlock() return r.rw.Read(p)}
func (r *replayer) Write(p []byte) (int, error) { r.Lock() defer r.Unlock() n, err := r.rw.Write(p) return n, err}
func (r *replayer) String() string { r.Lock() defer r.Unlock() return r.rw.String()}
func (r *replayer) Bytes() []byte { r.Lock() defer r.Unlock() return r.rw.Bytes()}
// TestConnections tests every possible permutation of connection protocols// proxied by cloudflared.//// WS - WS : When a websocket based ingress is configured on the origin and// the eyeball is also a websocket client streaming data.// TCP - TCP : When teamnet is enabled and an http or tcp service is running// on the origin.// TCP - WS: When teamnet is enabled and a websocket based service is running// on the origin.// WS - TCP: When a tcp based ingress is configured on the origin and the// eyeball sends tcp packets wrapped in websockets. (E.g: cloudflared access).func TestConnections(t *testing.T) { log := zerolog.Nop() replayer := &replayer{rw: bytes.NewBuffer([]byte{})} type args struct { ingressServiceScheme string originService func(*testing.T, net.Listener) eyeballResponseWriter connection.ResponseWriter eyeballRequestBody io.ReadCloser
// eyeball connection type. connectionType connection.Type
// requestheaders to be sent in the call to proxy.Proxy requestHeaders http.Header
// flowLimiterResponse is the response of the cfdflow.Limiter#Acquire method call flowLimiterResponse error }
originDialer := ingress.NewOriginDialer(ingress.OriginConfig{ DefaultDialer: testDefaultDialer, TCPWriteTimeout: 0, }, &log)
type want struct { message []byte headers http.Header err bool }
tests := []struct { name string args args want want }{ { name: "ws-ws proxy", args: args{ ingressServiceScheme: "ws://", originService: runEchoWSService, eyeballResponseWriter: newWSRespWriter(replayer), eyeballRequestBody: newWSRequestBody([]byte("test1")), connectionType: connection.TypeWebsocket, requestHeaders: map[string][]string{ // Example key from https://tools.ietf.org/html/rfc6455#section-1.2 "Sec-Websocket-Key": {"dGhlIHNhbXBsZSBub25jZQ=="}, "Test-Cloudflared-Echo": {"Echo"}, }, }, want: want{ message: []byte("echo-test1"), headers: map[string][]string{ "Connection": {"Upgrade"}, "Sec-Websocket-Accept": {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="}, "Upgrade": {"websocket"}, "Test-Cloudflared-Echo": {"Echo"}, }, }, }, { name: "tcp-tcp proxy", args: args{ ingressServiceScheme: "tcp://", originService: runEchoTCPService, eyeballResponseWriter: newTCPRespWriter(replayer), eyeballRequestBody: newTCPRequestBody([]byte("test2")), connectionType: connection.TypeTCP, requestHeaders: map[string][]string{ "Cf-Cloudflared-Proxy-Src": {"non-blank-value"}, }, }, want: want{ message: []byte("echo-test2"), headers: http.Header{}, }, }, { name: "tcp-ws proxy", args: args{ ingressServiceScheme: "ws://", originService: runEchoWSService, // eyeballResponseWriter gets set after roundtrip dial. eyeballRequestBody: newPipedWSRequestBody([]byte("test3")), requestHeaders: map[string][]string{ "Cf-Cloudflared-Proxy-Src": {"non-blank-value"}, }, connectionType: connection.TypeTCP, }, want: want{ message: []byte("echo-test3"), // We expect no headers here because they are sent back via // the stream. headers: http.Header{}, }, }, { name: "ws-tcp proxy", args: args{ ingressServiceScheme: "tcp://", originService: runEchoTCPService, eyeballResponseWriter: newWSRespWriter(replayer), eyeballRequestBody: newWSRequestBody([]byte("test4")), connectionType: connection.TypeWebsocket, requestHeaders: map[string][]string{ // Example key from https://tools.ietf.org/html/rfc6455#section-1.2 "Sec-Websocket-Key": {"dGhlIHNhbXBsZSBub25jZQ=="}, }, }, want: want{ message: []byte("echo-test4"), headers: map[string][]string{ "Connection": {"Upgrade"}, "Sec-Websocket-Accept": {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="}, "Upgrade": {"websocket"}, }, }, }, { // Send (unexpected) HTTP when origin expects WS (to unwrap for raw TCP) name: "http-(ws)tcp proxy", args: args{ ingressServiceScheme: "tcp://", originService: runEchoTCPService, eyeballResponseWriter: newMockHTTPRespWriter(), eyeballRequestBody: http.NoBody, connectionType: connection.TypeHTTP, requestHeaders: map[string][]string{ "Cf-Cloudflared-Proxy-Src": {"non-blank-value"}, }, }, want: want{ message: []byte{}, headers: map[string][]string{}, }, }, { name: "ws-ws proxy when origin is different", args: args{ ingressServiceScheme: "ws://", originService: runEchoWSService, eyeballResponseWriter: newWSRespWriter(replayer), eyeballRequestBody: newWSRequestBody([]byte("test1")), connectionType: connection.TypeWebsocket, requestHeaders: map[string][]string{ // Example key from https://tools.ietf.org/html/rfc6455#section-1.2 "Sec-Websocket-Key": {"dGhlIHNhbXBsZSBub25jZQ=="}, "Origin": {"Different origin"}, }, }, want: want{ message: []byte("Forbidden\n"), err: false, headers: map[string][]string{ "Content-Length": {"10"}, "Content-Type": {"text/plain; charset=utf-8"}, "Sec-Websocket-Version": {"13"}, "X-Content-Type-Options": {"nosniff"}, }, }, }, { name: "tcp-* proxy when origin service has already closed the connection/ is no longer running", args: args{ ingressServiceScheme: "tcp://", originService: func(t *testing.T, ln net.Listener) { // closing the listener created by the test. _ = ln.Close() }, eyeballResponseWriter: newTCPRespWriter(replayer), eyeballRequestBody: newTCPRequestBody([]byte("test2")), connectionType: connection.TypeTCP, requestHeaders: map[string][]string{ "Cf-Cloudflared-Proxy-Src": {"non-blank-value"}, }, }, want: want{ message: []byte{}, err: true, }, }, { name: "tcp-* proxy rate limited flow", args: args{ ingressServiceScheme: "tcp://", originService: runEchoTCPService, eyeballResponseWriter: newTCPRespWriter(replayer), eyeballRequestBody: newTCPRequestBody([]byte("rate-limited")), connectionType: connection.TypeTCP, requestHeaders: map[string][]string{ "Cf-Cloudflared-Proxy-Src": {"non-blank-value"}, }, flowLimiterResponse: cfdflow.ErrTooManyActiveFlows, }, want: want{ message: []byte{}, err: true, }, }, }
for _, test := range tests { t.Run(test.name, func(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) ln, err := net.Listen("tcp", "127.0.0.1:0") require.NoError(t, err) // Starts origin service test.args.originService(t, ln)
ingressRule := createSingleIngressConfig(t, test.args.ingressServiceScheme+ln.Addr().String()) _ = ingressRule.StartOrigins(&log, ctx.Done())
// Mock flow limiter ctrl := gomock.NewController(t) defer ctrl.Finish() flowLimiter := mocks.NewMockLimiter(ctrl) flowLimiter.EXPECT().Acquire("tcp").AnyTimes().Return(test.args.flowLimiterResponse) flowLimiter.EXPECT().Release().AnyTimes()
proxy := NewOriginProxy(ingressRule, originDialer, testTags, flowLimiter, &log)
dest := ln.Addr().String() req, err := http.NewRequest( http.MethodGet, test.args.ingressServiceScheme+ln.Addr().String(), test.args.eyeballRequestBody, ) require.NoError(t, err)
req.Header = test.args.requestHeaders respWriter := test.args.eyeballResponseWriter
if pipedReqBody, ok := test.args.eyeballRequestBody.(*pipedRequestBody); ok { respWriter = newTCPRespWriter(pipedReqBody.pipedConn) go func() { resp := pipedReqBody.roundtrip(test.args.ingressServiceScheme + ln.Addr().String()) _, _ = replayer.Write(resp) }() } if test.args.connectionType == connection.TypeTCP { rwa := connection.NewHTTPResponseReadWriterAcker(respWriter, respWriter.(http.Flusher), req) err = proxy.ProxyTCP(ctx, rwa, &connection.TCPRequest{Dest: dest}) } else { log := zerolog.Nop() err = proxy.ProxyHTTP(respWriter, tracing.NewTracedHTTPRequest(req, 0, &log), test.args.connectionType == connection.TypeWebsocket) }
cancel() require.Equal(t, test.want.err, err != nil) require.Equal(t, test.want.message, replayer.Bytes()) require.Equal(t, test.want.headers, respWriter.Header()) replayer.rw.Reset() }) }}
type requestBody struct { pw *io.PipeWriter pr *io.PipeReader}
func newWSRequestBody(data []byte) *requestBody { pr, pw := io.Pipe() go func() { _ = wsutil.WriteClientBinary(pw, data) }() return &requestBody{ pr: pr, pw: pw, }}
func newTCPRequestBody(data []byte) *requestBody { pr, pw := io.Pipe() go func() { _, _ = pw.Write(data) }() return &requestBody{ pr: pr, pw: pw, }}
func (r *requestBody) Read(p []byte) (n int, err error) { return r.pr.Read(p)}
func (r *requestBody) Close() error { _ = r.pw.Close() _ = r.pr.Close() return nil}
type pipedRequestBody struct { dialer gorillaWS.Dialer pipedConn net.Conn wsConn net.Conn messageToWrite []byte}
func newPipedWSRequestBody(data []byte) *pipedRequestBody { conn1, conn2 := net.Pipe() dialer := gorillaWS.Dialer{ NetDial: func(network, addr string) (net.Conn, error) { return conn2, nil }, } return &pipedRequestBody{ dialer: dialer, pipedConn: conn1, wsConn: conn2, messageToWrite: data, }}
func (p *pipedRequestBody) roundtrip(addr string) []byte { header := http.Header{} conn, resp, err := p.dialer.Dial(addr, header) if err != nil { panic(err) } defer func() { _ = conn.Close() }() defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusSwitchingProtocols { panic(fmt.Errorf("resp returned status code: %d", resp.StatusCode)) }
err = conn.WriteMessage(gorillaWS.TextMessage, p.messageToWrite) if err != nil { panic(err) }
_, data, err := conn.ReadMessage() if err != nil { panic(err) }
return data}
func (p *pipedRequestBody) Read(data []byte) (n int, err error) { return p.pipedConn.Read(data)}
func (p *pipedRequestBody) Close() error { return nil}
type wsRespWriter struct { w io.Writer responseHeaders http.Header code int}
// newWSRespWriter uses wsutil.WriteClientText to generate websocket frames.// and wsutil.ReadClientText to translate frames from server to byte data.// In essence, this acts as a wsClient.func newWSRespWriter(w io.Writer) *wsRespWriter { return &wsRespWriter{ w: w, }}
// Write is written to by ingress.Stream and serves as the output to the client.func (w *wsRespWriter) Write(p []byte) (int, error) { returnedMsg, err := wsutil.ReadServerBinary(bytes.NewBuffer(p)) if err != nil { // The data was not returned by a websocket connection. if err != io.ErrUnexpectedEOF { return w.w.Write(p) } } return w.w.Write(returnedMsg)}
func (w *wsRespWriter) WriteRespHeaders(status int, header http.Header) error { w.responseHeaders = header w.code = status return nil}
func (w *wsRespWriter) Flush() {}
func (w *wsRespWriter) AddTrailer(trailerName, trailerValue string) { // do nothing}
// respHeaders is a test function to read respHeadersfunc (w *wsRespWriter) Header() http.Header { // Removing indeterminstic header because it cannot be asserted. w.responseHeaders.Del("Date") return w.responseHeaders}
func (w *wsRespWriter) WriteHeader(status int) { // unused}
func (m *wsRespWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { panic("Hijack not implemented")}
type mockTCPRespWriter struct { w io.Writer responseHeaders http.Header code int}
func newTCPRespWriter(w io.Writer) *mockTCPRespWriter { return &mockTCPRespWriter{ w: w, }}
func (m *mockTCPRespWriter) Read(p []byte) (n int, err error) { return len(p), nil}
func (m *mockTCPRespWriter) Write(p []byte) (n int, err error) { return m.w.Write(p)}
func (m *mockTCPRespWriter) Flush() {}
func (m *mockTCPRespWriter) AddTrailer(trailerName, trailerValue string) { // do nothing}
func (m *mockTCPRespWriter) WriteRespHeaders(status int, header http.Header) error { m.responseHeaders = header m.code = status return nil}
// respHeaders is a test function to read respHeadersfunc (m *mockTCPRespWriter) Header() http.Header { return m.responseHeaders}
func (m *mockTCPRespWriter) WriteHeader(status int) { // do nothing}
func (m *mockTCPRespWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { panic("Hijack not implemented")}
func createSingleIngressConfig(t *testing.T, service string) ingress.Ingress { ingressConfig := &config.Configuration{ Ingress: []config.UnvalidatedIngressRule{ { Hostname: "*", Service: service, }, }, } ingressRule, err := ingress.ParseIngress(ingressConfig) require.NoError(t, err) return ingressRule}
func runEchoTCPService(t *testing.T, l net.Listener) { go func() { for { conn, err := l.Accept() if err != nil { panic(err) } defer func() { _ = conn.Close() }()
for { buf := make([]byte, 1024) size, err := conn.Read(buf) if err == io.EOF { return } data := []byte("echo-") data = append(data, buf[:size]...) _, err = conn.Write(data) if err != nil { t.Log(err) } return } } }()}
func runEchoWSService(t *testing.T, l net.Listener) { upgrader := gorillaWS.Upgrader{ ReadBufferSize: 10, WriteBufferSize: 10, }
ws := func(w http.ResponseWriter, r *http.Request) { header := make(http.Header) for k, vs := range r.Header { if k == "Test-Cloudflared-Echo" { header[k] = vs } } conn, err := upgrader.Upgrade(w, r, header) if err != nil { t.Log(err) return } defer func() { _ = conn.Close() }()
for { messageType, p, err := conn.ReadMessage() if err != nil { return } data := []byte("echo-") data = append(data, p...) if err := conn.WriteMessage(messageType, data); err != nil { return } } }
// nolint: gosec server := http.Server{ Handler: http.HandlerFunc(ws), }
go func() { err := server.Serve(l) if err != nil { panic(err) } }()}