Complete plaintext local sandbox integration support

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: 
 
 ```

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)
 }
 

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)
 	}

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

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)
 	}

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 {

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,19 +77,17 @@ 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.
 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
 }