Networking connectivity integration test using local proxy

.github/workflows/check.yml

@@ -20,7 +20,7 @@ jobs:
           distribution: 'zulu'
           java-version: ${{ matrix.java-version }}
 
-      - run: ./gradlew check testDebugUnitTestCoverage testReleaseUnitTestCoverage --profile
+      - run: ./gradlew check testDebugUnitTestCoverage testReleaseUnitTestCoverage --profile -PruntimeSecrets=USE_DUMMY_EMPTY_STRING_VALUES
         env:
           ORG_GRADLE_PROJECT_MAPBOX_DOWNLOADS_TOKEN: ${{ secrets.MAPBOX_DOWNLOADS_TOKEN }}
 
@@ -64,13 +64,6 @@ jobs:
           name: java-version-${{ matrix.java-version }}-publishing-sdk-build-reports
           path: publishing-sdk/build/reports
 
-      - uses: actions/upload-artifact@v3
-        name: Build Reports for subscribing-example-app
-        if: always()
-        with:
-          name: java-version-${{ matrix.java-version }}-subscribing-example-app-build-reports
-          path: subscribing-example-app/build/reports
-
       - uses: actions/upload-artifact@v3
         name: Build Reports for subscribing-java-testing
         if: always()

.github/workflows/docs.yml

@@ -21,7 +21,7 @@ jobs:
 
       - name: Build Documentation
         run: |
-          ./gradlew dokkaHtmlMultiModule
+          ./gradlew dokkaHtmlMultiModule -PruntimeSecrets=USE_DUMMY_EMPTY_STRING_VALUES
           ls -al build/dokka/htmlMultiModule
         env:
           ORG_GRADLE_PROJECT_MAPBOX_DOWNLOADS_TOKEN: ${{ secrets.MAPBOX_DOWNLOADS_TOKEN }}

.github/workflows/emulate.yml

@@ -32,7 +32,12 @@ jobs:
           #boot. See https://developer.android.com/studio/run/emulator-commandline#common for more
           emulator-options: -no-snapshot-save -no-window -gpu swiftshader_indirect -noaudio -no-boot-anim
           # ${{ condition && 'ifTrue' || 'ifFalse' }} is a workaround for a ternary operator https://github.com/actions/runner/issues/409#issuecomment-752775072
-          script: ${{ matrix.excludeModules && './gradlew connectedCheck -x :publishing-java-testing:connectedCheck -x :subscribing-java-testing:connectedCheck -x :publishing-example-app:connectedCheck -x :subscribing-example-app:connectedCheck' || './gradlew connectedCheck' }}
+          script: |
+            ${{
+              matrix.excludeModules
+                && './gradlew connectedCheck -x :publishing-java-testing:connectedCheck -x :subscribing-java-testing:connectedCheck -x :publishing-example-app:connectedCheck -x :subscribing-example-app:connectedCheck -PruntimeSecrets=FOR_ALL_PROJECTS_BECAUSE_WE_ARE_RUNNING_INTEGRATION_TESTS'
+                || './gradlew connectedCheck -PruntimeSecrets=FOR_ALL_PROJECTS_BECAUSE_WE_ARE_RUNNING_INTEGRATION_TESTS'
+            }}
         env:
           ORG_GRADLE_PROJECT_MAPBOX_DOWNLOADS_TOKEN: ${{ secrets.MAPBOX_DOWNLOADS_TOKEN }}
           ORG_GRADLE_PROJECT_MAPBOX_ACCESS_TOKEN: ${{ secrets.MAPBOX_ACCESS_TOKEN }}

.github/workflows/publish-github-packages.yml

@@ -37,4 +37,4 @@ jobs:
         ORG_GRADLE_PROJECT_SIGNING_KEY_BASE64: ${{ secrets.MAVEN_SIGNING_KEY_RING_FILE_BASE64 }}
       run: |
         echo "Publishing version ${{ github.event.inputs.version }} to GitHub Packages..."
-        ./gradlew publish -PpublishTarget=GitHubPackages
+        ./gradlew publish -PpublishTarget=GitHubPackages -PruntimeSecrets=USE_DUMMY_EMPTY_STRING_VALUES

.github/workflows/publish-maven-central.yml

@@ -37,4 +37,4 @@ jobs:
         ORG_GRADLE_PROJECT_SIGNING_KEY_BASE64: ${{ secrets.MAVEN_SIGNING_KEY_RING_FILE_BASE64 }}
       run: |
         echo "Publishing version ${{ github.event.inputs.version }} to MavenCentral..."
-        ./gradlew -PpublishTarget=MavenCentral publishToSonatype closeAndReleaseSonatypeStagingRepository
+        ./gradlew -PpublishTarget=MavenCentral -PruntimeSecrets=USE_DUMMY_EMPTY_STRING_VALUES publishToSonatype closeAndReleaseSonatypeStagingRepository

android-test-common/src/main/java/com/ably/tracking/test/android/common/AblyProxy.kt

@@ -0,0 +1,214 @@
+package com.ably.tracking.test.android.common
+
+import io.ably.lib.types.ClientOptions
+import io.ably.lib.util.Log
+import java.net.ServerSocket
+import java.net.Socket
+import java.net.SocketException
+import java.util.UUID
+import javax.net.ssl.SSLSocketFactory
+
+private const val AGENT_HEADER_NAME = "ably-asset-tracking-android-publisher-tests"
+
+private const val PROXY_HOST = "localhost"
+private const val PROXY_PORT = 13579
+private const val REALTIME_HOST = "realtime.ably.io"
+private const val REALTIME_PORT = 443
+
+/**
+ * A local proxy that can be used to intercept Realtime traffic for testing
+ */
+interface RealtimeProxy {
+    /**
+     * Start the proxy listening for connections
+     */
+    fun start()
+
+    /**
+     * Stop the proxy and close any active connetions
+     */
+    fun stop()
+
+    /**
+     * Ably ClientOptions that have been configured to direct traffic
+     * through this proxy service
+     */
+    val clientOptions: ClientOptions
+}
+
+/**
+ * A TCP Proxy, which can run locally and intercept traffic to Ably realtime.
+ *
+ * This proxy is only capable of simulating faults at the transport layer, such
+ * as connections being interrupted or packets being dropped entirely.
+ */
+class Layer4Proxy(
+    val listenHost: String = PROXY_HOST,
+    val listenPort: Int = PROXY_PORT,
+    private val targetAddress: String = REALTIME_HOST,
+    private val targetPort: Int = REALTIME_PORT,
+    private val apiKey: String,
+) : RealtimeProxy {
+
+    private val loggingTag = "Layer4Proxy"
+
+    private var server: ServerSocket? = null
+    private val sslSocketFactory = SSLSocketFactory.getDefault()
+    private val connections: MutableList<Layer4ProxyConnection> = mutableListOf()
+
+    /**
+     * Flag mutated by fault implementations to hang the TCP connection
+     */
+    var isForwarding = true
+
+    /**
+     * Block current thread and wait for a new incoming client connection on the server socket.
+     * Returns a connection object when a client has connected.
+     */
+    private fun accept(): Layer4ProxyConnection {
+        val clientSock = server?.accept()
+        testLogD("$loggingTag: accepted connection")
+
+        val serverSock = sslSocketFactory.createSocket(targetAddress, targetPort)
+        val conn = Layer4ProxyConnection(serverSock, clientSock!!, targetAddress, parentProxy = this)
+        connections.add(conn)
+        return conn
+    }
+
+    /**
+     * Pre-configured client options to configure AblyRealtime to send traffic locally through
+     * this proxy. Note that TLS is disabled, so that the proxy can act as a man in the middle.
+     */
+    override val clientOptions = ClientOptions().apply {
+        this.clientId = "AatTestProxy_${UUID.randomUUID()}"
+        this.agents = mapOf(AGENT_HEADER_NAME to BuildConfig.VERSION_NAME)
+        this.idempotentRestPublishing = true
+        this.autoConnect = false
+        this.key = apiKey
+        this.logHandler = Log.LogHandler { _, _, msg, tr ->
+            testLogD("${msg!!} - $tr")
+        }
+        this.realtimeHost = listenHost
+        this.port = listenPort
+        this.tls = false
+    }
+
+    /**
+     * Close open connections and stop listening for new local connections
+     */
+    override fun stop() {
+        server?.close()
+        server = null
+
+        connections.forEach {
+            it.stop()
+        }
+        connections.clear()
+    }
+
+    /**
+     * Begin a background thread listening for local Realtime connections
+     */
+    override fun start() {
+        server = ServerSocket(listenPort)
+        Thread {
+            while (true) {
+                testLogD("$loggingTag: proxy trying to accept")
+                try {
+                    val conn = this.accept()
+                    testLogD("$loggingTag: proxy starting to run")
+                    conn.run()
+                } catch (e: Exception) {
+                    testLogD("$loggingTag: proxy shutting down: " + e.message)
+                    break
+                }
+            }
+        }.start()
+    }
+}
+
+/**
+ * A TCP Proxy connection between a local client and the remote Ably service.
+ */
+internal class Layer4ProxyConnection(
+    private val server: Socket,
+    private val client: Socket,
+    private val targetHost: String,
+    private val parentProxy: Layer4Proxy
+) {
+
+    private val loggingTag = "Layer4ProxyConnection"
+
+    /**
+     * Starts two threads, one forwarding traffic in each direction between
+     * the local client and the Ably Realtime service.
+     */
+    fun run() {
+        Thread { proxy(server, client, true) }.start()
+        Thread { proxy(client, server) }.start()
+    }
+
+    /**
+     * Close socket connections, causing proxy threads to exit.
+     */
+    fun stop() {
+        try {
+            server.close()
+        } catch (e: Exception) {
+            testLogD("$loggingTag: stop() server: $e")
+        }
+
+        try {
+            client.close()
+        } catch (e: Exception) {
+            testLogD("$loggingTag: stop() client: $e")
+        }
+    }
+
+    /**
+     * Copies traffic between source and destination sockets, rewriting the
+     * HTTP host header if requested to remove the proxy host details.
+     */
+    private fun proxy(dstSock: Socket, srcSock: Socket, rewriteHost: Boolean = false) {
+        try {
+            val dst = dstSock.getOutputStream()
+            val src = srcSock.getInputStream()
+            val buff = ByteArray(4096)
+            var bytesRead: Int
+
+            // deal with the initial HTTP upgrade packet
+            bytesRead = src.read(buff)
+            if (bytesRead < 0) {
+                return
+            }
+
+            // HTTP is plaintext so we can just read it
+            val msg = String(buff, 0, bytesRead)
+            testLogD("$loggingTag: ${String(buff.copyOfRange(0, bytesRead))}")
+            if (rewriteHost) {
+                val newMsg = msg.replace(
+                    oldValue = "${parentProxy.listenHost}:${parentProxy.listenPort}",
+                    newValue = targetHost
+                )
+                val newBuff = newMsg.toByteArray()
+                dst.write(newBuff, 0, newBuff.size)
+            } else {
+                dst.write(buff, 0, bytesRead)
+            }
+
+            while (-1 != src.read(buff).also { bytesRead = it }) {
+                if (parentProxy.isForwarding) {
+                    dst.write(buff, 0, bytesRead)
+                }
+            }
+        } catch (ignored: SocketException) {
+        } catch (e: Exception) {
+            testLogD("$loggingTag: $e")
+        } finally {
+            try {
+                srcSock.close()
+            } catch (ignored: Exception) {
+            }
+        }
+    }
+}