This is an automated email from the ASF dual-hosted git repository.
tomaswolf pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mina-sshd.git
The following commit(s) were added to refs/heads/master by this push:
new 88632d115 Fix SOCKS5 dynamic forwarding of fragmented and pipelined
requests
88632d115 is described below
commit 88632d1156658288072a6c5c18394e80d23751ab
Author: Bruno A. Muciño <[email protected]>
AuthorDate: Tue Aug 18 10:40:37 2026 -0600
Fix SOCKS5 dynamic forwarding of fragmented and pipelined requests
The SOCKS5 proxy used for dynamic port forwarding parsed each read as a
complete message, throwing a buffer underflow when the greeting or the
CONNECT request arrived split across several reads. It also reused the
CONNECT request buffer as the reply, echoing request bytes (and any
application data pipelined after the request) back to the client instead
of forwarding them.
Rewrite Socks5 as a state machine with a retained buffer: parse only
once a message is complete and queue bytes received before the channel
opens, forwarding them once it is up.
The state is synchronized because channel open and write futures
complete on other threads.
---
.../org/apache/sshd/common/forward/SocksProxy.java | 205 +++++++++++++++------
.../src/test/java/org/apache/sshd/ProxyTest.java | 91 +++++++++
2 files changed, 241 insertions(+), 55 deletions(-)
diff --git
a/sshd-core/src/main/java/org/apache/sshd/common/forward/SocksProxy.java
b/sshd-core/src/main/java/org/apache/sshd/common/forward/SocksProxy.java
index 28a9d6fe2..96edb2ced 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/forward/SocksProxy.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/forward/SocksProxy.java
@@ -201,12 +201,20 @@ public class SocksProxy extends AbstractCloseable
implements IoHandler {
}
}
+ private enum Socks5State {
+ GREETING,
+ CONNECT_REQUEST,
+ OPENING_CHANNEL,
+ FORWARDING,
+ CLOSED
+ }
+
/**
* @see <A HREF="https://en.wikipedia.org/wiki/SOCKS#SOCKS5";>SOCKS5</A>
*/
public class Socks5 extends Proxy {
- private byte[] authMethods;
- private Buffer response;
+ private Socks5State state = Socks5State.GREETING;
+ private Buffer pending = new ByteArrayBuffer();
public Socks5(IoSession session) {
super(session);
@@ -214,65 +222,115 @@ public class SocksProxy extends AbstractCloseable
implements IoHandler {
@SuppressWarnings("synthetic-access")
@Override
- protected void onMessage(Buffer buffer) throws IOException {
- boolean debugEnabled = log.isDebugEnabled();
- if (authMethods == null) {
- int nbAuthMethods = getUByte(buffer);
- authMethods = new byte[nbAuthMethods];
- buffer.getRawBytes(authMethods);
+ protected synchronized void onMessage(Buffer buffer) throws
IOException {
+ if (state == Socks5State.FORWARDING && pending.available() == 0) {
+ super.onMessage(buffer);
+ return;
+ }
+ pending.putBuffer(buffer);
+ processPending();
+ }
+
+ protected void processPending() throws IOException {
+ if (state == Socks5State.OPENING_CHANNEL || state ==
Socks5State.CLOSED) {
+ return;
+ }
+
+ if (state == Socks5State.FORWARDING) {
+ forwardPending();
+ return;
+ }
+
+ if (state == Socks5State.GREETING) {
+ if (pending.available() < 1) {
+ return;
+ }
+ int authMethodsCount = pending.rawByte(pending.rpos()) & 0xFF;
+ if (pending.available() < authMethodsCount + 1) {
+ return;
+ }
+
+ pending.getUByte();
boolean foundNoAuth = false;
- for (int i = 0; i < nbAuthMethods; i++) {
- foundNoAuth |= authMethods[i] == 0;
+ for (int index = 0; index < authMethodsCount; index++) {
+ foundNoAuth |= pending.getUByte() == 0;
}
- buffer = new ByteArrayBuffer(Byte.SIZE, false);
- buffer.putByte((byte) 0x05);
- buffer.putByte((byte) (foundNoAuth ? 0x00 : 0xFF));
- session.writeBuffer(buffer);
+ pending.compact();
+ Buffer response = new ByteArrayBuffer(2, false);
+ response.putByte((byte) 0x05);
+ response.putByte((byte) (foundNoAuth ? 0x00 : 0xFF));
+ session.writeBuffer(response);
if (!foundNoAuth) {
throw new IllegalStateException("Received socks5 greeting
without NoAuth method");
- } else if (debugEnabled) {
+ } else if (log.isDebugEnabled()) {
log.debug("Received socks5 greeting");
}
- } else if (channel == null) {
- response = buffer;
- int version = getUByte(buffer);
+ state = Socks5State.CONNECT_REQUEST;
+ }
+
+ if (state == Socks5State.CONNECT_REQUEST) {
+ if (pending.available() < 4) {
+ return;
+ }
+ int offset = pending.rpos();
+ int version = pending.rawByte(offset) & 0xFF;
if (version != 0x05) {
throw new IllegalStateException("Unexpected version: " +
version);
}
- int cmd = buffer.getUByte();
+ int cmd = pending.rawByte(offset + 1) & 0xFF;
if (cmd != 1) { // establish a TCP/IP stream connection
throw new IllegalStateException("Unsupported socks
command: " + cmd);
}
- int res = buffer.getUByte();
+ int res = pending.rawByte(offset + 2) & 0xFF;
if (res != 0) {
- if (debugEnabled) {
+ if (log.isDebugEnabled()) {
log.debug("No zero reserved value: {}", res);
}
}
-
- int type = buffer.getUByte();
- String host;
+ int type = pending.rawByte(offset + 3) & 0xFF;
+ int addressLength;
if (type == 0x01) {
- host = Integer.toString(getUByte(buffer)) + "."
- + Integer.toString(getUByte(buffer)) + "."
- + Integer.toString(getUByte(buffer)) + "."
- + Integer.toString(getUByte(buffer));
+ addressLength = 4;
} else if (type == 0x03) {
- host = getBLString(buffer);
+ if (pending.available() < 5) {
+ return;
+ }
+ addressLength = pending.rawByte(offset + 4) & 0xFF;
} else if (type == 0x04) {
- host = Integer.toHexString(getUShort(buffer)) + ":"
- + Integer.toHexString(getUShort(buffer)) + ":"
- + Integer.toHexString(getUShort(buffer)) + ":"
- + Integer.toHexString(getUShort(buffer)) + ":"
- + Integer.toHexString(getUShort(buffer)) + ":"
- + Integer.toHexString(getUShort(buffer)) + ":"
- + Integer.toHexString(getUShort(buffer)) + ":"
- + Integer.toHexString(getUShort(buffer));
+ addressLength = 16;
} else {
throw new IllegalStateException("Unsupported address type:
" + type);
}
- int port = getUShort(buffer);
- if (debugEnabled) {
+ int requestLength = 4 + (type == 0x03 ? 1 : 0) + addressLength
+ 2;
+ if (pending.available() < requestLength) {
+ return;
+ }
+
+ getUByte(pending); // version already checked above
+ getUByte(pending); // command already checked above
+ getUByte(pending); // reserved byte
+ getUByte(pending); // address type
+ String host;
+ if (type == 0x01) {
+ host = Integer.toString(getUByte(pending)) + "."
+ + Integer.toString(getUByte(pending)) + "."
+ + Integer.toString(getUByte(pending)) + "."
+ + Integer.toString(getUByte(pending));
+ } else if (type == 0x03) {
+ host = getBLString(pending);
+ } else {
+ StringBuilder address = new StringBuilder();
+ for (int index = 0; index < 8; index++) {
+ if (index > 0) {
+ address.append(':');
+ }
+
address.append(Integer.toHexString(getUShort(pending)));
+ }
+ host = address.toString();
+ }
+ int port = getUShort(pending);
+ pending.compact();
+ if (log.isDebugEnabled()) {
log.debug("Received socks5 connection request to {}:{}",
host, port);
}
SshdSocketAddress remote = new SshdSocketAddress(host, port);
@@ -280,36 +338,73 @@ public class SocksProxy extends AbstractCloseable
implements IoHandler {
channel.setStreaming(Streaming.Async);
session.suspendRead();
service.registerChannel(channel);
+ state = Socks5State.OPENING_CHANNEL;
channel.open().addListener(this::onChannelOpened);
- } else {
- if (debugEnabled) {
- log.debug("Received socks5 connection message");
- }
- super.onMessage(buffer);
}
}
@SuppressWarnings("synthetic-access")
- protected void onChannelOpened(OpenFuture future) {
- session.resumeRead();
- int wpos = response.wpos();
- response.rpos(0);
- response.wpos(1);
+ protected synchronized void onChannelOpened(OpenFuture future) {
Throwable t = future.getException();
if (t != null) {
service.unregisterChannel(channel);
channel.close(true);
- response.putByte((byte) 0x01);
- } else {
- response.putByte((byte) 0x00);
+ pending.clear(true);
+ state = Socks5State.CLOSED;
+ sendReply((byte) 0x01, () -> session.close(false));
+ return;
}
- response.wpos(wpos);
+
+ state = Socks5State.FORWARDING;
+ sendReply((byte) 0x00, this::forwardPending);
+ }
+
+ protected void sendReply(byte status, Runnable completion) {
+ Buffer response = new ByteArrayBuffer(10, false);
+ // VER, REP, RSV, ATYP=IPv4, BND.ADDR=0.0.0.0, BND.PORT=0
+ response.putRawBytes(new byte[] { 0x05, status, 0x00, 0x01, 0, 0,
0, 0, 0, 0 });
try {
- session.writeBuffer(response);
+ session.writeBuffer(response).addListener(future -> {
+ if (future.isWritten()) {
+ completion.run();
+ } else {
+ session.close(true);
+ }
+ });
} catch (IOException e) {
log.error("Failed ({}) to send channel open response for {}:
{}", e.getClass().getSimpleName(), channel,
e.getMessage());
- throw new IllegalStateException("Failed to send packet", e);
+ session.close(true);
+ }
+ }
+
+ protected synchronized void forwardPending() {
+ if (state != Socks5State.FORWARDING) {
+ return;
+ }
+ Buffer payload = pending;
+ pending = new ByteArrayBuffer();
+ if (payload.available() == 0) {
+ session.resumeRead();
+ return;
+ }
+ int payloadSize = payload.available();
+ session.suspendRead();
+ try {
+ ThreadUtils.runAsInternal(channel.getAsyncIn(), out ->
out.writeBuffer(payload).addListener(future -> {
+ if (future.getException() == null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Forwarded {} queued socks5 bytes",
payloadSize);
+ }
+ session.resumeRead();
+ } else {
+ session.close(true);
+ }
+ }));
+ } catch (IOException e) {
+ log.error("Failed ({}) to forward {} queued bytes for {}: {}",
e.getClass().getSimpleName(), payloadSize,
+ channel, e.getMessage());
+ session.close(true);
}
}
diff --git a/sshd-core/src/test/java/org/apache/sshd/ProxyTest.java
b/sshd-core/src/test/java/org/apache/sshd/ProxyTest.java
index a99f16196..196dcf28d 100644
--- a/sshd-core/src/test/java/org/apache/sshd/ProxyTest.java
+++ b/sshd-core/src/test/java/org/apache/sshd/ProxyTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.sshd;
+import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -51,6 +52,7 @@ import org.junit.jupiter.api.TestMethodOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -319,6 +321,95 @@ public class ProxyTest extends BaseTestSupport {
}
}
+ @Test
+ void socks5HandlesFragmentedIPv4ConnectRequest() throws Exception {
+ final byte[] request = {
+ 5, 1, 0, 1, 127, 0, 0, 1,
+ (byte) (echoPort >>> Byte.SIZE), (byte) echoPort
+ };
+
+ assertFragmentedSocks5Connect(request);
+ }
+
+ @Test
+ void socks5HandlesFragmentedDomainConnectRequest() throws Exception {
+ final byte[] host = TEST_LOCALHOST.getBytes(StandardCharsets.US_ASCII);
+ final byte[] request = new byte[5 + host.length + 2];
+
+ request[0] = 5;
+ request[1] = 1;
+ request[2] = 0;
+ request[3] = 3;
+ request[4] = (byte) host.length;
+
+ System.arraycopy(host, 0, request, 5, host.length);
+
+ request[request.length - 2] = (byte) (echoPort >>> Byte.SIZE);
+ request[request.length - 1] = (byte) echoPort;
+
+ assertFragmentedSocks5Connect(request);
+ }
+
+ @Test
+ void socks5HandlesFragmentedIPv6ConnectRequest() throws Exception {
+ byte[] request = {
+ 5, 1, 0, 4,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, // ::1
+ (byte) (echoPort >>> Byte.SIZE), (byte) echoPort
+ };
+ assertFragmentedSocks5Connect(request);
+ }
+
+ /**
+ * Sends a SOCKS5 greeting and CONNECT request one byte at a time, in
separate write() calls, so the address and
+ * port are each split across multiple I/O callbacks, then sends
application data without waiting for the CONNECT
+ * reply. Verifies that the proxy produces exactly one clean CONNECT reply
and that the payload reaches the
+ * destination.
+ */
+ private void assertFragmentedSocks5Connect(final byte[] request) throws
Exception {
+ final byte[] payload =
getCurrentTestName().getBytes(StandardCharsets.UTF_8);
+
+ try (ClientSession session = createNativeSession(null);
+ DynamicPortForwardingTracker tracker
+ = session.createDynamicPortForwardingTracker(new
SshdSocketAddress(TEST_LOCALHOST, 0));
+
+ Socket socket = new Socket(TEST_LOCALHOST,
tracker.getBoundAddress().getPort())) {
+ socket.setSoTimeout((int) TimeUnit.SECONDS.toMillis(10L));
+
+ final OutputStream output = socket.getOutputStream();
+ final DataInputStream input = new
DataInputStream(socket.getInputStream());
+
+ output.write(5); // version
+ output.flush();
+ output.write(1); // method count
+ output.flush();
+ output.write(0); // no-auth
+ output.flush();
+
+ assertArrayEquals(new byte[] { 5, 0 }, readNBytes(input, 2),
"Unexpected greeting reply");
+
+ for (byte b : request) {
+ output.write(b);
+ output.flush();
+ }
+ output.write(payload); // application data sent before the CONNECT
reply is received
+ output.flush();
+
+ assertArrayEquals(new byte[] { 5, 0, 0, 1, 0, 0, 0, 0, 0, 0 },
readNBytes(input, 10),
+ "Unexpected SOCKS5 CONNECT reply");
+
+ assertArrayEquals(payload, readNBytes(input, payload.length),
+ "Payload was not forwarded intact exactly once");
+ }
+ }
+
+ private static byte[] readNBytes(final DataInputStream input, final int
len) throws IOException {
+ final byte[] data = new byte[len];
+ input.readFully(data);
+
+ return data;
+ }
+
protected ClientSession createNativeSession(PortForwardingEventListener
listener) throws Exception {
client = setupTestClient();
CoreModuleProperties.WINDOW_SIZE.set(client, 2048L);
