diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 24568f9a..2cd6b4d9 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -68,6 +68,26 @@ export ABLY_PROTOCOL="application/json" && go test -tags=integration -p 1 -race export ABLY_PROTOCOL="application/x-msgpack" && go test -tags=integration -p 1 -race -v -timeout 120m ./... ``` +The integration suite can provision its app from a local compatibility +sandbox by setting `ABLY_LOCAL_SANDBOX_URL`. If that sandbox returns a +plaintext loopback child endpoint, set `ABLY_LOCAL_PLAINTEXT_REALTIME=1` to +keep the SDK's logical TLS and fallback-host behavior while tunnelling the +allowlisted test traffic to the child. The harness then uses these variables: + +- `ABLY_LOCAL_FALLBACK_HOSTS`: comma-separated logical Realtime hosts that may + be redirected to the provisioned child. +- `ABLY_LOCAL_REST_HOSTS`: comma-separated logical REST hosts that may be + redirected to the provisioned child. +- `ABLY_LOCAL_INTERNET_ENDPOINT`: loopback `host:port` for the internet-up + fallback probe. +- `ABLY_CREATE_JWT_URL`: URL of a local JWT test fixture. +- `ABLY_STATS_FIXTURE_URL`: URL of a local stats fixture. + +The plaintext routing rejects non-loopback child and internet-probe endpoints, +and blocks logical hosts that are not explicitly allowlisted. It is disabled +by default, so cloud sandbox runs and direct `ABLY_LOCAL_SANDBOX_URL` routing +remain available. + Depending on which protocol they are to be run for. It is also necessary to clean the test cache in between runs of these tests which can be done with the command: ``` diff --git a/ably/export_test.go b/ably/export_test.go index 020fa480..88f9a9d9 100644 --- a/ably/export_test.go +++ b/ably/export_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptrace" "net/url" + "sync" "time" ) @@ -365,7 +366,28 @@ func ChannelModeToFlag(mode ChannelMode) ProtoFlag { return mode.toFlag() } +var ( + websocketURLTransformMu sync.RWMutex + websocketURLTransform func(*url.URL) (*url.URL, error) +) + +func SetWebsocketURLTransform(transform func(*url.URL) (*url.URL, error)) { + websocketURLTransformMu.Lock() + defer websocketURLTransformMu.Unlock() + websocketURLTransform = transform +} + func DialWebsocket(proto string, u *url.URL, timeout time.Duration) (Conn, error) { + websocketURLTransformMu.RLock() + transform := websocketURLTransform + websocketURLTransformMu.RUnlock() + if transform != nil { + var err error + u, err = transform(u) + if err != nil { + return nil, err + } + } return dialWebsocket(proto, u, timeout, nil) } diff --git a/ably/main_integration_test.go b/ably/main_integration_test.go index 2bae8d05..43cf2af0 100644 --- a/ably/main_integration_test.go +++ b/ably/main_integration_test.go @@ -5,18 +5,56 @@ package ably_test import ( "fmt" + "net" + "net/url" "os" + "strconv" + "strings" "testing" + "github.com/ably/ably-go/ably" "github.com/ably/ably-go/internal/ablytest" ) +func localPlaintextRealtimeOption(config *ablytest.Config) ably.ClientOption { + allowedHosts := make(map[string]struct{}) + for _, host := range strings.Split(os.Getenv("ABLY_LOCAL_FALLBACK_HOSTS"), ",") { + if host = strings.TrimSpace(host); host != "" { + allowedHosts[host] = struct{}{} + } + } + endpoint := net.JoinHostPort(config.LocalEndpoint, strconv.Itoa(config.LocalPort)) + + ably.SetWebsocketURLTransform(func(u *url.URL) (*url.URL, error) { + hostname := u.Hostname() + ip := net.ParseIP(hostname) + loopback := hostname == "localhost" || (ip != nil && ip.IsLoopback()) + if !loopback { + if _, allowed := allowedHosts[hostname]; !allowed { + return nil, fmt.Errorf("refusing plaintext realtime test transport for non-loopback host %q", hostname) + } + } + + localURL := *u + localURL.Scheme = "ws" + if !loopback { + localURL.Host = endpoint + } + return &localURL, nil + }) + return ably.WithDial(ably.DialWebsocket) +} + // TestMain tears down the shared sandbox app once after all tests in this // package have run. The app itself is provisioned lazily on first use (see // ablytest.NewSandbox), so there is no setup here; if no test provisions it, // CloseSharedApp is a no-op. func TestMain(m *testing.M) { + if os.Getenv("ABLY_LOCAL_PLAINTEXT_REALTIME") == "1" { + ablytest.LocalRealtimeOption = localPlaintextRealtimeOption + } code := m.Run() + ably.SetWebsocketURLTransform(nil) if err := ablytest.CloseSharedApp(); err != nil { fmt.Fprintf(os.Stderr, "warning: failed to tear down shared sandbox app: %v\n", err) } diff --git a/ably/rest_channel_integration_test.go b/ably/rest_channel_integration_test.go index 964d94a1..f93a6e90 100644 --- a/ably/rest_channel_integration_test.go +++ b/ably/rest_channel_integration_test.go @@ -8,6 +8,7 @@ import ( "crypto/tls" "encoding/base64" "fmt" + "net" "net/http" "net/http/httptest" "net/url" @@ -317,12 +318,18 @@ func TestIdempotent_retry(t *testing.T) { } serverURL, _ := url.Parse(server.URL) - // Resolve the real destination from app.Options, not nopts alone, so the - // proxy forwards token requests and successful retries to the provisioned - // app's endpoint/port. Against a per-test local child that address is not - // derivable from the endpoint name, so a bare nopts URL would send those - // requests to the wrong server (which 404s the app id). + // Resolve the real destination from app.Options, not nopts alone. A + // plaintext compatibility harness preserves the logical endpoint in the + // client options, so its proxy destination must use the provisioned local + // child's address explicitly. defaultURL, _ := url.Parse(ably.ApplyOptionsWithDefaults(app.Options(nopts...)...).RestURL()) + if app.Config.LocalEndpoint != "" && app.Config.LocalPort != 0 { + scheme := "https" + if !app.Config.LocalTLS { + scheme = "http" + } + defaultURL, _ = url.Parse(fmt.Sprintf("%s://%s", scheme, net.JoinHostPort(app.Config.LocalEndpoint, strconv.Itoa(app.Config.LocalPort)))) + } proxy := func(r *http.Request) (*url.URL, error) { if !strings.HasPrefix(r.URL.Path, "/channels/") { // this is to handle token requests diff --git a/ably/rest_client_integration_test.go b/ably/rest_client_integration_test.go index 62d048fd..89d10eeb 100644 --- a/ably/rest_client_integration_test.go +++ b/ably/rest_client_integration_test.go @@ -15,7 +15,9 @@ import ( "net/http" "net/http/httptest" "net/url" + "os" "regexp" + "strconv" "strings" "sync" "sync/atomic" @@ -518,12 +520,19 @@ func TestRest_rememberHostFallback(t *testing.T) { } // set up the proxy to forward all requests except a specific fallback to the server, - // whilst that fallback goes to the regular endpoint. Resolve that endpoint - // from app.Options (not nopts alone) so it is the provisioned app's - // host/port — for a per-test local child that isn't derivable from the - // endpoint name, and a bare nopts URL would 404 the app id. + // whilst that fallback goes to the regular endpoint. A plaintext + // compatibility harness preserves the logical endpoint in the client + // options, so its proxy destination must use the provisioned local child's + // address explicitly. serverURL, _ := url.Parse(server.URL) defaultURL, _ := url.Parse(ably.ApplyOptionsWithDefaults(app.Options(nopts...)...).RestURL()) + if app.Config.LocalEndpoint != "" && app.Config.LocalPort != 0 { + scheme := "https" + if !app.Config.LocalTLS { + scheme = "http" + } + defaultURL, _ = url.Parse(fmt.Sprintf("%s://%s", scheme, net.JoinHostPort(app.Config.LocalEndpoint, strconv.Itoa(app.Config.LocalPort)))) + } proxy := func(r *http.Request) (*url.URL, error) { if r.URL.Hostname() == "fallback2" { @@ -802,7 +811,11 @@ func postStats(app *ablytest.Sandbox, stats []*ably.Stats) error { return fmt.Errorf("marshaling stats: %w", err) } - req, err := http.NewRequest("POST", "https://sandbox-rest.ably.io/stats", bytes.NewReader(statsJSON)) + statsURL := os.Getenv("ABLY_STATS_FIXTURE_URL") + if statsURL == "" { + statsURL = "https://sandbox-rest.ably.io/stats" + } + req, err := http.NewRequest("POST", statsURL, bytes.NewReader(statsJSON)) if err != nil { return fmt.Errorf("creating request: %w", err) } diff --git a/internal/ablytest/ablytest.go b/internal/ablytest/ablytest.go index b1373247..6ebc8cc7 100644 --- a/internal/ablytest/ablytest.go +++ b/internal/ablytest/ablytest.go @@ -104,6 +104,9 @@ func init() { if s := os.Getenv("ABLY_LOCAL_SANDBOX_URL"); s != "" { LocalSandboxURL = strings.TrimRight(s, "/") } + if s := os.Getenv("ABLY_CREATE_JWT_URL"); s != "" { + CREATE_JWT_URL = s + } } func MergeOptions(opts ...[]ably.ClientOption) []ably.ClientOption { diff --git a/internal/ablytest/recorders.go b/internal/ablytest/recorders.go index af50443d..701e890a 100644 --- a/internal/ablytest/recorders.go +++ b/internal/ablytest/recorders.go @@ -14,7 +14,7 @@ import ( // RoundTripRecorder is a http.Transport wrapper which records // HTTP request/response pairs. type RoundTripRecorder struct { - *http.Transport + http.RoundTripper mtx sync.Mutex reqs []*http.Request @@ -77,7 +77,7 @@ func (rec *RoundTripRecorder) RoundTrip(req *http.Request) (*http.Response, erro if atomic.LoadInt32(&rec.stopped) == 0 { return rec.roundTrip(req) } - return rec.Transport.RoundTrip(req) + return rec.RoundTripper.RoundTrip(req) } // Stop makes the recorder stop recording new requests/responses. @@ -85,11 +85,9 @@ func (rec *RoundTripRecorder) Stop() { atomic.StoreInt32(&rec.stopped, 1) } -// Hijack injects http.Transport into the wrapper. +// Hijack injects an HTTP transport into the wrapper. func (rec *RoundTripRecorder) Hijack(rt http.RoundTripper) http.RoundTripper { - if tr, ok := rt.(*http.Transport); ok { - rec.Transport = tr - } + rec.RoundTripper = rt return rec } @@ -106,7 +104,7 @@ func (rec *RoundTripRecorder) roundTrip(req *http.Request) (*http.Response, erro if req.Body != nil { req.Body = io.NopCloser(io.TeeReader(req.Body, &buf)) } - resp, err := rec.Transport.RoundTrip(req) + resp, err := rec.RoundTripper.RoundTrip(req) req.Body = body(buf.Bytes()) buf.Reset() if resp != nil && resp.Body != nil { @@ -115,10 +113,14 @@ func (rec *RoundTripRecorder) roundTrip(req *http.Request) (*http.Response, erro resp.Body = body(buf.Bytes()) } rec.mtx.Lock() - respCopy := *resp - respCopy.Body = body(buf.Bytes()) rec.reqs = append(rec.reqs, req) - rec.resps = append(rec.resps, &respCopy) + if resp != nil { + respCopy := *resp + respCopy.Body = body(buf.Bytes()) + rec.resps = append(rec.resps, &respCopy) + } else { + rec.resps = append(rec.resps, nil) + } rec.mtx.Unlock() return resp, err } diff --git a/internal/ablytest/sandbox.go b/internal/ablytest/sandbox.go index 6c19d870..0aba5ae7 100644 --- a/internal/ablytest/sandbox.go +++ b/internal/ablytest/sandbox.go @@ -15,7 +15,9 @@ import ( "os" "path" "path/filepath" + "reflect" "runtime" + "strconv" "strings" "sync" "sync/atomic" @@ -136,9 +138,10 @@ func PresenceFixturesCipher() ably.CipherParams { } type Sandbox struct { - Config *Config - Endpoint string - client *http.Client + Config *Config + Endpoint string + client *http.Client + localEndpoint string // local is set when this app was provisioned against a local sandbox // (see LocalSandboxURL) rather than the cloud sandbox; it selects the local @@ -146,6 +149,73 @@ type Sandbox struct { local bool } +// LocalRealtimeOption returns the client option used to route a plaintext +// local sandbox's Realtime connections. Integration test packages can set it +// to a test-only dialer without adding local-harness behavior to the SDK. +var LocalRealtimeOption func(*Config) ably.ClientOption + +func loopbackEndpoint(host string, port int) (string, error) { + ip := net.ParseIP(host) + if host != "localhost" && (ip == nil || !ip.IsLoopback()) { + return "", fmt.Errorf("local sandbox endpoint %q is not loopback", host) + } + if port < 1 || port > 65535 { + return "", fmt.Errorf("local sandbox port %d is invalid", port) + } + return net.JoinHostPort(host, strconv.Itoa(port)), nil +} + +type localPlaintextTransport struct { + base http.RoundTripper + endpoint string +} + +func (transport localPlaintextTransport) RoundTrip(request *http.Request) (*http.Response, error) { + originalRequest := request + hostname := request.URL.Hostname() + ip := net.ParseIP(hostname) + allowed := hostname == "localhost" || (ip != nil && (ip.IsLoopback() || ip.IsUnspecified())) + for _, localHost := range strings.Split(os.Getenv("ABLY_LOCAL_REST_HOSTS"), ",") { + if strings.TrimSpace(localHost) == hostname { + allowed = true + break + } + } + if !allowed { + return nil, &net.DNSError{Err: "blocked by local compatibility harness", Name: hostname, IsNotFound: true} + } + request = request.Clone(request.Context()) + request.Host = request.URL.Host + request.URL.Scheme = "http" + if ip == nil || (!ip.IsLoopback() && !ip.IsUnspecified()) { + request.URL.Host = transport.endpoint + } + if hostname == "internet-up.ably-realtime.com" { + endpointURL, err := url.Parse("http://" + strings.TrimSpace(os.Getenv("ABLY_LOCAL_INTERNET_ENDPOINT"))) + if err != nil || endpointURL.Hostname() == "" { + return nil, errors.New("ABLY_LOCAL_INTERNET_ENDPOINT is required for the local internet probe") + } + endpoint, err := loopbackEndpoint(endpointURL.Hostname(), urlPort(endpointURL)) + if err != nil { + return nil, fmt.Errorf("invalid ABLY_LOCAL_INTERNET_ENDPOINT: %w", err) + } + request.URL.Host = endpoint + } + response, err := transport.base.RoundTrip(request) + if response != nil { + response.Request = originalRequest + } + return response, err +} + +func urlPort(u *url.URL) int { + value, err := strconv.Atoi(u.Port()) + if err != nil { + return 0 + } + return value +} + func NewRealtime(opts ...ably.ClientOption) (*Sandbox, *ably.Realtime) { app := MustSandbox() client, err := ably.NewRealtime(app.Options(opts...)...) @@ -213,7 +283,6 @@ func provisionSandbox(endpoint string) (*Sandbox, error) { client: NewHTTPClient(), local: LocalSandboxURL != "", } - p := []byte(loadAppSetup().PostApps) const RetryCount = 4 @@ -253,6 +322,16 @@ func provisionSandbox(endpoint string) (*Sandbox, error) { if err := json.NewDecoder(resp.Body).Decode(app.Config); err != nil { return nil, err } + if app.local && !app.Config.LocalTLS && LocalRealtimeOption != nil { + app.localEndpoint, err = loopbackEndpoint(app.Config.LocalEndpoint, app.Config.LocalPort) + if err != nil { + return nil, err + } + app.client.Transport = localPlaintextTransport{ + base: app.client.Transport, + endpoint: app.localEndpoint, + } + } return app, nil } } @@ -332,6 +411,13 @@ func (app *Sandbox) Options(opts ...ably.ClientOption) []ably.ClientOption { Hijack(http.RoundTripper) http.RoundTripper } appHTTPClient := NewHTTPClient() + logicalLocalRouting := app.local && !app.Config.LocalTLS && LocalRealtimeOption != nil + if logicalLocalRouting { + appHTTPClient.Transport = localPlaintextTransport{ + base: appHTTPClient.Transport, + endpoint: app.localEndpoint, + } + } appOpts := []ably.ClientOption{ ably.WithKey(app.Key()), ably.WithEndpoint(app.Endpoint), @@ -340,17 +426,25 @@ func (app *Sandbox) Options(opts ...ably.ClientOption) []ably.ClientOption { ably.WithLogLevel(DefaultLogLevel), } - // local sandbox: route to the app's child server (its own host/port, - // plain ws/http), overriding the cloud endpoint set above. Basic auth is - // allowed without TLS since the child terminates plaintext. + // Keep the logical endpoint and TLS semantics when a test-only local + // Realtime dialer is installed. Otherwise retain the direct local sandbox + // routing used by ABLY_LOCAL_SANDBOX_URL. if app.local { - appOpts = append(appOpts, - ably.WithEndpoint(app.Config.LocalEndpoint), - ably.WithTLS(app.Config.LocalTLS), - ably.WithPort(app.Config.LocalPort), - ) - if !app.Config.LocalTLS { - appOpts = append(appOpts, ably.WithInsecureAllowBasicAuthWithoutTLS()) + if logicalLocalRouting { + appOpts = append(appOpts, + ably.WithTLS(true), + LocalRealtimeOption(app.Config), + ) + } else { + appOpts = append(appOpts, + ably.WithEndpoint(app.Config.LocalEndpoint), + ably.WithTLS(app.Config.LocalTLS), + ably.WithPort(app.Config.LocalPort), + ably.WithTLSPort(app.Config.LocalPort), + ) + if !app.Config.LocalTLS { + appOpts = append(appOpts, ably.WithInsecureAllowBasicAuthWithoutTLS()) + } } } @@ -360,6 +454,17 @@ func (app *Sandbox) Options(opts ...ably.ClientOption) []ably.ClientOption { if hijacker, ok := httpClient.Transport.(transportHijacker); ok { appHTTPClient.Transport = hijacker.Hijack(appHTTPClient.Transport) opts = append(opts, ably.WithHTTPClient(appHTTPClient)) + } else if logicalLocalRouting { + if transport, ok := httpClient.Transport.(*http.Transport); ok && + (transport.Proxy == nil || + reflect.ValueOf(transport.Proxy).Pointer() == reflect.ValueOf(http.ProxyFromEnvironment).Pointer()) { + wrapped := *httpClient + wrapped.Transport = localPlaintextTransport{ + base: httpClient.Transport, + endpoint: app.localEndpoint, + } + opts = append(opts, ably.WithHTTPClient(&wrapped)) + } } } appOpts = MergeOptions(appOpts, opts) diff --git a/internal/ablytest/sandbox_test.go b/internal/ablytest/sandbox_test.go new file mode 100644 index 00000000..dbbd0fee --- /dev/null +++ b/internal/ablytest/sandbox_test.go @@ -0,0 +1,104 @@ +//go:build !integration +// +build !integration + +package ablytest + +import ( + "io" + "net" + "net/http" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (fn roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { + return fn(request) +} + +func TestLoopbackEndpoint(t *testing.T) { + for _, host := range []string{"localhost", "127.0.0.1", "::1"} { + endpoint, err := loopbackEndpoint(host, 7100) + require.NoError(t, err) + assert.Equal(t, net.JoinHostPort(host, "7100"), endpoint) + } + + _, err := loopbackEndpoint("example.com", 7100) + require.Error(t, err) + assert.Contains(t, err.Error(), "is not loopback") + _, err = loopbackEndpoint("127.0.0.1", 0) + require.Error(t, err) + assert.Contains(t, err.Error(), "port 0 is invalid") +} + +func TestLocalPlaintextTransportRoutesAllowedHostsToLoopback(t *testing.T) { + t.Setenv("ABLY_LOCAL_REST_HOSTS", "primary.example") + original, err := http.NewRequest(http.MethodGet, "https://primary.example/path", nil) + require.NoError(t, err) + + transport := localPlaintextTransport{ + endpoint: "127.0.0.1:7100", + base: roundTripFunc(func(request *http.Request) (*http.Response, error) { + assert.Equal(t, "http://127.0.0.1:7100/path", request.URL.String()) + assert.Equal(t, "primary.example", request.Host) + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("ok")), + Request: request, + }, nil + }), + } + + response, err := transport.RoundTrip(original) + require.NoError(t, err) + assert.Same(t, original, response.Request) +} + +func TestLocalPlaintextTransportRejectsUnlistedHosts(t *testing.T) { + t.Setenv("ABLY_LOCAL_REST_HOSTS", "primary.example") + request, err := http.NewRequest(http.MethodGet, "https://external.example/path", nil) + require.NoError(t, err) + + transport := localPlaintextTransport{base: http.DefaultTransport, endpoint: "127.0.0.1:7100"} + _, err = transport.RoundTrip(request) + var dnsError *net.DNSError + assert.ErrorAs(t, err, &dnsError) +} + +func TestRoundTripRecorderWrapsAnyTransport(t *testing.T) { + recorder := &RoundTripRecorder{} + transport := roundTripFunc(func(request *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusNoContent, + Body: io.NopCloser(strings.NewReader("")), + Request: request, + }, nil + }) + recorder.Hijack(transport) + + request, err := http.NewRequest(http.MethodGet, "https://example.com", nil) + require.NoError(t, err) + response, err := recorder.RoundTrip(request) + require.NoError(t, err) + assert.Equal(t, http.StatusNoContent, response.StatusCode) + assert.Equal(t, 1, recorder.Len()) +} + +func TestRoundTripRecorderHandlesTransportErrorsWithoutAResponse(t *testing.T) { + recorder := &RoundTripRecorder{} + recorder.Hijack(roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, assert.AnError + })) + + request, err := http.NewRequest(http.MethodGet, "https://example.com", nil) + require.NoError(t, err) + response, err := recorder.RoundTrip(request) + assert.Nil(t, response) + assert.Equal(t, assert.AnError, err) + assert.Equal(t, 1, recorder.Len()) + assert.Nil(t, recorder.Response(0)) +}