From 17a4af8df38191a6724f98d428af7d4ea7836412 Mon Sep 17 00:00:00 2001 From: Oscar Mira Date: Sun, 3 May 2026 14:09:10 +0200 Subject: [PATCH 1/5] Add Flow.zipWithPrevious extension --- .../src/main/java/org/signal/core/util/FlowExtensions.kt | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/core/util-jvm/src/main/java/org/signal/core/util/FlowExtensions.kt b/core/util-jvm/src/main/java/org/signal/core/util/FlowExtensions.kt index e013909b639..0b60964f025 100644 --- a/core/util-jvm/src/main/java/org/signal/core/util/FlowExtensions.kt +++ b/core/util-jvm/src/main/java/org/signal/core/util/FlowExtensions.kt @@ -10,6 +10,7 @@ import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.channelFlow import kotlinx.coroutines.flow.conflate import kotlinx.coroutines.flow.filterNot +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.onEach import kotlin.time.Duration @@ -35,3 +36,11 @@ fun Flow.throttleLatest(timeout: Duration, emitImmediately: (T) -> Boolea } } } + +fun Flow.zipWithPrevious(transform: suspend (T?, T) -> R): Flow = flow { + var prev: T? = null + collect { cur -> + emit(transform(prev, cur)) + prev = cur + } +} From 31e56edf9fd3cef8e69a87a207415a046b810126 Mon Sep 17 00:00:00 2001 From: Oscar Mira Date: Mon, 4 May 2026 22:15:27 +0200 Subject: [PATCH 2/5] Refactor network connection listener Replace NetworkConnectionListener with InternetConnectivityMonitor, which uses a coroutine-based Flow to aggregate per-network state and derive a single ConnectivityState. The monitor tracks all networks instead of relying on the default network callback, so VPN and underlying networks are considered. Co-authored-by: S1m <31284753+p1gp1g@users.noreply.github.com> --- .../securesms/dependencies/AppDependencies.kt | 12 - .../messages/IncomingMessageObserver.kt | 42 +-- .../messages/NetworkConnectionListener.kt | 159 ---------- .../net/InternetConnectivityMonitor.kt | 284 ++++++++++++++++++ .../messages/NetworkConnectionListenerTest.kt | 59 ---- 5 files changed, 305 insertions(+), 251 deletions(-) delete mode 100644 app/src/main/java/org/thoughtcrime/securesms/messages/NetworkConnectionListener.kt create mode 100644 app/src/main/java/org/thoughtcrime/securesms/net/InternetConnectivityMonitor.kt delete mode 100644 app/src/test/java/org/thoughtcrime/securesms/messages/NetworkConnectionListenerTest.kt diff --git a/app/src/main/java/org/thoughtcrime/securesms/dependencies/AppDependencies.kt b/app/src/main/java/org/thoughtcrime/securesms/dependencies/AppDependencies.kt index 32307e74403..ee1e68b80fa 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/dependencies/AppDependencies.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/dependencies/AppDependencies.kt @@ -12,7 +12,6 @@ import org.signal.core.util.billing.BillingApi import org.signal.core.util.concurrent.DeadlockDetector import org.signal.core.util.concurrent.LatestValueObservable import org.signal.core.util.contentproviders.BlobProvider -import org.signal.core.util.orNull import org.signal.core.util.resettableLazy import org.signal.donations.permits.DonationPermitsRepository import org.signal.glide.SignalGlideDependencies @@ -83,7 +82,6 @@ import org.whispersystems.signalservice.api.services.ProfileService import org.whispersystems.signalservice.api.storage.StorageServiceApi import org.whispersystems.signalservice.api.websocket.SignalWebSocket import org.whispersystems.signalservice.api.websocket.WebSocketConnectionState -import org.whispersystems.signalservice.internal.configuration.HttpProxy import org.whispersystems.signalservice.internal.configuration.SignalServiceConfiguration import org.whispersystems.signalservice.internal.push.PushServiceSocket import java.util.function.Supplier @@ -448,16 +446,6 @@ object AppDependencies { networkModule.openConnections() } - fun onSystemHttpProxyChange(systemHttpProxy: HttpProxy?): Boolean { - val currentSystemProxy = signalServiceNetworkAccess.getConfiguration().systemHttpProxy.orNull() - return if (currentSystemProxy?.host != systemHttpProxy?.host || currentSystemProxy?.port != systemHttpProxy?.port) { - resetNetwork() - true - } else { - false - } - } - interface Provider { fun providePushServiceSocket(signalServiceConfiguration: SignalServiceConfiguration, groupsV2Operations: GroupsV2Operations): PushServiceSocket fun provideSignalRestClient(signalServiceConfiguration: SignalServiceConfiguration): SignalRestClient diff --git a/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt b/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt index 07c72616be7..1c2936ffa94 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt @@ -26,7 +26,6 @@ import org.thoughtcrime.securesms.groups.GroupsV2ProcessingLock import org.thoughtcrime.securesms.groups.v2.processing.GroupsV2StateProcessor import org.thoughtcrime.securesms.jobmanager.Job import org.thoughtcrime.securesms.jobmanager.impl.BackoffUtil -import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint import org.thoughtcrime.securesms.jobs.PushProcessMessageErrorJob import org.thoughtcrime.securesms.jobs.PushProcessMessageJob import org.thoughtcrime.securesms.jobs.RequestGroupV2InfoJob @@ -34,8 +33,9 @@ import org.thoughtcrime.securesms.keyvalue.SignalStore import org.thoughtcrime.securesms.keyvalue.isDecisionPending import org.thoughtcrime.securesms.messages.MessageDecryptor.FollowUpOperation import org.thoughtcrime.securesms.messages.protocol.BufferedProtocolStore +import org.thoughtcrime.securesms.net.ConnectivityState +import org.thoughtcrime.securesms.net.InternetConnectivityMonitor import org.thoughtcrime.securesms.notifications.NotificationChannels -import org.thoughtcrime.securesms.push.SignalServiceNetworkAccess.Companion.toApplicableSystemHttpProxy import org.thoughtcrime.securesms.recipients.RecipientId import org.thoughtcrime.securesms.service.SafeForegroundService import org.thoughtcrime.securesms.util.AlarmSleepTimer @@ -48,7 +48,6 @@ import org.whispersystems.signalservice.api.messages.EnvelopeResponse import org.whispersystems.signalservice.api.websocket.SignalWebSocket import org.whispersystems.signalservice.api.websocket.WebSocketConnectionState import org.whispersystems.signalservice.api.websocket.WebSocketUnavailableException -import org.whispersystems.signalservice.internal.configuration.HttpProxy import org.whispersystems.signalservice.internal.push.Envelope import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.Semaphore @@ -106,13 +105,19 @@ class IncomingMessageObserver( private val lock: ReentrantLock = ReentrantLock() private val connectionNecessarySemaphore = Semaphore(0) - private var previousSystemHttpProxy: HttpProxy? = null - private val networkConnectionListener = NetworkConnectionListener( + + /** Tracks Internet connection as reported by [InternetConnectivityMonitor]. */ + @Volatile + private var internetConnection: ConnectivityState = ConnectivityState.OFFLINE + + private val internetConnectivityMonitor = InternetConnectivityMonitor( context = context, - onNetworkLost = { isNetworkUnavailable -> + onConnectivityUpdated = { state -> lock.withLock { - AppDependencies.libsignalNetwork.onNetworkChange() - if (isNetworkUnavailable()) { + internetConnection = state + if (state.isOnline) { + AppDependencies.libsignalNetwork.onNetworkChange() + } else { Log.w(TAG, "Lost network connection. Resetting the drained state.") decryptionDrained = false authWebSocket.disconnect() @@ -122,15 +127,9 @@ class IncomingMessageObserver( connectionNecessarySemaphore.release() } }, - onProxySettingsChanged = { proxyInfo -> - val systemHttpProxy = proxyInfo.toApplicableSystemHttpProxy() - if (systemHttpProxy?.host != previousSystemHttpProxy?.host || systemHttpProxy?.port != previousSystemHttpProxy?.port) { - val networkReset = AppDependencies.onSystemHttpProxyChange(systemHttpProxy) - if (networkReset) { - Log.i(TAG, "System proxy configuration changed, network reset.") - } - } - previousSystemHttpProxy = systemHttpProxy + onProxyChanged = { + Log.i(TAG, "System proxy configuration changed, network reset.") + AppDependencies.resetNetwork() } ) @@ -173,7 +172,7 @@ class IncomingMessageObserver( } }) - networkConnectionListener.register() + internetConnectivityMonitor.register() webSocketStateDisposable = authWebSocket .state @@ -243,7 +242,7 @@ class IncomingMessageObserver( val registered = SignalStore.account.isRegistered val unauthorizedReceived = TextSecurePreferences.isUnauthorizedReceived(context) val fcmEnabled = SignalStore.account.fcmEnabled - val hasNetwork = NetworkConstraint.isMet(context) + val hasNetwork = internetConnection.isOnline val hasProxy = SignalStore.proxy.isProxyEnabled val forceWebsocket = SignalStore.settings.forceWebsocketMode.isEnabled val websocketAlreadyOpen = isConnectionAvailable() @@ -265,7 +264,8 @@ class IncomingMessageObserver( } private fun isConnectionAvailable(): Boolean { - return !TextSecurePreferences.isUnauthorizedReceived(context) && SignalStore.account.isRegistered && (authWebSocket.stateSnapshot == WebSocketConnectionState.CONNECTED || (authWebSocket.shouldSendKeepAlives() && NetworkConstraint.isMet(context))) + val hasNetwork = internetConnection.isOnline + return !TextSecurePreferences.isUnauthorizedReceived(context) && SignalStore.account.isRegistered && (authWebSocket.stateSnapshot == WebSocketConnectionState.CONNECTED || (authWebSocket.shouldSendKeepAlives() && hasNetwork)) } private fun waitForConnectionNecessary() { @@ -285,7 +285,7 @@ class IncomingMessageObserver( fun terminate() { Log.w(TAG, "Termination! ${this.hashCode()}", Throwable()) INSTANCE_COUNT.decrementAndGet() - networkConnectionListener.unregister() + internetConnectivityMonitor.unregister() webSocketStateDisposable.dispose() terminated = true authWebSocket.disconnect() diff --git a/app/src/main/java/org/thoughtcrime/securesms/messages/NetworkConnectionListener.kt b/app/src/main/java/org/thoughtcrime/securesms/messages/NetworkConnectionListener.kt deleted file mode 100644 index baa3d95db4d..00000000000 --- a/app/src/main/java/org/thoughtcrime/securesms/messages/NetworkConnectionListener.kt +++ /dev/null @@ -1,159 +0,0 @@ -/* - * Copyright 2023 Signal Messenger, LLC - * SPDX-License-Identifier: AGPL-3.0-only - */ - -package org.thoughtcrime.securesms.messages - -import android.content.BroadcastReceiver -import android.content.Context -import android.content.Intent -import android.content.IntentFilter -import android.net.ConnectivityManager -import android.net.LinkProperties -import android.net.Network -import android.net.NetworkCapabilities -import android.net.NetworkRequest -import android.net.ProxyInfo -import android.os.Build -import org.signal.core.util.ServiceUtil -import org.signal.core.util.logging.Log -import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint - -/** - * Backcompat listener for determining when the network connection is lost. - * On API 28+, [onNetworkLost] is invoked when the system notifies the app that the network is lost. - * On earlier versions, [onNetworkLost] is invoked on any network change (gained, lost, losing, etc) - * Therefore, [onNetworkLost] is a higher-order function, which takes a function to determine conditionally if it should run. - * API 28+ only runs on lost networks, so it provides a conditional that's always true because that is guaranteed by the call site. - * Earlier versions use [NetworkConstraint.isMet] to query the current network state upon receiving the broadcast. - */ -class NetworkConnectionListener(private val context: Context, private val onNetworkLost: (() -> Boolean) -> Unit, private val onProxySettingsChanged: ((ProxyInfo?) -> Unit)) { - companion object { - private val TAG = Log.tag(NetworkConnectionListener::class.java) - } - - private val connectivityManager = ServiceUtil.getConnectivityManager(context) - private val lastNetworkCapabilities = mutableMapOf() - private val lastValidatedNetworkCapabilities = mutableMapOf() - - // onCapabilitiesChanged gets called every ~30 seconds or so in my testing, as the network estimator runs - // and updates the bandwidth estimated capabilities. This is thus essential for preventing log spam. - private fun logCapabilitiesIfChanged( - network: Network, - networkCapabilities: NetworkCapabilities, - callbackType: String, - lastLogs: MutableMap - ): Boolean { - val currentLog = buildString { - append(callbackType) - append(" onCapabilitiesChanged($network, ") - append("hasInternet=${networkCapabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)}, ") - append("validated=${networkCapabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)}, ") - append("captivePortal=${networkCapabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_CAPTIVE_PORTAL)})") - } - - if (lastLogs[network] != currentLog) { - Log.d(TAG, currentLog) - lastLogs[network] = currentLog - return true - } - - return false - } - - private val networkChangedCallback: ConnectivityManager.NetworkCallback = object : ConnectivityManager.NetworkCallback() { - override fun onUnavailable() { - super.onUnavailable() - Log.d(TAG, "ConnectivityManager.NetworkCallback onUnavailable()") - onNetworkLost { true } - } - - override fun onBlockedStatusChanged(network: Network, blocked: Boolean) { - super.onBlockedStatusChanged(network, blocked) - Log.d(TAG, "ConnectivityManager.NetworkCallback onBlockedStatusChanged($network, $blocked)") - onNetworkLost { blocked } - } - - override fun onAvailable(network: Network) { - super.onAvailable(network) - Log.d(TAG, "ConnectivityManager.NetworkCallback onAvailable($network)") - onNetworkLost { false } - } - - override fun onLost(network: Network) { - super.onLost(network) - Log.d(TAG, "ConnectivityManager.NetworkCallback onLost($network)") - onNetworkLost { true } - } - - override fun onLinkPropertiesChanged(network: Network, linkProperties: LinkProperties) { - super.onLinkPropertiesChanged(network, linkProperties) - Log.d(TAG, "ConnectivityManager.NetworkCallback onLinkPropertiesChanged($network)") - onProxySettingsChanged(linkProperties.httpProxy) - } - - override fun onCapabilitiesChanged(network: Network, networkCapabilities: NetworkCapabilities) { - super.onCapabilitiesChanged(network, networkCapabilities) - if (logCapabilitiesIfChanged(network, networkCapabilities, "ConnectivityManager.NetworkCallback", lastNetworkCapabilities)) { - onNetworkLost { !networkCapabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) } - } - } - } - - private val connectionReceiver = object : BroadcastReceiver() { - override fun onReceive(context: Context, intent: Intent) { - Log.d(TAG, "BroadcastReceiver onReceive().") - onNetworkLost { !NetworkConstraint.isMet(context) } - } - } - - private val logOnlyValidatedNetworkCallback: ConnectivityManager.NetworkCallback = object : ConnectivityManager.NetworkCallback() { - override fun onUnavailable() { - Log.d(TAG, "ValidatedNetworkCallback onUnavailable()") - super.onUnavailable() - } - - override fun onAvailable(network: Network) { - super.onAvailable(network) - Log.d(TAG, "ValidatedNetworkCallback onAvailable($network)") - } - - override fun onLost(network: Network) { - super.onLost(network) - Log.d(TAG, "ValidatedNetworkCallback onLost($network)") - } - - override fun onBlockedStatusChanged(network: Network, blocked: Boolean) { - super.onBlockedStatusChanged(network, blocked) - Log.d(TAG, "ValidatedNetworkCallback onBlockedStatusChanged($network, $blocked)") - } - - // This should not be strictly necessary, but seeing as what we're doing here is trying to get more - // insight into cases where Android's ConnectivityManager is behaving unexpectedly, I tend towards - // logging more rather than less. - override fun onCapabilitiesChanged(network: Network, networkCapabilities: NetworkCapabilities) { - super.onCapabilitiesChanged(network, networkCapabilities) - logCapabilitiesIfChanged(network, networkCapabilities, "ValidatedNetworkCallback", lastValidatedNetworkCapabilities) - } - } - - fun register() { - if (Build.VERSION.SDK_INT >= 28) { - connectivityManager.registerDefaultNetworkCallback(networkChangedCallback) - val validatedNetworkRequest = NetworkRequest.Builder().addCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED).build() - connectivityManager.registerNetworkCallback(validatedNetworkRequest, logOnlyValidatedNetworkCallback) - } else { - context.registerReceiver(connectionReceiver, IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION)) - } - } - - fun unregister() { - if (Build.VERSION.SDK_INT >= 28) { - connectivityManager.unregisterNetworkCallback(networkChangedCallback) - connectivityManager.unregisterNetworkCallback(logOnlyValidatedNetworkCallback) - } else { - context.unregisterReceiver(connectionReceiver) - } - } -} diff --git a/app/src/main/java/org/thoughtcrime/securesms/net/InternetConnectivityMonitor.kt b/app/src/main/java/org/thoughtcrime/securesms/net/InternetConnectivityMonitor.kt new file mode 100644 index 00000000000..2c6a662acaa --- /dev/null +++ b/app/src/main/java/org/thoughtcrime/securesms/net/InternetConnectivityMonitor.kt @@ -0,0 +1,284 @@ +/* + * Copyright 2026 Signal Messenger, LLC + * SPDX-License-Identifier: AGPL-3.0-only + */ + +package org.thoughtcrime.securesms.net + +import android.content.BroadcastReceiver +import android.content.Context +import android.content.Intent +import android.content.IntentFilter +import android.net.ConnectivityManager +import android.net.LinkProperties +import android.net.Network +import android.net.NetworkCapabilities +import android.net.NetworkRequest +import android.net.Proxy +import android.os.Build +import androidx.core.content.ContextCompat +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.channels.awaitClose +import kotlinx.coroutines.channels.trySendBlocking +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.callbackFlow +import kotlinx.coroutines.flow.conflate +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.retryWhen +import kotlinx.coroutines.launch +import org.signal.core.util.ServiceUtil +import org.signal.core.util.concurrent.SignalDispatchers +import org.signal.core.util.logging.Log +import org.signal.core.util.zipWithPrevious + +enum class ConnectivityState { + OFFLINE, + ONLINE, + ONLINE_VPN, + BLOCKED, + BLOCKED_VPN; + + /** Returns true if network traffic expected to reach the Internet. */ + val isOnline: Boolean + get() = this == ONLINE || this == ONLINE_VPN +} + +/** + * Monitors internet connectivity and proxy settings changes. + * + * [onConnectivityUpdated] is invoked when the [ConnectivityState] changes. + * The current state is delivered immediately upon registration if the device already has network access. + * + * [onProxyChanged] is invoked either the default network has changed or any network's proxy has changed. + * + * Both callbacks are invoked on [SignalDispatchers.IO] but may run concurrently. + * Callers are responsible for thread safety. + */ +class InternetConnectivityMonitor( + private val context: Context, + private val onConnectivityUpdated: (ConnectivityState) -> Unit, + private val onProxyChanged: () -> Unit +) { + companion object { + private val TAG = Log.tag(InternetConnectivityMonitor::class.java) + } + + private val scope = CoroutineScope(SignalDispatchers.IO) + private var monitorJob: Job? = null + + @Synchronized + fun register() { + if (monitorJob != null) return + + monitorJob = scope.launch { + launch { + connectivityStateFlow() + .retryWhen { cause, _ -> + val retrying = cause is NetworkStateStaleException + Log.i(TAG, "Re-registering callback ($retrying): ${cause.message}") + retrying + } + .distinctUntilChanged() + .zipWithPrevious { prevState, state -> + val log = buildString { + append("Connectivity state changed: ") + prevState?.let { append("$it -> ") } + append(state) + } + Log.i(TAG, log) + state + }.collect { state -> + onConnectivityUpdated(state) + } + } + launch { + proxyChangesFlow() + .collect { + Log.i(TAG, "Default network or system proxy config changed") + onProxyChanged() + } + } + } + } + + @Synchronized + fun unregister() { + monitorJob?.cancel() + monitorJob = null + } + + private data class NetworkState( + val validated: Boolean, + val blocked: Boolean, + val onVpn: Boolean + ) { + val isReachable: Boolean get() = validated && !blocked + + companion object { + val DOWN = NetworkState(validated = false, blocked = false, onVpn = false) + } + } + + /** + * Aggregates the state of all available networks to determine true Internet reachability. + * + * Android's default network callback only reports the single "best" network. However, to correctly + * handle VPNs (especially VPNs with kill-switch) we must track all networks. + * + * Basically, this callback tracks the list of active networks with and without a VPN: + * - No VPN / No kill-switch: Internet is available when we have a non-blocked, validated + * underlying network (e.g., WiFi or Cellular). + * - With a VPN kill-switch: The system blocks direct access to underlying networks. + * Internet is only available if we have BOTH an underlying network AND a valid, + * non-blocked VPN network. If either fails, Internet access is lost. + */ + private class NetworkAggregationCallback( + private val onNetworkStateChanged: (NetworkState) -> Unit, + private val onVpnLoss: () -> Unit + ) : ConnectivityManager.NetworkCallback() { + + private val networks = mutableMapOf() + + override fun onCapabilitiesChanged(network: Network, capabilities: NetworkCapabilities) { + val validated = capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED) + val vpn = capabilities.hasTransport(NetworkCapabilities.TRANSPORT_VPN) + Log.d(TAG, "onCapabilitiesChanged($network, validated=$validated, vpn=$vpn)") + val existing = networks[network] + if (existing == null) { + // For new networks, we initially assume blocked = false: + // On API 29+, the actual status will be set by an incoming onBlockedStatusChanged callback. + // Otherwise, onBlockedStatusChanged is unsupported and isReachable rely only on `validated`. + networks[network] = NetworkState(validated = validated, blocked = false, onVpn = vpn) + + // API 26+ guarantees that onLinkPropertiesChanged is always called next for new networks, + // followed by onBlockedStatusChanged (on API 29+). + // We skip notifying here for newer APIs to avoid emitting a fast, incorrect ONLINE -> BLOCKED + // sequence if the network was already blocked. + if (Build.VERSION.SDK_INT < 26) { + notifyAggregatedState() + } + } else { + // For existing networks, this is a capability update, so we notify immediately. + // onBlockedStatusChanged isn't necessary called afterward. + networks[network] = existing.copy(validated = validated, onVpn = vpn) + notifyAggregatedState() + } + } + + override fun onLinkPropertiesChanged(network: Network, linkProperties: LinkProperties) { + // onBlockedStatusChanged will handle the notification on API 29+ + if (Build.VERSION.SDK_INT < 29) { + notifyAggregatedState() + } + } + + override fun onBlockedStatusChanged(network: Network, blocked: Boolean) { + Log.d(TAG, "onBlockedStatusChanged($network, blocked=$blocked)") + val existing = networks[network] ?: return + networks[network] = existing.copy(blocked = blocked) + notifyAggregatedState() + } + + override fun onLost(network: Network) { + Log.d(TAG, "onLost($network)") + val lossState = networks.remove(network) + if (lossState?.onVpn == true) { + onVpnLoss() + } else { + notifyAggregatedState() + } + } + + private fun notifyAggregatedState() { + onNetworkStateChanged(networks.bestNetworkState()) + } + + private fun Map.bestNetworkState(): NetworkState { + return if (isEmpty()) { + NetworkState.DOWN + } else { + // A VPN network is only considered validated if there's also an underlying + // non-VPN network. + val hasUnderlyingNet = values.any { !it.onVpn } + val eligibleStates = values.map { state -> + if (state.onVpn) { + state.copy(validated = state.validated && hasUnderlyingNet) + } else state + } + eligibleStates.maxBy { it.rank } + } + } + + private val NetworkState.rank: Int + get() = when { + isReachable && onVpn -> 4 + isReachable -> 3 + blocked && onVpn -> 2 + blocked -> 1 + else -> 0 + } + } + + private fun connectivityStateFlow(): Flow = callbackFlow { + val connectivityManager = ServiceUtil.getConnectivityManager(context) + val callback = NetworkAggregationCallback( + onNetworkStateChanged = { + val connectivityState = when { + it.isReachable && it.onVpn -> ConnectivityState.ONLINE_VPN + it.isReachable -> ConnectivityState.ONLINE + it.blocked && it.onVpn -> ConnectivityState.BLOCKED_VPN + it.blocked -> ConnectivityState.BLOCKED + else -> ConnectivityState.OFFLINE + } + // Should not block as we conflate the flow + trySendBlocking(connectivityState) + }, + onVpnLoss = { + // VPN transport disconnected. For always-on VPNs with a kill switch, + // the underlying network may still appear "UP" but traffic is blocked. + // Restart the flow to re-evaluate connectivity. + close(NetworkStateStaleException("VPN loss")) + } + ) + + val request = NetworkRequest.Builder() + .removeCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN) // VPNs are excluded by default + .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) + .build() + + connectivityManager.registerNetworkCallback(request, callback) + + awaitClose { + connectivityManager.unregisterNetworkCallback(callback) + } + }.conflate() + + private fun proxyChangesFlow(): Flow = callbackFlow { + // Rely on the system-wide PROXY_CHANGE_ACTION sticky broadcast rather than + // per-network LinkProperties in the NetworkCallback. This ensures we catch + // proxy changes that occur when the system switches the default active network + // even if the new network's properties haven't changed. + val changeReceiver = object : BroadcastReceiver() { + override fun onReceive(context: Context, intent: Intent) { + trySendBlocking(Unit) + } + } + + ContextCompat.registerReceiver( + context, + changeReceiver, + IntentFilter(Proxy.PROXY_CHANGE_ACTION), + ContextCompat.RECEIVER_NOT_EXPORTED + ) + + awaitClose { + context.unregisterReceiver(changeReceiver) + } + }.conflate() + + /** + * Thrown when the tracked network state is no longer reliable. + */ + class NetworkStateStaleException(message: String) : Exception(message) +} diff --git a/app/src/test/java/org/thoughtcrime/securesms/messages/NetworkConnectionListenerTest.kt b/app/src/test/java/org/thoughtcrime/securesms/messages/NetworkConnectionListenerTest.kt deleted file mode 100644 index b44838bd58b..00000000000 --- a/app/src/test/java/org/thoughtcrime/securesms/messages/NetworkConnectionListenerTest.kt +++ /dev/null @@ -1,59 +0,0 @@ -package org.thoughtcrime.securesms.messages - -import android.app.Application -import android.net.ConnectivityManager -import android.net.Network -import android.net.NetworkCapabilities -import androidx.test.core.app.ApplicationProvider -import assertk.assertThat -import assertk.assertions.containsExactly -import io.mockk.mockk -import org.junit.Test -import org.junit.runner.RunWith -import org.robolectric.RobolectricTestRunner -import org.robolectric.annotation.Config -import org.robolectric.util.ReflectionHelpers - -@RunWith(RobolectricTestRunner::class) -@Config(manifest = Config.NONE, application = Application::class, sdk = [31]) -class NetworkConnectionListenerTest { - - @Test - fun `default network capability changes notify listener`() { - val unavailableEvents = mutableListOf() - val listener = NetworkConnectionListener( - context = ApplicationProvider.getApplicationContext(), - onNetworkLost = { isNetworkUnavailable -> unavailableEvents += isNetworkUnavailable() }, - onProxySettingsChanged = {} - ) - val callback = ReflectionHelpers.getField(listener, "networkChangedCallback") - val network = mockk() - - callback.onCapabilitiesChanged(network, capabilities(hasInternet = true, validated = false)) - callback.onCapabilitiesChanged(network, capabilities(hasInternet = true, validated = false)) - callback.onCapabilitiesChanged(network, capabilities(hasInternet = true, validated = true)) - callback.onCapabilitiesChanged(network, capabilities(hasInternet = false, validated = false)) - - assertThat(unavailableEvents).containsExactly(false, false, true) - } - - private fun capabilities(hasInternet: Boolean, validated: Boolean): NetworkCapabilities { - val capabilities = NetworkCapabilities() - - if (hasInternet) { - capabilities.addCapabilityReflectively(NetworkCapabilities.NET_CAPABILITY_INTERNET) - } - - if (validated) { - capabilities.addCapabilityReflectively(NetworkCapabilities.NET_CAPABILITY_VALIDATED) - } - - return capabilities - } - - private fun NetworkCapabilities.addCapabilityReflectively(capability: Int) { - val method = NetworkCapabilities::class.java.getDeclaredMethod("addCapability", Int::class.javaPrimitiveType) - method.isAccessible = true - method.invoke(this, capability) - } -} From 4179339eab64f0ca9f4ae56d73237c0d229d0207 Mon Sep 17 00:00:00 2001 From: Oscar Mira Date: Mon, 4 May 2026 22:17:16 +0200 Subject: [PATCH 3/5] Replace LinkProperties proxy detection with ProxySelector The ProxySelector approach is simpler and handles PAC transparently. On proxy change detection in IncomingMessageObserver, reconfigure the libsignal network proxy and reconnect the websockets, instead of resetting the entire network module. --- .../securesms/dependencies/AppDependencies.kt | 6 +- .../ApplicationDependencyProvider.java | 8 +- .../dependencies/NetworkDependenciesModule.kt | 4 +- .../dependencies/NetworkProxyState.kt | 27 +++++ .../messages/IncomingMessageObserver.kt | 18 +++- .../securesms/net/NetworkProxy.kt | 101 ++++++++++++++++++ .../push/SignalServiceNetworkAccess.kt | 35 ------ .../MockApplicationDependencyProvider.kt | 2 +- .../testutil/MockAppDependenciesRule.kt | 1 + .../org/signal/core/util/net/SystemProxy.kt | 25 +++++ .../sample/RegistrationApplication.kt | 1 - .../internal/configuration/HttpProxy.kt | 6 -- .../SignalServiceConfiguration.kt | 1 - .../websocket/LibSignalNetworkExtensions.kt | 27 ----- .../network/rest/SignalRestClientTest.kt | 1 - 15 files changed, 186 insertions(+), 77 deletions(-) create mode 100644 app/src/main/java/org/thoughtcrime/securesms/dependencies/NetworkProxyState.kt create mode 100644 app/src/main/java/org/thoughtcrime/securesms/net/NetworkProxy.kt create mode 100644 core/util/src/main/java/org/signal/core/util/net/SystemProxy.kt delete mode 100644 lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/configuration/HttpProxy.kt diff --git a/app/src/main/java/org/thoughtcrime/securesms/dependencies/AppDependencies.kt b/app/src/main/java/org/thoughtcrime/securesms/dependencies/AppDependencies.kt index ee1e68b80fa..1fa27acf956 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/dependencies/AppDependencies.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/dependencies/AppDependencies.kt @@ -312,6 +312,10 @@ object AppDependencies { val libsignalNetwork: Network get() = networkModule.libsignalNetwork + @JvmStatic + val networkProxyState: NetworkProxyState + get() = networkModule.networkProxyState + @JvmStatic val authWebSocket: SignalWebSocket.AuthenticatedWebSocket get() = networkModule.authWebSocket @@ -488,7 +492,7 @@ object AppDependencies { fun provideClientZkReceiptOperations(signalServiceConfiguration: SignalServiceConfiguration): ClientZkReceiptOperations fun provideScheduledMessageManager(): ScheduledMessageManager fun providePinnedMessageManager(): PinnedMessageManager - fun provideLibsignalNetwork(config: SignalServiceConfiguration): Network + fun provideLibsignalNetwork(config: SignalServiceConfiguration, proxyState: NetworkProxyState): Network fun provideBillingApi(): BillingApi fun provideArchiveApi(authWebSocket: SignalWebSocket.AuthenticatedWebSocket, unauthWebSocket: SignalWebSocket.UnauthenticatedWebSocket, pushServiceSocket: PushServiceSocket, signalServiceConfiguration: SignalServiceConfiguration): ArchiveApi fun provideKeysApi(authWebSocket: SignalWebSocket.AuthenticatedWebSocket, unauthWebSocket: SignalWebSocket.UnauthenticatedWebSocket): KeysApi diff --git a/app/src/main/java/org/thoughtcrime/securesms/dependencies/ApplicationDependencyProvider.java b/app/src/main/java/org/thoughtcrime/securesms/dependencies/ApplicationDependencyProvider.java index a9ad7ab1926..2292fa61dc9 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/dependencies/ApplicationDependencyProvider.java +++ b/app/src/main/java/org/thoughtcrime/securesms/dependencies/ApplicationDependencyProvider.java @@ -85,6 +85,8 @@ import org.thoughtcrime.securesms.megaphone.MegaphoneRepository; import org.thoughtcrime.securesms.messages.IncomingMessageObserver; import org.thoughtcrime.securesms.net.DeviceTransferBlockingInterceptor; +import org.thoughtcrime.securesms.net.NetworkProxyKt; +import org.thoughtcrime.securesms.net.ProxyConfig; import org.thoughtcrime.securesms.net.SignalWebSocketHealthMonitor; import org.thoughtcrime.securesms.net.StandardUserAgentInterceptor; import org.thoughtcrime.securesms.notifications.MessageNotifier; @@ -332,10 +334,14 @@ public ApplicationDependencyProvider(@NonNull Application context) { } @Override - public @NonNull Network provideLibsignalNetwork(@NonNull SignalServiceConfiguration config) { + public @NonNull Network provideLibsignalNetwork(@NonNull SignalServiceConfiguration config, @NonNull NetworkProxyState proxyState) { Network network = new Network(BuildConfig.LIBSIGNAL_NET_ENV, StandardUserAgentInterceptor.USER_AGENT, RemoteConfig.getLibsignalConfigs(), Network.BuildVariant.PRODUCTION); LibSignalNetworkExtensions.applyConfiguration(network, config); + ProxyConfig proxyConfig = NetworkProxyKt.resolveProxyConfig(config.getSignalProxy().orElse(null)); + NetworkProxyKt.configureProxy(network, proxyConfig); + proxyState.update(proxyConfig); + return network; } diff --git a/app/src/main/java/org/thoughtcrime/securesms/dependencies/NetworkDependenciesModule.kt b/app/src/main/java/org/thoughtcrime/securesms/dependencies/NetworkDependenciesModule.kt index cc9fb8b4304..fc4597e6dec 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/dependencies/NetworkDependenciesModule.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/dependencies/NetworkDependenciesModule.kt @@ -121,9 +121,11 @@ class NetworkDependenciesModule( } val libsignalNetwork: Network by lazy { - provider.provideLibsignalNetwork(signalServiceNetworkAccess.getConfiguration()) + provider.provideLibsignalNetwork(signalServiceNetworkAccess.getConfiguration(), networkProxyState) } + val networkProxyState: NetworkProxyState = NetworkProxyState() + val authWebSocket: SignalWebSocket.AuthenticatedWebSocket by lazy { provider.provideAuthWebSocket({ signalServiceNetworkAccess.getConfiguration() }, { libsignalNetwork }).also { disposables += it.state.subscribe { s -> webSocketStateSubject.onNext(s) } diff --git a/app/src/main/java/org/thoughtcrime/securesms/dependencies/NetworkProxyState.kt b/app/src/main/java/org/thoughtcrime/securesms/dependencies/NetworkProxyState.kt new file mode 100644 index 00000000000..efc48918a82 --- /dev/null +++ b/app/src/main/java/org/thoughtcrime/securesms/dependencies/NetworkProxyState.kt @@ -0,0 +1,27 @@ +/* + * Copyright 2026 Signal Messenger, LLC + * SPDX-License-Identifier: AGPL-3.0-only + */ + +package org.thoughtcrime.securesms.dependencies + +import org.thoughtcrime.securesms.net.ProxyConfig +import java.util.concurrent.atomic.AtomicReference + +/** + * Tracks the proxy configuration that has been applied to the [org.signal.libsignal.net.Network] + * instance so callers can detect changes and restart connections when needed. + */ +class NetworkProxyState { + + private val current = AtomicReference(ProxyConfig.Direct) + + val currentConfig: ProxyConfig + get() = current.get() + + /** Returns true if the proxy changed and connections should be restarted. */ + fun update(proxyConfig: ProxyConfig): Boolean { + val prev = current.getAndUpdate { proxyConfig } + return prev != proxyConfig + } +} diff --git a/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt b/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt index 1c2936ffa94..8f27a9cd82c 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt @@ -17,6 +17,7 @@ import org.signal.core.util.SleepTimer import org.signal.core.util.UptimeSleepTimer import org.signal.core.util.concurrent.SignalExecutors import org.signal.core.util.logging.Log +import org.signal.core.util.orNull import org.signal.storageservice.storage.protos.groups.local.DecryptedGroup import org.thoughtcrime.securesms.R import org.thoughtcrime.securesms.crypto.ReentrantSessionLock @@ -35,6 +36,8 @@ import org.thoughtcrime.securesms.messages.MessageDecryptor.FollowUpOperation import org.thoughtcrime.securesms.messages.protocol.BufferedProtocolStore import org.thoughtcrime.securesms.net.ConnectivityState import org.thoughtcrime.securesms.net.InternetConnectivityMonitor +import org.thoughtcrime.securesms.net.configureProxy +import org.thoughtcrime.securesms.net.resolveProxyConfig import org.thoughtcrime.securesms.notifications.NotificationChannels import org.thoughtcrime.securesms.recipients.RecipientId import org.thoughtcrime.securesms.service.SafeForegroundService @@ -48,6 +51,7 @@ import org.whispersystems.signalservice.api.messages.EnvelopeResponse import org.whispersystems.signalservice.api.websocket.SignalWebSocket import org.whispersystems.signalservice.api.websocket.WebSocketConnectionState import org.whispersystems.signalservice.api.websocket.WebSocketUnavailableException +import org.whispersystems.signalservice.internal.configuration.SignalProxy import org.whispersystems.signalservice.internal.push.Envelope import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.Semaphore @@ -92,6 +96,9 @@ class IncomingMessageObserver( private val censored: Boolean get() = AppDependencies.signalServiceNetworkAccess.isCensored() + private val signalProxy: SignalProxy? + get() = AppDependencies.signalServiceNetworkAccess.getConfiguration().signalProxy.orNull() + /** * Stops the foreground service for websocket users. */ @@ -128,8 +135,15 @@ class IncomingMessageObserver( } }, onProxyChanged = { - Log.i(TAG, "System proxy configuration changed, network reset.") - AppDependencies.resetNetwork() + val proxyConfig = resolveProxyConfig(signalProxy) + val proxyChanged = AppDependencies.networkProxyState.update(proxyConfig) + if (proxyChanged) { + AppDependencies.libsignalNetwork.configureProxy(proxyConfig) + Log.i(TAG, "Proxy config changed, disconnecting websocket...") + decryptionDrained = false + authWebSocket.disconnect() + unauthWebSocket.disconnect() + } } ) diff --git a/app/src/main/java/org/thoughtcrime/securesms/net/NetworkProxy.kt b/app/src/main/java/org/thoughtcrime/securesms/net/NetworkProxy.kt new file mode 100644 index 00000000000..0d24dea67fa --- /dev/null +++ b/app/src/main/java/org/thoughtcrime/securesms/net/NetworkProxy.kt @@ -0,0 +1,101 @@ +/* + * Copyright 2026 Signal Messenger, LLC + * SPDX-License-Identifier: AGPL-3.0-only + */ + +package org.thoughtcrime.securesms.net + +import androidx.annotation.WorkerThread +import org.signal.core.util.logging.Log +import org.signal.core.util.net.resolveSystemProxy +import org.signal.libsignal.net.Network +import org.thoughtcrime.securesms.BuildConfig +import org.whispersystems.signalservice.internal.configuration.SignalProxy +import java.io.IOException +import java.net.InetSocketAddress +import java.net.Proxy + +private const val TAG = "NetworkProxy" + +sealed interface ProxyConfig { + data object Direct : ProxyConfig + + data class ProxyAddress( + val scheme: ProxyScheme, + val host: String, + val port: Int + ) : ProxyConfig +} + +/** Supported proxy schemes by [Network]. */ +enum class ProxyScheme(val value: String) { + TLS(Network.SIGNAL_TLS_PROXY_SCHEME), + SOCKS("socks5"), + HTTP("http") +} + +fun SignalProxy.toProxyConfig() = ProxyConfig.ProxyAddress(ProxyScheme.TLS, host, port) + +fun Proxy.toProxyConfig(): ProxyConfig? { + val sa = address() as? InetSocketAddress ?: return null + val scheme = when (type()) { + Proxy.Type.HTTP -> ProxyScheme.HTTP + Proxy.Type.SOCKS -> ProxyScheme.SOCKS + Proxy.Type.DIRECT -> return ProxyConfig.Direct + } + return ProxyConfig.ProxyAddress(scheme, sa.hostString, sa.port) +} + +/** + * Resolves the appropriate proxy configuration or system defaults. + * Falls back to direct connection if no valid proxy is found. + * + * @param signalProxy Explicit Signal TLS proxy setting, if configured + * @return The resolved proxy configuration + */ +@WorkerThread +fun resolveProxyConfig(signalProxy: SignalProxy?): ProxyConfig { + return signalProxy?.toProxyConfig() + ?: resolveSystemProxy(targetUrl = BuildConfig.SIGNAL_URL)?.toProxyConfig() + ?: ProxyConfig.Direct +} + +/** + * Configures the [Network] instance with the given proxy settings. + * + * TLS Proxies: configuration errors mark the proxy as invalid, causing future connections to + * fail until the proxy setting is changed. These are explicitly configured by the user in the app + * and must be respected. + * + * System Proxies: the Android system settings screen explicitly calls out that apps are allowed + * to ignore the proxy setting, so if configuration fails, we fall back to direct connection + * rather than breaking connectivity. + */ +fun Network.configureProxy(config: ProxyConfig) { + when (config) { + ProxyConfig.Direct -> { + Log.i(TAG, "No proxy configured.") + clearProxy() + } + + is ProxyConfig.ProxyAddress -> { + try { + setProxy(config.scheme.value, config.host, config.port, null, null) + Log.i(TAG, "Proxy configured: ${config.scheme}:${config.port}") + } catch (e: IOException) { + when (config.scheme) { + ProxyScheme.TLS -> { + Log.e(TAG, "Invalid Signal TLS proxy config! Failing connections until changed.", e) + setInvalidProxy() + } + + ProxyScheme.HTTP, + ProxyScheme.SOCKS -> { + Log.w(TAG, "Failed to configure ${config.scheme} proxy, falling back to direct connection.", e) + clearProxy() + } + } + } + } + } +} diff --git a/app/src/main/java/org/thoughtcrime/securesms/push/SignalServiceNetworkAccess.kt b/app/src/main/java/org/thoughtcrime/securesms/push/SignalServiceNetworkAccess.kt index 54781a44437..91cd344b37d 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/push/SignalServiceNetworkAccess.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/push/SignalServiceNetworkAccess.kt @@ -1,10 +1,6 @@ package org.thoughtcrime.securesms.push import android.content.Context -import android.net.ConnectivityManager -import android.net.ProxyInfo -import android.net.Uri -import androidx.core.content.ContextCompat import com.google.i18n.phonenumbers.PhoneNumberUtil import okhttp3.CipherSuite import okhttp3.ConnectionSpec @@ -25,7 +21,6 @@ import org.thoughtcrime.securesms.net.StandardUserAgentInterceptor import org.thoughtcrime.securesms.net.StaticDns import org.thoughtcrime.securesms.net.StorageServiceSizeLoggingInterceptor import org.whispersystems.signalservice.api.push.TrustStore -import org.whispersystems.signalservice.internal.configuration.HttpProxy import org.whispersystems.signalservice.internal.configuration.SignalCdnUrl import org.whispersystems.signalservice.internal.configuration.SignalCdsiUrl import org.whispersystems.signalservice.internal.configuration.SignalServiceConfiguration @@ -139,33 +134,6 @@ class SignalServiceNetworkAccess(context: Context) { .build() private val APP_CONNECTION_SPEC = ConnectionSpec.MODERN_TLS - - @Suppress("DEPRECATION") - private fun getSystemHttpProxy(context: Context): HttpProxy? { - val connectivityManager = ContextCompat.getSystemService(context, ConnectivityManager::class.java) ?: return null - - val proxyInfo = connectivityManager - .activeNetwork - ?.let { connectivityManager.getLinkProperties(it)?.httpProxy } - - return proxyInfo.toApplicableSystemHttpProxy() - } - - fun ProxyInfo?.toApplicableSystemHttpProxy(): HttpProxy? { - return this - ?.takeIf { !it.exclusionList.contains(BuildConfig.SIGNAL_URL.stripProtocol()) } - // NB: Edit carefully, dear reader, as the line below is written from hard won experience. - // It turns out, that despite being documented *nowhere*, if a PAC file is set - // as the system proxy, proxyInfo.host will return "localhost" and proxyInfo.port - // will return -1. - // I learnt this by reading the AOSP source code for ProxyInfo: - // https://android.googlesource.com/platform/frameworks/base/+/4696ee4/core/java/android/net/ProxyInfo.java#107 - // So, if we do not explicitly check that a PAC file is not set, the proxy - // we pass to libsignal may be syntactically invalid, and the user may be - // rendered unable to connect. - ?.takeIf { it.pacFileUrl == Uri.EMPTY } - ?.let { proxy -> HttpProxy(proxy.host, proxy.port) } - } } private val serviceTrustStore: TrustStore = SignalServiceTrustStore(context) @@ -221,7 +189,6 @@ class SignalServiceNetworkAccess(context: Context) { networkInterceptors = interceptors, dns = Optional.of(DNS), signalProxy = Optional.empty(), - systemHttpProxy = Optional.empty(), zkGroupServerPublicParams = zkGroupServerPublicParams, genericServerPublicParams = genericServerPublicParams, backupServerPublicParams = backupServerPublicParams, @@ -281,7 +248,6 @@ class SignalServiceNetworkAccess(context: Context) { networkInterceptors = interceptors, dns = Optional.of(DNS), signalProxy = if (SignalStore.proxy.isProxyEnabled) Optional.ofNullable(SignalStore.proxy.proxy) else Optional.empty(), - systemHttpProxy = Optional.ofNullable(getSystemHttpProxy(context)), zkGroupServerPublicParams = zkGroupServerPublicParams, genericServerPublicParams = genericServerPublicParams, backupServerPublicParams = backupServerPublicParams, @@ -354,7 +320,6 @@ class SignalServiceNetworkAccess(context: Context) { networkInterceptors = interceptors, dns = Optional.of(DNS), signalProxy = Optional.empty(), - systemHttpProxy = Optional.empty(), zkGroupServerPublicParams = zkGroupServerPublicParams, genericServerPublicParams = genericServerPublicParams, backupServerPublicParams = backupServerPublicParams, diff --git a/app/src/test/java/org/thoughtcrime/securesms/dependencies/MockApplicationDependencyProvider.kt b/app/src/test/java/org/thoughtcrime/securesms/dependencies/MockApplicationDependencyProvider.kt index 4e8f6f5d8bb..d83058a5777 100644 --- a/app/src/test/java/org/thoughtcrime/securesms/dependencies/MockApplicationDependencyProvider.kt +++ b/app/src/test/java/org/thoughtcrime/securesms/dependencies/MockApplicationDependencyProvider.kt @@ -247,7 +247,7 @@ class MockApplicationDependencyProvider : AppDependencies.Provider { return mockk(relaxed = true) } - override fun provideLibsignalNetwork(config: SignalServiceConfiguration): Network { + override fun provideLibsignalNetwork(config: SignalServiceConfiguration, proxyState: NetworkProxyState): Network { return mockk(relaxed = true) } diff --git a/app/src/test/java/org/thoughtcrime/securesms/testutil/MockAppDependenciesRule.kt b/app/src/test/java/org/thoughtcrime/securesms/testutil/MockAppDependenciesRule.kt index 386c16bf8de..23bb3984130 100644 --- a/app/src/test/java/org/thoughtcrime/securesms/testutil/MockAppDependenciesRule.kt +++ b/app/src/test/java/org/thoughtcrime/securesms/testutil/MockAppDependenciesRule.kt @@ -25,6 +25,7 @@ class MockAppDependenciesRule : ExternalResource() { "databaseObserver", "groupsV2Authorization", "isInitialized", + "networkProxyState", "okHttpClient", "signalOkHttpClient", "webSocketObserver" diff --git a/core/util/src/main/java/org/signal/core/util/net/SystemProxy.kt b/core/util/src/main/java/org/signal/core/util/net/SystemProxy.kt new file mode 100644 index 00000000000..19dab31761a --- /dev/null +++ b/core/util/src/main/java/org/signal/core/util/net/SystemProxy.kt @@ -0,0 +1,25 @@ +/* + * Copyright 2026 Signal Messenger, LLC + * SPDX-License-Identifier: AGPL-3.0-only + */ + +package org.signal.core.util.net + +import androidx.annotation.WorkerThread +import java.net.Proxy +import java.net.ProxySelector +import java.net.URI + +/** + * Resolves the system-configured [Proxy] for the given URL using [ProxySelector]. + * + * @return null if no proxy is configured. + */ +@WorkerThread +fun resolveSystemProxy( + targetUrl: String, + proxySelector: ProxySelector? = ProxySelector.getDefault() +): Proxy? { + val proxyList = proxySelector?.select(URI.create(targetUrl)) + return proxyList?.firstOrNull() +} diff --git a/demo/registration/src/main/java/org/signal/registration/sample/RegistrationApplication.kt b/demo/registration/src/main/java/org/signal/registration/sample/RegistrationApplication.kt index 5becd206546..2cd8942693f 100644 --- a/demo/registration/src/main/java/org/signal/registration/sample/RegistrationApplication.kt +++ b/demo/registration/src/main/java/org/signal/registration/sample/RegistrationApplication.kt @@ -126,7 +126,6 @@ class RegistrationApplication : Application() { networkInterceptors = emptyList(), dns = Optional.empty(), signalProxy = Optional.empty(), - systemHttpProxy = Optional.empty(), zkGroupServerPublicParams = Base64.decode("ABSY21VckQcbSXVNCGRYJcfWHiAMZmpTtTELcDmxgdFbtp/bWsSxZdMKzfCp8rvIs8ocCU3B37fT3r4Mi5qAemeGeR2X+/YmOGR5ofui7tD5mDQfstAI9i+4WpMtIe8KC3wU5w3Inq3uNWVmoGtpKndsNfwJrCg0Hd9zmObhypUnSkfYn2ooMOOnBpfdanRtrvetZUayDMSC5iSRcXKpdlukrpzzsCIvEwjwQlJYVPOQPj4V0F4UXXBdHSLK05uoPBCQG8G9rYIGedYsClJXnbrgGYG3eMTG5hnx4X4ntARBgELuMWWUEEfSK0mjXg+/2lPmWcTZWR9nkqgQQP0tbzuiPm74H2wMO4u1Wafe+UwyIlIT9L7KLS19Aw8r4sPrXZSSsOZ6s7M1+rTJN0bI5CKY2PX29y5Ok3jSWufIKcgKOnWoP67d5b2du2ZVJjpjfibNIHbT/cegy/sBLoFwtHogVYUewANUAXIaMPyCLRArsKhfJ5wBtTminG/PAvuBdJ70Z/bXVPf8TVsR292zQ65xwvWTejROW6AZX6aqucUjlENAErBme1YHmOSpU6tr6doJ66dPzVAWIanmO/5mgjNEDeK7DDqQdB1xd03HT2Qs2TxY3kCK8aAb/0iM0HQiXjxZ9HIgYhbtvGEnDKW5ILSUydqH/KBhW4Pb0jZWnqN/YgbWDKeJxnDbYcUob5ZY5Lt5ZCMKuaGUvCJRrCtuugSMaqjowCGRempsDdJEt+cMaalhZ6gczklJB/IbdwENW9KeVFPoFNFzhxWUIS5ML9riVYhAtE6JE5jX0xiHNVIIPthb458cfA8daR0nYfYAUKogQArm0iBezOO+mPk5vCNWI+wwkyFCqNDXz/qxl1gAntuCJtSfq9OC3NkdhQlgYQ=="), genericServerPublicParams = Base64.decode("AHILOIrFPXX9laLbalbA9+L1CXpSbM/bTJXZGZiuyK1JaI6dK5FHHWL6tWxmHKYAZTSYmElmJ5z2A5YcirjO/yfoemE03FItyaf8W1fE4p14hzb5qnrmfXUSiAIVrhaXVwIwSzH6RL/+EO8jFIjJ/YfExfJ8aBl48CKHgu1+A6kWynhttonvWWx6h7924mIzW0Czj2ROuh4LwQyZypex4GuOPW8sgIT21KNZaafgg+KbV7XM1x1tF3XA17B4uGUaDbDw2O+nR1+U5p6qHPzmJ7ggFjSN6Utu+35dS1sS0P9N"), backupServerPublicParams = Base64.decode("AHYrGb9IfugAAJiPKp+mdXUx+OL9zBolPYHYQz6GI1gWjpEu5me3zVNSvmYY4zWboZHif+HG1sDHSuvwFd0QszSwuSF4X4kRP3fJREdTZ5MCR0n55zUppTwfHRW2S4sdQ0JGz7YDQIJCufYSKh0pGNEHL6hv79Agrdnr4momr3oXdnkpVBIp3HWAQ6IbXQVSG18X36GaicI1vdT0UFmTwU2KTneluC2eyL9c5ff8PcmiS+YcLzh0OKYQXB5ZfQ06d6DiINvDQLy75zcfUOniLAj0lGJiHxGczin/RXisKSR8"), diff --git a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/configuration/HttpProxy.kt b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/configuration/HttpProxy.kt deleted file mode 100644 index da28349ad5c..00000000000 --- a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/configuration/HttpProxy.kt +++ /dev/null @@ -1,6 +0,0 @@ -package org.whispersystems.signalservice.internal.configuration - -/** - * HTTP Proxy configuration from Android OS configuration. - */ -class HttpProxy(val host: String, val port: Int) diff --git a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/configuration/SignalServiceConfiguration.kt b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/configuration/SignalServiceConfiguration.kt index 3c4c70607e2..dfad0111e83 100644 --- a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/configuration/SignalServiceConfiguration.kt +++ b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/configuration/SignalServiceConfiguration.kt @@ -17,7 +17,6 @@ data class SignalServiceConfiguration( val networkInterceptors: List, val dns: Optional, val signalProxy: Optional, - val systemHttpProxy: Optional, val zkGroupServerPublicParams: ByteArray, val genericServerPublicParams: ByteArray, val backupServerPublicParams: ByteArray, diff --git a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/websocket/LibSignalNetworkExtensions.kt b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/websocket/LibSignalNetworkExtensions.kt index 743a6d8cae0..7c381c47c69 100644 --- a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/websocket/LibSignalNetworkExtensions.kt +++ b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/internal/websocket/LibSignalNetworkExtensions.kt @@ -6,11 +6,8 @@ package org.whispersystems.signalservice.internal.websocket -import org.signal.core.util.logging.Log -import org.signal.core.util.orNull import org.signal.libsignal.net.Network import org.whispersystems.signalservice.internal.configuration.SignalServiceConfiguration -import java.io.IOException private const val TAG = "LibSignalNetworkExtensions" @@ -18,29 +15,5 @@ private const val TAG = "LibSignalNetworkExtensions" * Helper method to apply settings from the SignalServiceConfiguration. */ fun Network.applyConfiguration(config: SignalServiceConfiguration) { - val signalProxy = config.signalProxy.orNull() - val systemHttpProxy = config.systemHttpProxy.orNull() - - when { - (signalProxy != null) -> { - try { - this.setProxy(signalProxy.host, signalProxy.port) - } catch (e: IOException) { - Log.e(TAG, "Invalid proxy configuration set! Failing connections until changed.") - this.setInvalidProxy() - } - } - (systemHttpProxy != null) -> { - try { - this.setProxy("http", systemHttpProxy.host, systemHttpProxy.port, "", "") - } catch (e: IOException) { - // The Android settings screen where this is set explicitly calls out that apps are allowed to - // ignore the HTTP Proxy setting, so if using the specified proxy would cause us to break, let's - // try just ignoring it and seeing if that still lets us connect. - Log.w(TAG, "Failed to set system HTTP proxy, ignoring and continuing...") - } - } - } - this.setCensorshipCircumventionEnabled(config.censored) } diff --git a/lib/network/src/test/java/org/signal/network/rest/SignalRestClientTest.kt b/lib/network/src/test/java/org/signal/network/rest/SignalRestClientTest.kt index 008e16416f7..0c99179a220 100644 --- a/lib/network/src/test/java/org/signal/network/rest/SignalRestClientTest.kt +++ b/lib/network/src/test/java/org/signal/network/rest/SignalRestClientTest.kt @@ -176,7 +176,6 @@ class SignalRestClientTest { networkInterceptors = emptyList(), dns = Optional.empty(), signalProxy = Optional.empty(), - systemHttpProxy = Optional.empty(), zkGroupServerPublicParams = ByteArray(0), genericServerPublicParams = ByteArray(0), backupServerPublicParams = ByteArray(0), From ca897f6972c0b64966da564e5ec017bcea4d90b2 Mon Sep 17 00:00:00 2001 From: Oscar Mira Date: Sun, 3 May 2026 14:24:22 +0200 Subject: [PATCH 4/5] Fix semaphore lost wakeup in waitForConnectionNecessary It eliminates the race where an initial drainPermits() in the wait loop could silently discard a valid wake-up signal. --- .../messages/IncomingMessageObserver.kt | 33 ++++++++++--------- 1 file changed, 18 insertions(+), 15 deletions(-) diff --git a/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt b/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt index 8f27a9cd82c..817b53cf349 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt @@ -131,7 +131,7 @@ class IncomingMessageObserver( // TODO [no-more-rest] Move the connection listener to a neutral location so this isn't passed in unauthWebSocket.disconnect() } - connectionNecessarySemaphore.release() + notifyConnectionConditionsChanged() } }, onProxyChanged = { @@ -194,7 +194,7 @@ class IncomingMessageObserver( .subscribeBy { if (it == WebSocketConnectionState.CONNECTED) { lock.withLock { - connectionNecessarySemaphore.release() + notifyConnectionConditionsChanged() } } } @@ -202,14 +202,14 @@ class IncomingMessageObserver( authWebSocket.addKeepAliveChangeListener { SignalExecutors.BOUNDED.execute { lock.withLock { - connectionNecessarySemaphore.release() + notifyConnectionConditionsChanged() } } } } fun notifyRegistrationStateChanged() { - connectionNecessarySemaphore.release() + notifyConnectionConditionsChanged() } fun notifyRestoreDecisionMade() { @@ -232,7 +232,7 @@ class IncomingMessageObserver( lock.withLock { appVisible = true BackgroundService.start(context) - connectionNecessarySemaphore.release() + notifyConnectionConditionsChanged() } } @@ -240,7 +240,7 @@ class IncomingMessageObserver( lock.withLock { appVisible = false lastInteractionTime = System.currentTimeMillis() - connectionNecessarySemaphore.release() + notifyConnectionConditionsChanged() } } @@ -283,19 +283,22 @@ class IncomingMessageObserver( } private fun waitForConnectionNecessary() { - try { - connectionNecessarySemaphore.drainPermits() - while (!isConnectionNecessary() && !isConnectionAvailable()) { - val numberDrained = connectionNecessarySemaphore.drainPermits() - if (numberDrained == 0) { - connectionNecessarySemaphore.acquire() - } + while (!isConnectionNecessary() && !isConnectionAvailable()) { + val numberDrained = connectionNecessarySemaphore.drainPermits() + if (numberDrained == 0) { + connectionNecessarySemaphore.acquireUninterruptibly() } - } catch (e: InterruptedException) { - throw AssertionError(e) } } + /** + * Signals that a condition affecting the connection decision may have changed. + */ + private fun notifyConnectionConditionsChanged() { + connectionNecessarySemaphore.drainPermits() + connectionNecessarySemaphore.release() + } + fun terminate() { Log.w(TAG, "Termination! ${this.hashCode()}", Throwable()) INSTANCE_COUNT.decrementAndGet() From bccdc6044bef95998abd0e640f066a954822eed8 Mon Sep 17 00:00:00 2001 From: Oscar Mira Date: Sun, 3 May 2026 17:28:59 +0200 Subject: [PATCH 5/5] Remove redundant lock from IncomingMessageObserver Following the previous refactor, the ReentrantLock is no longer needed. The only remaining shared state is appVisible and lastInteractionTime, which are replaced with an AppState data class to keep reads consistent. --- .../messages/IncomingMessageObserver.kt | 84 ++++++++----------- 1 file changed, 35 insertions(+), 49 deletions(-) diff --git a/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt b/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt index 817b53cf349..b16ec57cfac 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt @@ -58,8 +58,6 @@ import java.util.concurrent.Semaphore import java.util.concurrent.TimeUnit import java.util.concurrent.TimeoutException import java.util.concurrent.atomic.AtomicInteger -import java.util.concurrent.locks.ReentrantLock -import kotlin.concurrent.withLock import kotlin.math.round import kotlin.time.Duration.Companion.minutes import kotlin.time.Duration.Companion.seconds @@ -110,7 +108,6 @@ class IncomingMessageObserver( private val decryptionDrainedListeners: MutableList = CopyOnWriteArrayList() - private val lock: ReentrantLock = ReentrantLock() private val connectionNecessarySemaphore = Semaphore(0) /** Tracks Internet connection as reported by [InternetConnectivityMonitor]. */ @@ -120,19 +117,17 @@ class IncomingMessageObserver( private val internetConnectivityMonitor = InternetConnectivityMonitor( context = context, onConnectivityUpdated = { state -> - lock.withLock { - internetConnection = state - if (state.isOnline) { - AppDependencies.libsignalNetwork.onNetworkChange() - } else { - Log.w(TAG, "Lost network connection. Resetting the drained state.") - decryptionDrained = false - authWebSocket.disconnect() - // TODO [no-more-rest] Move the connection listener to a neutral location so this isn't passed in - unauthWebSocket.disconnect() - } - notifyConnectionConditionsChanged() + internetConnection = state + if (state.isOnline) { + AppDependencies.libsignalNetwork.onNetworkChange() + } else { + Log.w(TAG, "Lost network connection. Resetting the drained state.") + decryptionDrained = false + authWebSocket.disconnect() + // TODO [no-more-rest] Move the connection listener to a neutral location so this isn't passed in + unauthWebSocket.disconnect() } + notifyConnectionConditionsChanged() }, onProxyChanged = { val proxyConfig = resolveProxyConfig(signalProxy) @@ -149,8 +144,14 @@ class IncomingMessageObserver( private val messageContentProcessor = MessageContentProcessor.create(context) - private var appVisible = false - private var lastInteractionTime: Long = System.currentTimeMillis() + private data class AppState( + val isForeground: Boolean, + val lastInteractionTime: Long + ) + + @Volatile + private var appState = AppState(false, System.currentTimeMillis()) + private var webSocketStateDisposable = Disposable.disposed() @Volatile @@ -178,11 +179,11 @@ class IncomingMessageObserver( AppForegroundObserver.addListener(object : AppForegroundObserver.Listener { override fun onForeground() { - SignalExecutors.BOUNDED.execute { onAppForegrounded() } + onAppForegrounded() } override fun onBackground() { - SignalExecutors.BOUNDED.execute { onAppBackgrounded() } + onAppBackgrounded() } }) @@ -193,18 +194,12 @@ class IncomingMessageObserver( .observeOn(Schedulers.computation()) .subscribeBy { if (it == WebSocketConnectionState.CONNECTED) { - lock.withLock { - notifyConnectionConditionsChanged() - } + notifyConnectionConditionsChanged() } } authWebSocket.addKeepAliveChangeListener { - SignalExecutors.BOUNDED.execute { - lock.withLock { - notifyConnectionConditionsChanged() - } - } + notifyConnectionConditionsChanged() } } @@ -229,29 +224,21 @@ class IncomingMessageObserver( } private fun onAppForegrounded() { - lock.withLock { - appVisible = true - BackgroundService.start(context) - notifyConnectionConditionsChanged() - } + appState = appState.copy(isForeground = true) + BackgroundService.start(context) + notifyConnectionConditionsChanged() } private fun onAppBackgrounded() { - lock.withLock { - appVisible = false - lastInteractionTime = System.currentTimeMillis() - notifyConnectionConditionsChanged() - } + appState = appState.copy(isForeground = false, lastInteractionTime = System.currentTimeMillis()) + notifyConnectionConditionsChanged() } private fun isConnectionNecessary(): Boolean { - val timeIdle: Long - val appVisibleSnapshot: Boolean - - lock.withLock { - appVisibleSnapshot = appVisible - timeIdle = if (appVisibleSnapshot) 0 else System.currentTimeMillis() - lastInteractionTime - } + val appStateSnapshot = appState + val isForeground = appStateSnapshot.isForeground + val lastInteractionTime = appStateSnapshot.lastInteractionTime + val timeIdle = if (isForeground) 0 else System.currentTimeMillis() - lastInteractionTime val registered = SignalStore.account.isRegistered val unauthorizedReceived = TextSecurePreferences.isUnauthorizedReceived(context) @@ -261,19 +248,18 @@ class IncomingMessageObserver( val forceWebsocket = SignalStore.settings.forceWebsocketMode.isEnabled val websocketAlreadyOpen = isConnectionAvailable() - val lastInteractionString = if (appVisibleSnapshot) "N/A" else timeIdle.toString() + " ms (" + (if (timeIdle < maxBackgroundTime) "within limit" else "over limit") + ")" + val lastInteractionString = if (isForeground) "N/A" else timeIdle.toString() + " ms (" + (if (timeIdle < maxBackgroundTime) "within limit" else "over limit") + ")" val conclusion = registered && !unauthorizedReceived && - (appVisibleSnapshot || timeIdle < maxBackgroundTime || !fcmEnabled || forceWebsocket) && + (isForeground || timeIdle < maxBackgroundTime || !fcmEnabled || forceWebsocket) && hasNetwork val needsConnectionString = if (conclusion) "Needs Connection" else "Does Not Need Connection" Log.d( TAG, - "[$needsConnectionString] Network: $hasNetwork, Foreground: $appVisibleSnapshot, Time Since Last Interaction: $lastInteractionString, FCM: $fcmEnabled, WS Open or Keep-alives: $websocketAlreadyOpen, Registered: $registered, Unauthorized: $unauthorizedReceived, Proxy: $hasProxy, Force websocket: $forceWebsocket" + "[$needsConnectionString] Network: $hasNetwork, Foreground: $isForeground, Time Since Last Interaction: $lastInteractionString, FCM: $fcmEnabled, WS Open or Keep-alives: $websocketAlreadyOpen, Registered: $registered, Unauthorized: $unauthorizedReceived, Proxy: $hasProxy, Force websocket: $forceWebsocket" ) - return conclusion } @@ -546,7 +532,7 @@ class IncomingMessageObserver( } } - if (!appVisible) { + if (!appState.isForeground) { BackgroundService.stop(context) } } catch (e: Throwable) {