feat(service): probe upstream after connect, classify failures, reap stuck engine thread

app/src/main/java/chat/vojo/proxy/service/ProxyVpnService.kt

@@ -20,7 +20,10 @@ import chat.vojo.proxy.MainActivity
 import chat.vojo.proxy.R
 import chat.vojo.proxy.core.AppSettings
 import chat.vojo.proxy.core.ProxyConfig
+import chat.vojo.proxy.core.ProxyType
 import chat.vojo.proxy.core.Tun2Socks
+import chat.vojo.proxy.core.net.Destination
+import chat.vojo.proxy.core.net.readExact
 import chat.vojo.proxy.core.formatBytes
 import chat.vojo.proxy.core.formatUptime
 import chat.vojo.proxy.core.socks.LocalSocks5Server
@@ -111,6 +114,19 @@ class ProxyVpnService : VpnService(), SocketProtector {
 
     private fun startTunnel() { // lifecycle thread
         if (running) return
+        if (engineThread?.isAlive == true) {
+            // A previous engine overran the stop deadline and a reaper is still
+            // finishing it. Minting a new engine now collides with the native
+            // double-start guard (ret -2), so refuse and clean up: clear the FGS
+            // notification goForeground() just showed and end the service so the UI
+            // doesn't sit on a stale "Подключение…". The daemon reaper finishes
+            // closing the old fd; the user can reconnect once it's done.
+            VpnState.event("Туннель ещё останавливается", "подождите пару секунд и повторите", VpnState.TunnelEvent.Tone.WARN)
+            VpnState.setStopped(null)
+            stopForeground(STOP_FOREGROUND_REMOVE)
+            stopSelf()
+            return
+        }
         stopRequested = false
         tornDown = false
         lastUpstreamWarn = 0L
@@ -137,36 +153,174 @@ class ProxyVpnService : VpnService(), SocketProtector {
             vpnInterface = pfd
             VpnState.event("Интерфейс поднят", "$TUN_IPV4 · mtu ${state.settings.mtu}")
             VpnState.event("SOCKS5-мост", "127.0.0.1:${socks.port}")
-            VpnState.event("DNS через туннель", state.settings.dns)
+            VpnState.event("DNS через туннель", dnsServerFor(state.settings))
 
-            val hevConfig = buildEngineConfig(socks.port, state.settings)
+            val hevConfig = buildEngineConfig(socks, state.settings)
             val fd = pfd.fd
             engineThread = Thread({
                 val ret = Tun2Socks.nativeStart(hevConfig, fd)
                 Log.i(TAG, "engine returned $ret")
-                lifecycle.execute { onEngineExit(gen, ret) }
+                // If onDestroy already called lifecycle.shutdown(), this throws
+                // RejectedExecutionException on this thread (an uncaught crash). The
+                // engine is exiting during teardown anyway, so dropping the callback
+                // is correct.
+                runCatching { lifecycle.execute { onEngineExit(gen, ret) } }
             }, "vojo-tun2socks").apply { start() }
 
             running = true
             VpnState.setConnected(VpnState.Applied(config.id, state.settings))
-            VpnState.event("Подключено", "${config.name} · ${config.type.label}", VpnState.TunnelEvent.Tone.OK)
+            // The tun interface is up and routing immediately. Whether the *proxy*
+            // actually works (reachable, creds correct) is a separate question that
+            // a green "Подключено" must not pre-answer — the async probe below
+            // confirms it with OK, or names the real failure. Until then: NEUTRAL.
+            VpnState.event("Туннель поднят", "${config.name} · ${config.type.label}")
             updateNotification(getString(R.string.notif_connected, config.name))
             startStatsLoop()
             registerNetworkCallback()
+            runUpstreamProbe(upstream, config.type, gen)
         } catch (e: Throwable) {
             Log.e(TAG, "failed to start tunnel", e)
             fail(e.message ?: "Не удалось запустить туннель")
         }
     }
 
-    /** Throttled warning surfaced to the journal when upstream connects fail. */
-    private fun onUpstreamUnreachable() {
+    /** Throttled, classified diagnostic surfaced to the journal when upstream flows fail. */
+    private fun onUpstreamUnreachable(error: Throwable) {
         if (!running) return
         val now = System.currentTimeMillis()
-        if (now - lastUpstreamWarn > 30_000) {
+        if (now - lastUpstreamWarn < 30_000) return
         lastUpstreamWarn = now
-            VpnState.event("Прокси не отвечает", "проверьте сервер или сеть", VpnState.TunnelEvent.Tone.WARN)
+        val (label, detail, tone) = classifyUpstreamError(error)
+        VpnState.event(label, detail, tone)
     }
+
+    /**
+     * Turns a raw upstream failure into a specific, user-actionable journal line so
+     * "bad password" can never look like "dropped Wi-Fi". Used by both the live relay
+     * error path and the one-shot connectivity probe.
+     */
+    private fun classifyUpstreamError(e: Throwable): Triple<String, String?, VpnState.TunnelEvent.Tone> {
+        val warn = VpnState.TunnelEvent.Tone.WARN
+        val err = VpnState.TunnelEvent.Tone.ERROR
+        val msg = e.message ?: ""
+        return when {
+            e is javax.net.ssl.SSLPeerUnverifiedException ->
+                Triple("Сертификат прокси не прошёл проверку", msg.ifBlank { null }, err)
+            e is javax.net.ssl.SSLException ->
+                Triple("Ошибка TLS с прокси", msg.ifBlank { null }, err)
+            e is java.security.GeneralSecurityException ->
+                // Fires only when the server sends bytes that fail our AEAD tag (tampered or
+                // a true cipher mismatch). The COMMON wrong-key case never reaches here —
+                // most SS servers just drop the connection (see ShadowsocksKeyLikelyWrong).
+                Triple("Неверный пароль или шифр Shadowsocks", null, err)
+            e is ShadowsocksKeyLikelyWrong ->
+                Triple("Похоже, неверный ключ или шифр Shadowsocks", "сервер разорвал соединение, не ответив", warn)
+            e is java.net.UnknownHostException ->
+                Triple("Не удалось разрешить адрес прокси", msg.ifBlank { null }, err)
+            msg.contains("auth", ignoreCase = true) || msg.contains("407") ->
+                Triple("Прокси отклонил аутентификацию", "проверьте логин и пароль", err)
+            e is java.net.SocketTimeoutException || msg.contains("timed out", ignoreCase = true) ->
+                Triple("Прокси не отвечает", "таймаут — проверьте сервер или сеть", warn)
+            e is java.net.ConnectException ->
+                Triple("Не удалось подключиться к прокси", msg.ifBlank { null }, err)
+            // SOCKS5 rep!=0 / HTTP non-2xx: the proxy answered but refused the request.
+            // Could be access/policy (some proxies, e.g. 3proxy, accept any user/pass at
+            // the auth step and deny here instead) or an unreachable target — we can't
+            // tell which, so don't claim it's specifically the destination.
+            msg.contains("connect failed", ignoreCase = true) || Regex("rep=\\d").containsMatchIn(msg) ->
+                Triple("Прокси отклонил запрос", "доступ, политика или недоступный адрес", warn)
+            else ->
+                Triple("Ошибка прокси", msg.ifBlank { e.javaClass.simpleName }, warn)
+        }
+    }
+
+    /**
+     * One-shot proof that the configured proxy actually works: opens a real tunnelled
+     * connection to a stable target (1.1.1.1:53) right after connect. Success flips the
+     * journal to a confident OK; failure names the real cause. Makes "is the proxy up?"
+     * a deterministic, observable signal instead of a guess. The tunnel stays up either
+     * way so the user can read the diagnosis and fix the config.
+     */
+    private fun runUpstreamProbe(upstream: Upstream, type: ProxyType, gen: Int) {
+        scope.launch {
+            val startedAt = System.currentTimeMillis()
+            val result = runCatching {
+                // Shadowsocks has no connect-time handshake, so a wrong key passes
+                // connect() silently — force a DNS-over-TCP round trip to :53, which makes
+                // the server decrypt our frame. A bad key then fails the probe (usually the
+                // server dropping us before replying, occasionally an AEAD error), so we
+                // never show a false "Прокси подтверждён". SOCKS5/HTTP(S) authenticate during
+                // the handshake, so a plain CONNECT to :443 confirms them without
+                // false-failing proxies that block :53.
+                if (type == ProxyType.SHADOWSOCKS) probeRoundTrip(upstream)
+                else upstream.connectTcp(probeDestination(443)).use { }
+            }
+            // Drop a probe whose tunnel was already superseded (restart/stop) so a
+            // stale verdict can't land in a newer tunnel's journal.
+            if (!running || gen != engineGen) return@launch
+            result
+                .onSuccess {
+                    val ms = System.currentTimeMillis() - startedAt
+                    VpnState.event("Прокси подтверждён", "ответ за ${ms} мс", VpnState.TunnelEvent.Tone.OK)
+                }
+                .onFailure { e ->
+                    Log.w(TAG, "upstream probe failed", e)
+                    val (label, detail, tone) = classifyUpstreamError(e)
+                    VpnState.event(label, detail, tone)
+                }
+        }
+    }
+
+    /**
+     * Real end-to-end round trip through the proxy: a DNS-over-TCP query to 1.1.1.1:53.
+     * Forcing a *read* validates more than reachability + auth — for Shadowsocks it
+     * triggers AEAD decryption, so a wrong key/cipher fails the probe instead of showing a
+     * false "Прокси подтверждён" (connectTcp alone does no SS round trip). The failure is
+     * usually NOT a crypto exception: most servers drop the connection on a bad key, which
+     * we see as a premature EOF and translate to a hedged key hint.
+     */
+    private fun probeRoundTrip(upstream: Upstream) {
+        upstream.connectTcp(probeDestination(53)).use { conn ->
+            conn.socket.soTimeout = PROBE_TIMEOUT_MS
+            val q = dnsProbeQuery()
+            conn.output.write(byteArrayOf((q.size ushr 8).toByte(), q.size.toByte()))
+            conn.output.write(q)
+            conn.output.flush()
+            try {
+                conn.input.readExact(2) // a real reply proves the SS key decrypts end to end
+            } catch (e: java.net.SocketTimeoutException) {
+                throw e // proxy genuinely silent — keep the distinct "не отвечает" diagnosis
+            } catch (e: java.io.IOException) {
+                // Connected and sent our frame, but the peer died before replying — a clean
+                // EOF or a reset, with no AEAD exception. That's the common wrong-key/cipher
+                // signature for Shadowsocks (servers usually just drop). Reclassify so the
+                // journal can hint at the key instead of a generic error.
+                throw ShadowsocksKeyLikelyWrong(e)
+            }
+        }
+    }
+
+    /** Premature close on the Shadowsocks probe — a likely (not certain) wrong key/cipher. */
+    private class ShadowsocksKeyLikelyWrong(cause: Throwable) :
+        java.io.IOException("shadowsocks probe closed before reply", cause)
+
+    /** SOCKS address for the probe target 1.1.1.1:[port]. */
+    private fun probeDestination(port: Int): Destination {
+        val raw = byteArrayOf(0x01, 1, 1, 1, 1, (port ushr 8).toByte(), port.toByte())
+        return Destination(host = "1.1.1.1", port = port, atyp = 0x01, raw = raw)
+    }
+
+    /** Minimal DNS A query for cloudflare.com with recursion desired. */
+    private fun dnsProbeQuery(): ByteArray {
+        val out = java.io.ByteArrayOutputStream()
+        out.write(byteArrayOf(0x13, 0x37, 0x01, 0x00, 0, 1, 0, 0, 0, 0, 0, 0)) // id, flags=RD, qd=1
+        for (label in "cloudflare.com".split('.')) {
+            out.write(label.length)
+            out.write(label.toByteArray(Charsets.US_ASCII))
+        }
+        out.write(0) // end of name
+        out.write(byteArrayOf(0, 1, 0, 1)) // QTYPE=A, QCLASS=IN
+        return out.toByteArray()
     }
 
     /** Engine thread exited on its own (not via a requested stop) — tear down loudly. */
@@ -181,9 +335,16 @@ class ProxyVpnService : VpnService(), SocketProtector {
             .setMtu(settings.mtu)
             .addAddress(TUN_IPV4, 32)
             .addRoute("0.0.0.0", 0)
-            .addDnsServer(settings.dns.ifBlank { "1.1.1.1" })
+            .addDnsServer(dnsServerFor(settings))
             .setBlocking(false)
 
+        // IPv6 is only pulled into the tun when the user enables it AND the engine is
+        // told to carry it (buildEngineConfig adds the `ipv6:` line). We must NOT route
+        // v6 into the tun while disabled: hev does not silently drop unconfigured-family
+        // packets — it forwards the v6 CONNECT to the upstream, and a v6-incapable proxy
+        // rejects it (rep=5), which breaks Happy-Eyeballs apps that then prefer v6.
+        // With v6 absent here, Android adds its own `::/0 unreachable` route, so native
+        // v6 is blackholed (no leak) and apps fall back to v4 through the proxy.
         if (settings.ipv6) {
             builder.addAddress(TUN_IPV6, 128).addRoute("::", 0)
         }
@@ -219,15 +380,25 @@ class ProxyVpnService : VpnService(), SocketProtector {
         }
     }
 
-    private fun buildEngineConfig(socksPort: Int, settings: AppSettings): String = buildString {
+    /** DNS server routed through the tunnel; avoids a v6 literal while v6 isn't routed. */
+    private fun dnsServerFor(settings: AppSettings): String {
+        val dns = settings.dns.ifBlank { "1.1.1.1" }
+        return if (!settings.ipv6 && ':' in dns) "1.1.1.1" else dns
+    }
+
+    private fun buildEngineConfig(socks: LocalSocks5Server, settings: AppSettings): String = buildString {
         append("tunnel:\n")
         append("  mtu: ${settings.mtu}\n")
         append("  ipv4: $TUN_IPV4\n")
         if (settings.ipv6) append("  ipv6: '$TUN_IPV6'\n")
         append("socks5:\n")
         append("  address: 127.0.0.1\n")
-        append("  port: $socksPort\n")
+        append("  port: ${socks.port}\n")
         append("  udp: 'udp'\n")
+        // Authenticate the engine to the loopback bridge so no other local app can
+        // use it. Tokens are hex, safe to embed unquoted-style in single quotes.
+        append("  username: '${socks.authUser}'\n")
+        append("  password: '${socks.authPass}'\n")
         append("misc:\n")
         append("  connect-timeout: 8000\n")
         append("  log-level: warn\n")
@@ -261,12 +432,20 @@ class ProxyVpnService : VpnService(), SocketProtector {
 
     private fun registerNetworkCallback() {
         val cm = getSystemService(ConnectivityManager::class.java) ?: return
+        val registeredAt = System.currentTimeMillis()
         lastNetwork = cm.activeNetwork
         val cb = object : ConnectivityManager.NetworkCallback() {
             override fun onAvailable(network: Network) {
                 runCatching { setUnderlyingNetworks(arrayOf(network)) }
-                if (network != lastNetwork) {
+                if (network == lastNetwork) return
+                val previous = lastNetwork
                 lastNetwork = network
+                // Suppress bring-up churn: the first callbacks merely report the
+                // current underlying network(s) as the VPN settles, not a real roam.
+                // Only announce a change once we had a previous network and the tunnel
+                // has been up long enough that this is a genuine handover.
+                if (previous == null || System.currentTimeMillis() - registeredAt < NETWORK_SETTLE_MS) return
+                run {
                     val caps = runCatching { cm.getNetworkCapabilities(network) }.getOrNull()
                     val name = when {
                         caps?.hasTransport(NetworkCapabilities.TRANSPORT_WIFI) == true -> "Wi-Fi"
@@ -323,7 +502,6 @@ class ProxyVpnService : VpnService(), SocketProtector {
         networkCallback = null
 
         val engine = engineThread
-        engineThread = null
         if (engine != null) {
             // nativeStop only signals the engine's event fd; the native `running`
             // flag and the tun fd aren't released until nativeStart() returns and
@@ -336,19 +514,25 @@ class ProxyVpnService : VpnService(), SocketProtector {
                 runCatching { engine.join(ENGINE_JOIN_STEP_MS) }
                 waited += ENGINE_JOIN_STEP_MS
             }
-            if (engine.isAlive) Log.w(TAG, "engine thread did not exit in time")
         }
 
         server?.stop()
         server = null
 
-        // Only reclaim the tun fd once the engine has truly stopped touching it;
-        // closing it out from under live native code is undefined behaviour.
         if (engine == null || !engine.isAlive) {
+            engineThread = null
+            // Safe to reclaim the tun fd: the engine has stopped touching it.
             runCatching { vpnInterface?.close() }
             vpnInterface = null
         } else {
-            Log.w(TAG, "leaving tun fd open — engine still alive after stop")
+            // The engine overran the stop deadline. Do NOT drop the handle or close
+            // the fd under live native code, and do NOT let startTunnel mint a second
+            // engine (it would hit the C double-start guard, ret -2). Retain
+            // engineThread + vpnInterface and let a reaper finish teardown once the
+            // native engine finally exits. startTunnel() refuses to start while
+            // engineThread.isAlive, so the fd can't be overwritten in the meantime.
+            Log.w(TAG, "engine still alive after stop deadline; reaping in background")
+            reapStuckEngine(engine)
         }
 
         running = false
@@ -364,6 +548,32 @@ class ProxyVpnService : VpnService(), SocketProtector {
         }
     }
 
+    /**
+     * Finishes teardown of a native engine that overran the stop deadline, off the
+     * lifecycle thread so it never blocks it. Keeps re-kicking nativeStop until the
+     * engine truly exits, then closes the retained tun fd and clears the handle (on
+     * the lifecycle thread) so a subsequent connect is allowed again.
+     */
+    private fun reapStuckEngine(engine: Thread) {
+        val fd = vpnInterface
+        Thread({
+            while (engine.isAlive) {
+                runCatching { Tun2Socks.nativeStop() }
+                runCatching { engine.join(ENGINE_JOIN_STEP_MS) }
+            }
+            runCatching { fd?.close() }
+            runCatching {
+                lifecycle.execute {
+                    if (engineThread === engine) {
+                        engineThread = null
+                        vpnInterface = null
+                    }
+                    Log.i(TAG, "reaped stuck engine; tun fd released")
+                }
+            }
+        }, "vojo-engine-reaper").apply { isDaemon = true; start() }
+    }
+
     override fun onRevoke() {
         // User revoked VPN permission or another VPN took over.
         lifecycle.execute {
@@ -465,6 +675,13 @@ class ProxyVpnService : VpnService(), SocketProtector {
         private const val ENGINE_JOIN_TIMEOUT_MS = 5000L
         private const val ENGINE_JOIN_STEP_MS = 250L
 
+        // Grace window after connect during which underlying-network callbacks are
+        // treated as bring-up settling, not a user-visible roam.
+        private const val NETWORK_SETTLE_MS = 4000L
+
+        // Per-read timeout for the one-shot connectivity probe round trip.
+        private const val PROBE_TIMEOUT_MS = 8000
+
         const val TUN_IPV4 = "198.18.0.1"
         private const val TUN_IPV6 = "fc00::1"