diff --git a/app/src/main/java/chat/vojo/proxy/core/socks/LocalSocks5Server.kt b/app/src/main/java/chat/vojo/proxy/core/socks/LocalSocks5Server.kt index 313b038..46790dc 100644 --- a/app/src/main/java/chat/vojo/proxy/core/socks/LocalSocks5Server.kt +++ b/app/src/main/java/chat/vojo/proxy/core/socks/LocalSocks5Server.kt @@ -16,6 +16,7 @@ import kotlinx.coroutines.cancel import kotlinx.coroutines.delay import kotlinx.coroutines.isActive import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Semaphore import java.io.BufferedInputStream import java.io.Closeable import java.io.InputStream @@ -26,8 +27,10 @@ import java.net.InetSocketAddress import java.net.ServerSocket import java.net.Socket import java.net.SocketAddress +import java.security.SecureRandom import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong private const val TAG = "LocalSocks5" private const val RELAY_BUFFER = 32 * 1024 @@ -35,15 +38,30 @@ private const val SOCKS_VERSION = 0x05 private const val CMD_CONNECT = 0x01 private const val CMD_UDP_ASSOCIATE = 0x03 private const val DNS_TCP_TIMEOUT_MS = 10_000 +// A relay flow idle this long in BOTH directions is almost certainly dead. Kept +// generous so a legitimately quiet but live flow (idle SSH, an idle websocket) is +// not reset; it only bounds truly-dead half-closed-then-silent flows. +private const val RELAY_IDLE_TIMEOUT_MS = 30 * 60 * 1000L +// How often the idle watchdog wakes to compare against last activity. +private const val RELAY_POLL_MS = 30_000L +// Bound the SOCKS handshake so a local client that connects but never speaks can't +// pin a control-pool thread forever (loopback DoS). Reset to infinite once relaying. +private const val BRIDGE_HANDSHAKE_TIMEOUT_MS = 10_000 +// Cap concurrent DNS-over-TCP fallbacks so a retrying resolver can't fan out into +// hundreds of upstream CONNECTs (only used for HTTP upstreams without UDP). +private const val MAX_INFLIGHT_DNS_TCP = 8 /** * A loopback SOCKS5 server the native tun2socks engine connects to. Each inbound * SOCKS5 connection is bridged to the configured [upstream] proxy. Supports TCP * CONNECT and UDP ASSOCIATE (with a DNS-over-TCP fallback for HTTP upstreams). + * + * [onUpstreamError] receives the actual failure so the caller can classify it + * (auth vs TLS vs timeout vs wrong Shadowsocks key) instead of a generic "down". */ class LocalSocks5Server( private val upstream: Upstream, - private val onUpstreamError: (() -> Unit)? = null, + private val onUpstreamError: ((Throwable) -> Unit)? = null, ) { private val serverSocket = ServerSocket().apply { reuseAddress = true @@ -53,9 +71,22 @@ class LocalSocks5Server( /** Ephemeral loopback port the engine should target. */ val port: Int get() = serverSocket.localPort - // Elastic IO threads; blocking reads park their thread, so idle costs nothing. - private val dispatcher = Dispatchers.IO.limitedParallelism(512) - private val scope = CoroutineScope(SupervisorJob() + dispatcher) + // Random per-session credentials the engine must present. Without this the + // loopback bridge accepts NO-AUTH from any local app, so a process not in the + // per-app whitelist could find the ephemeral port and tunnel through the proxy + // anyway — the whitelist would stop being an access-control boundary. + val authUser: String = randomToken() + val authPass: String = randomToken() + + // Two pools so a flood of established relays can't starve connection setup: + // accept + the SOCKS handshake run on the small control pool; the long-lived + // blocking relay/UDP pumps run on the large relay pool. Blocking reads park + // their thread, so idle flows cost nothing. + private val controlDispatcher = Dispatchers.IO.limitedParallelism(64) + private val relayDispatcher = Dispatchers.IO.limitedParallelism(512) + private val scope = CoroutineScope(SupervisorJob() + controlDispatcher) + private val relayScope = CoroutineScope(SupervisorJob() + relayDispatcher) + private val dnsTcpLimit = Semaphore(MAX_INFLIGHT_DNS_TCP) private val open = ConcurrentHashMap.newKeySet() fun start() { @@ -67,6 +98,7 @@ class LocalSocks5Server( open.forEach { runCatching { it.close() } } open.clear() scope.cancel() + relayScope.cancel() } private fun track(c: Closeable) { open.add(c) } @@ -84,14 +116,23 @@ class LocalSocks5Server( delay(100) continue } + // Track before launching: if stop() races in between accept() and the + // coroutine starting, the socket is still in `open` (or closed right here) + // and never leaks. + track(client) + if (!scope.isActive || serverSocket.isClosed) { + closeQuietly(client); untrack(client); break + } scope.launch { handleClient(client) } } } private suspend fun handleClient(client: Socket) { - track(client) + track(client) // idempotent: acceptLoop already tracked it try { client.tcpNoDelay = true + // Bound the handshake; CMD handlers reset this to 0 before the long-lived phase. + runCatching { client.soTimeout = BRIDGE_HANDSHAKE_TIMEOUT_MS } val cin = BufferedInputStream(client.getInputStream()) val cout = client.getOutputStream() @@ -99,9 +140,19 @@ class LocalSocks5Server( if (cin.read() != SOCKS_VERSION) return val nMethods = cin.read() if (nMethods < 0) return - cin.readFully(ByteArray(nMethods)) - cout.write(byteArrayOf(SOCKS_VERSION.toByte(), 0x00)) // no-auth + val methods = ByteArray(nMethods) + cin.readFully(methods) + // Require username/password (0x02). The engine is configured with our + // random per-session credentials, so a local app outside the whitelist + // that probes this port can't tunnel through it. + if (methods.none { (it.toInt() and 0xff) == 0x02 }) { + cout.write(byteArrayOf(SOCKS_VERSION.toByte(), 0xFF.toByte())) // no acceptable method + cout.flush() + return + } + cout.write(byteArrayOf(SOCKS_VERSION.toByte(), 0x02)) // select user/pass cout.flush() + if (!authenticate(cin, cout)) return // Request: VER, CMD, RSV, [ATYP+ADDR+PORT] val reqHead = cin.readExact(3) @@ -120,7 +171,9 @@ class LocalSocks5Server( else -> writeReply(cout, rep = 0x07) // command not supported } } catch (e: Exception) { - // fallthrough to cleanup + // Don't discard the cause — a debug breadcrumb so "some flows stall" + // isn't a dead end in logcat. The user journal stays quiet. + Log.d(TAG, "client flow ended: ${e.javaClass.simpleName}: ${e.message}") } finally { closeQuietly(client) untrack(client) @@ -137,13 +190,16 @@ class LocalSocks5Server( upstream.connectTcp(dest) } catch (e: Exception) { Log.w(TAG, "upstream connect failed for $dest: ${e.message}") - onUpstreamError?.invoke() + onUpstreamError?.invoke(e) runCatching { writeReply(cout, rep = 0x05) } // connection refused return } track(conn) try { writeReply(cout, rep = 0x00) // success + // Handshake done — the relay phase is long-lived, governed by the idle + // watchdog below, so drop the handshake read timeout. + runCatching { client.soTimeout = 0 } // Half-close aware relay: EOF in one direction only shuts down the // peer's write side (hev forwards SHUT_WR from apps); the connection @@ -161,16 +217,41 @@ class LocalSocks5Server( if (finished.incrementAndGet() >= 2) closeBoth() } else closeBoth() } - val up = scope.launch { - val clean = runCatching { pump(cin, conn.output) }.isSuccess + // Shared activity clock fed by BOTH pumps. A watchdog reclaims the flow + // only when it's been idle in BOTH directions past the deadline — so an + // active one-way transfer (long upload, SSE download) is never killed, + // while a half-closed-then-silent flow can't pin a thread forever. + val lastActivity = AtomicLong(System.currentTimeMillis()) + val downBytes = AtomicLong(0) + val up = relayScope.launch { + val clean = runCatching { pump(cin, conn.output, lastActivity, AtomicLong(0)) }.isSuccess doneOne(clean) { conn.shutdownOutput() } } - val down = scope.launch { - val clean = runCatching { pump(conn.input, cout) }.isSuccess - doneOne(clean) { cout.flush(); client.shutdownOutput() } + val down = relayScope.launch { + val result = runCatching { pump(conn.input, cout, lastActivity, downBytes) } + // A wrong Shadowsocks key fails decrypting the FIRST upstream frame, before + // any plaintext is relayed. A GeneralSecurityException AFTER bytes have flowed + // is mid-stream corruption — NOT a wrong key — so don't tell a correctly-authed + // user to change their password. (Many SS servers instead just drop the + // connection on a bad key, which surfaces as a SocketException → generic error.) + val e = result.exceptionOrNull() + if (e is java.security.GeneralSecurityException && downBytes.get() == 0L) { + onUpstreamError?.invoke(e) + } + doneOne(result.isSuccess) { cout.flush(); client.shutdownOutput() } + } + val watchdog = relayScope.launch { + while (true) { + delay(RELAY_POLL_MS) + if (System.currentTimeMillis() - lastActivity.get() >= RELAY_IDLE_TIMEOUT_MS) { + closeBoth() // unblocks both pumps + break + } + } } up.join() down.join() + watchdog.cancel() } finally { runCatching { conn.close() } untrack(conn) @@ -194,11 +275,13 @@ class LocalSocks5Server( cout.flush() session.start() + // Long-lived keep-alive; drop the handshake read timeout. + runCatching { client.soTimeout = 0 } // Hold the association open until the engine closes the control connection. val drain = ByteArray(512) while (cin.read(drain) >= 0) { /* keep-alive */ } } catch (e: Exception) { - // fallthrough to cleanup + Log.d(TAG, "udp associate ended: ${e.javaClass.simpleName}: ${e.message}") } finally { session.stop() closeQuietly(relay) @@ -208,15 +291,18 @@ class LocalSocks5Server( /** Bridges UDP datagrams between the engine's relay socket and the upstream proxy. */ private inner class UdpSession(private val relay: DatagramSocket) { - private val assoc: UdpAssociation? = runCatching { upstream.openUdpAssociation() }.getOrNull() + private val assoc: UdpAssociation? = + runCatching { upstream.openUdpAssociation() } + .onFailure { Log.w(TAG, "UDP association unavailable: ${it.message}") } + .getOrNull() private val jobs = mutableListOf() @Volatile private var engineSource: SocketAddress? = null @Volatile private var running = true fun start() { assoc?.let { track(it) } - jobs += scope.launch { forwardLoop() } - if (assoc != null) jobs += scope.launch { replyLoop() } + jobs += relayScope.launch { forwardLoop() } + if (assoc != null) jobs += relayScope.launch { replyLoop() } } fun stop() { @@ -249,7 +335,13 @@ class LocalSocks5Server( assoc.send(dest, buf, dataStart, dataLen) } else if (dest.port == 53) { val q = buf.copyOfRange(dataStart, dp.length) - scope.launch { dnsOverTcp(dest, q) } + // Bounded: drop (let the resolver retry) rather than fan out + // an unbounded burst of upstream CONNECTs under a DNS storm. + if (dnsTcpLimit.tryAcquire()) { + relayScope.launch { + try { dnsOverTcp(dest, q) } finally { dnsTcpLimit.release() } + } + } } } catch (e: Exception) { // malformed datagram; skip @@ -282,7 +374,7 @@ class LocalSocks5Server( val resp = conn.input.readExact(rlen) sendToEngine(dest.raw, resp) } catch (e: Exception) { - // drop on failure + Log.d(TAG, "dns-over-tcp failed for $dest: ${e.message}") } finally { runCatching { conn.close() } untrack(conn) @@ -298,18 +390,40 @@ class LocalSocks5Server( } } + /** RFC 1929 user/pass subnegotiation; true only when the engine's creds match. */ + private fun authenticate(cin: InputStream, cout: OutputStream): Boolean { + if (cin.read() != 0x01) return false // subnegotiation version + val ulen = cin.read(); if (ulen < 0) return false + val user = ByteArray(ulen); cin.readFully(user) + val plen = cin.read(); if (plen < 0) return false + val pass = ByteArray(plen); cin.readFully(pass) + val ok = String(user, Charsets.US_ASCII) == authUser && String(pass, Charsets.US_ASCII) == authPass + val status: Byte = if (ok) 0x00 else 0x01 + cout.write(byteArrayOf(0x01, status)) + cout.flush() + return ok + } + + private fun randomToken(): String { + val b = ByteArray(12) + SecureRandom().nextBytes(b) + return b.joinToString("") { "%02x".format(it) } + } + private fun writeReply(out: OutputStream, rep: Int) { // VER, REP, RSV, ATYP=IPv4, BND.ADDR=0.0.0.0, BND.PORT=0 out.write(byteArrayOf(SOCKS_VERSION.toByte(), rep.toByte(), 0x00, 0x01, 0, 0, 0, 0, 0, 0)) out.flush() } - private fun pump(input: InputStream, output: OutputStream) { + private fun pump(input: InputStream, output: OutputStream, lastActivity: AtomicLong, bytesRelayed: AtomicLong) { val buf = ByteArray(RELAY_BUFFER) while (true) { val n = input.read(buf) if (n < 0) break if (n > 0) { + lastActivity.set(System.currentTimeMillis()) + bytesRelayed.addAndGet(n.toLong()) output.write(buf, 0, n) output.flush() }