better tor management (#373)

This commit is contained in:
callebtc
2025-09-03 02:22:27 +02:00
committed by GitHub
parent 8b3dc71dc6
commit 905ccf5f17
@@ -16,6 +16,8 @@ import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.withTimeoutOrNull
import kotlinx.coroutines.CompletableDeferred
import java.net.InetSocketAddress import java.net.InetSocketAddress
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
@@ -31,6 +33,7 @@ object TorManager {
private const val RESTART_DELAY_MS = 2000L // 2 seconds between stop/start private const val RESTART_DELAY_MS = 2000L // 2 seconds between stop/start
private const val INACTIVITY_TIMEOUT_MS = 5000L // 5 seconds of no activity before restart private const val INACTIVITY_TIMEOUT_MS = 5000L // 5 seconds of no activity before restart
private const val MAX_RETRY_ATTEMPTS = 5 private const val MAX_RETRY_ATTEMPTS = 5
private const val STOP_TIMEOUT_MS = 7000L
private val appScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) private val appScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
@@ -51,19 +54,24 @@ object TorManager {
private enum class LifecycleState { STOPPED, STARTING, RUNNING, STOPPING } private enum class LifecycleState { STOPPED, STARTING, RUNNING, STOPPING }
@Volatile private var lifecycleState: LifecycleState = LifecycleState.STOPPED @Volatile private var lifecycleState: LifecycleState = LifecycleState.STOPPED
enum class TorState { OFF, STARTING, BOOTSTRAPPING, RUNNING, STOPPING, ERROR }
data class TorStatus( data class TorStatus(
val mode: TorMode = TorMode.OFF, val mode: TorMode = TorMode.OFF,
val running: Boolean = false, val running: Boolean = false,
val bootstrapPercent: Int = 0, val bootstrapPercent: Int = 0, // kept for backwards compatibility with UI; 0 or 100 only
val lastLogLine: String = "" val lastLogLine: String = "",
val state: TorState = TorState.OFF
) )
private val _status = MutableStateFlow(TorStatus()) private val _status = MutableStateFlow(TorStatus())
val statusFlow: StateFlow<TorStatus> = _status.asStateFlow() val statusFlow: StateFlow<TorStatus> = _status.asStateFlow()
private val stateChangeDeferred = AtomicReference<CompletableDeferred<TorState>?>(null)
fun isProxyEnabled(): Boolean { fun isProxyEnabled(): Boolean {
val s = _status.value val s = _status.value
return s.mode != TorMode.OFF && s.running && s.bootstrapPercent >= 100 && socksAddr != null return s.mode != TorMode.OFF && s.running && s.bootstrapPercent >= 100 && socksAddr != null && s.state == TorState.RUNNING
} }
fun init(application: Application) { fun init(application: Application) {
@@ -104,9 +112,12 @@ object TorManager {
TorMode.OFF -> { TorMode.OFF -> {
Log.i(TAG, "applyMode: OFF -> stopping tor") Log.i(TAG, "applyMode: OFF -> stopping tor")
lifecycleState = LifecycleState.STOPPING lifecycleState = LifecycleState.STOPPING
stopArti() _status.value = _status.value.copy(mode = TorMode.OFF, running = false, bootstrapPercent = 0, state = TorState.STOPPING)
stopArti() // non-suspending immediate request
// Best-effort wait for STOPPED before we declare OFF
waitForStateTransition(target = TorState.OFF, timeoutMs = STOP_TIMEOUT_MS)
socksAddr = null socksAddr = null
_status.value = _status.value.copy(mode = TorMode.OFF, running = false, bootstrapPercent = 0) _status.value = _status.value.copy(mode = TorMode.OFF, running = false, bootstrapPercent = 0, state = TorState.OFF)
currentSocksPort = DEFAULT_SOCKS_PORT currentSocksPort = DEFAULT_SOCKS_PORT
bindRetryAttempts = 0 bindRetryAttempts = 0
lifecycleState = LifecycleState.STOPPED lifecycleState = LifecycleState.STOPPED
@@ -124,8 +135,8 @@ object TorManager {
} }
bindRetryAttempts = 0 bindRetryAttempts = 0
lifecycleState = LifecycleState.STARTING lifecycleState = LifecycleState.STARTING
_status.value = _status.value.copy(mode = TorMode.ON, running = false, bootstrapPercent = 0, state = TorState.STARTING)
startArti(application, useDelay = false) startArti(application, useDelay = false)
_status.value = _status.value.copy(mode = TorMode.ON)
// Defer enabling proxy until bootstrap completes // Defer enabling proxy until bootstrap completes
appScope.launch { appScope.launch {
waitUntilBootstrapped() waitUntilBootstrapped()
@@ -146,7 +157,8 @@ object TorManager {
private suspend fun startArti(application: Application, useDelay: Boolean = false) { private suspend fun startArti(application: Application, useDelay: Boolean = false) {
try { try {
stopArtiInternal() // Ensure any previous instance is fully stopped before starting a new one
stopArtiAndWait()
Log.i(TAG, "Starting Arti on port $currentSocksPort") Log.i(TAG, "Starting Arti on port $currentSocksPort")
if (useDelay) { if (useDelay) {
@@ -159,15 +171,7 @@ object TorManager {
Log.i(TAG, "arti: $s") Log.i(TAG, "arti: $s")
lastLogTime.set(System.currentTimeMillis()) lastLogTime.set(System.currentTimeMillis())
_status.value = _status.value.copy(lastLogLine = s) _status.value = _status.value.copy(lastLogLine = s)
if ( handleArtiLogLine(s)
s.contains("Sufficiently bootstrapped", ignoreCase = true) ||
s.contains("AMEx: state changed to Running", ignoreCase = true)
) {
_status.value = _status.value.copy(bootstrapPercent = 100)
retryAttempts = 0 // Reset retry attempts on successful bootstrap
bindRetryAttempts = 0 // Reset bind retry attempts on successful bootstrap
startInactivityMonitoring()
}
} }
val proxy = ArtiProxy.Builder(application) val proxy = ArtiProxy.Builder(application)
@@ -180,12 +184,13 @@ object TorManager {
proxy.start() proxy.start()
lastLogTime.set(System.currentTimeMillis()) lastLogTime.set(System.currentTimeMillis())
_status.value = _status.value.copy(running = true, bootstrapPercent = 0) _status.value = _status.value.copy(running = true, bootstrapPercent = 0, state = TorState.STARTING)
lifecycleState = LifecycleState.RUNNING lifecycleState = LifecycleState.RUNNING
startInactivityMonitoring() startInactivityMonitoring()
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "Error starting Arti on port $currentSocksPort: ${e.message}") Log.e(TAG, "Error starting Arti on port $currentSocksPort: ${e.message}")
_status.value = _status.value.copy(state = TorState.ERROR)
// Check if this is a bind error // Check if this is a bind error
val isBindError = isBindError(e) val isBindError = isBindError(e)
@@ -198,7 +203,7 @@ object TorManager {
} else if (isBindError) { } else if (isBindError) {
Log.e(TAG, "Max bind retry attempts reached ($MAX_RETRY_ATTEMPTS), giving up") Log.e(TAG, "Max bind retry attempts reached ($MAX_RETRY_ATTEMPTS), giving up")
lifecycleState = LifecycleState.STOPPED lifecycleState = LifecycleState.STOPPED
_status.value = _status.value.copy(running = false, bootstrapPercent = 0) _status.value = _status.value.copy(running = false, bootstrapPercent = 0, state = TorState.ERROR)
} else { } else {
// For non-bind errors, use the existing retry mechanism // For non-bind errors, use the existing retry mechanism
scheduleRetry(application) scheduleRetry(application)
@@ -235,12 +240,21 @@ object TorManager {
private fun stopArti() { private fun stopArti() {
stopArtiInternal() stopArtiInternal()
socksAddr = null socksAddr = null
_status.value = _status.value.copy(running = false, bootstrapPercent = 0) _status.value = _status.value.copy(running = false, bootstrapPercent = 0, state = TorState.STOPPING)
}
private suspend fun stopArtiAndWait(timeoutMs: Long = STOP_TIMEOUT_MS) {
// Request stop
stopArtiInternal()
// Wait for confirmation via logs (Stopped) or timeout
waitForStateTransition(target = TorState.OFF, timeoutMs = timeoutMs)
// Small grace period before relaunch to let file locks clear
delay(200)
} }
private suspend fun restartArti(application: Application) { private suspend fun restartArti(application: Application) {
Log.i(TAG, "Restarting Arti (keeping SOCKS proxy enabled)...") Log.i(TAG, "Restarting Arti (keeping SOCKS proxy enabled)...")
stopArtiInternal() stopArtiAndWait()
delay(RESTART_DELAY_MS) delay(RESTART_DELAY_MS)
startArti(application, useDelay = false) // Already delayed above startArti(application, useDelay = false) // Already delayed above
} }
@@ -302,23 +316,68 @@ object TorManager {
retryJob = null retryJob = null
} }
// Removed Tor resource installation: not needed for Arti
/**
* Build an execution command that works on Android 10+ where app data dirs are mounted noexec.
* We invoke the platform dynamic linker and pass the PIE binary path as its first arg.
*/
// Removed exec command builder: not needed for Arti
private suspend fun waitUntilBootstrapped() { private suspend fun waitUntilBootstrapped() {
val current = _status.value val current = _status.value
if (!current.running) return if (!current.running) return
if (current.bootstrapPercent >= 100) return if (current.bootstrapPercent >= 100 && current.state == TorState.RUNNING) return
// Suspend until we observe 100% at least once // Suspend until we observe RUNNING at least once
while (true) { while (true) {
val s = statusFlow.first { it.bootstrapPercent >= 100 || !it.running } val s = statusFlow.first { (it.bootstrapPercent >= 100 && it.state == TorState.RUNNING) || !it.running || it.state == TorState.ERROR }
if (!s.running) return if (!s.running || s.state == TorState.ERROR) return
if (s.bootstrapPercent >= 100) return if (s.bootstrapPercent >= 100 && s.state == TorState.RUNNING) return
}
}
private fun handleArtiLogLine(s: String) {
when {
s.contains("AMEx: state changed to Initialized", ignoreCase = true) -> {
_status.value = _status.value.copy(state = TorState.STARTING)
completeWaitersIf(TorState.STARTING)
}
s.contains("AMEx: state changed to Starting", ignoreCase = true) -> {
_status.value = _status.value.copy(state = TorState.STARTING)
completeWaitersIf(TorState.STARTING)
}
s.contains("Sufficiently bootstrapped; system SOCKS now functional", ignoreCase = true) -> {
_status.value = _status.value.copy(bootstrapPercent = 100, state = TorState.BOOTSTRAPPING)
retryAttempts = 0
bindRetryAttempts = 0
startInactivityMonitoring()
}
s.contains("AMEx: state changed to Running", ignoreCase = true) -> {
// If we already saw Sufficiently bootstrapped, mark as RUNNING and ready.
val bp = if (_status.value.bootstrapPercent >= 100) 100 else 100 // treat Running as ready
_status.value = _status.value.copy(state = TorState.RUNNING, bootstrapPercent = bp, running = true)
completeWaitersIf(TorState.RUNNING)
}
s.contains("AMEx: state changed to Stopping", ignoreCase = true) -> {
_status.value = _status.value.copy(state = TorState.STOPPING, running = false)
}
s.contains("AMEx: state changed to Stopped", ignoreCase = true) -> {
_status.value = _status.value.copy(state = TorState.OFF, running = false, bootstrapPercent = 0)
completeWaitersIf(TorState.OFF)
}
s.contains("Another process has the lock on our state files", ignoreCase = true) -> {
// Signal error; we'll likely need to wait longer before restart
_status.value = _status.value.copy(state = TorState.ERROR)
}
}
}
private fun completeWaitersIf(state: TorState) {
stateChangeDeferred.getAndSet(null)?.let { def ->
def.complete(state)
}
}
private suspend fun waitForStateTransition(target: TorState, timeoutMs: Long): TorState? {
val def = CompletableDeferred<TorState>()
stateChangeDeferred.getAndSet(def)?.cancel()
return withTimeoutOrNull(timeoutMs) {
// Fast-path: if we're already there
val cur = _status.value.state
if (cur == target) return@withTimeoutOrNull cur
def.await()
} }
} }