heaven/proxy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(socks): require per-session auth on loopback bridge; both-direction idle watchdog; split control/relay dispatchers

Browse source
This commit is contained in:
heaven 2026-06-16 12:30:53 +03:00
parent 1b06c3c01f
commit 67deab0c8e
1 changed files with 135 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

156
app/src/main/java/chat/vojo/proxy/core/socks/LocalSocks5Server.kt
View file

@ -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<Closeable>()
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<Job>()
@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()
}