Skip to content

Commit cc644a1

Browse files
nhachichanhachicha
authored
JAVA-6194 Add MongoSocksProxyException for CMAP backpressure labeling (#1968)
JAVA-6035: Add backpressure flag to connection handshake (#1906)
1 parent f5f50b8 commit cc644a1

8 files changed

Lines changed: 520 additions & 47 deletions

File tree

‎driver-core/src/main/com/mongodb/MongoSocksProxyException.java‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
/*
2+
* Copyright 2008-present MongoDB, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.mongodb;
18+
19+
import com.mongodb.lang.Nullable;
20+
21+
/**
22+
* Thrown when an error occurs while connecting via a SOCKS5 proxy.
23+
*
24+
* <p>{@link #getProxyReplyCode()} returns the RFC 1928 reply code sent by the proxy when a
25+
* non-success CONNECT reply was successfully parsed; it returns {@code null} otherwise
26+
* (including for any failure that did not produce a parsed CONNECT reply, e.g. proxy TCP
27+
* connect failure, negotiation failure, authentication failure, failure connecting through
28+
* the proxy to the target server that did not yield a parsed reply, or an I/O error
29+
* mid-CONNECT).
30+
*
31+
* <p>RFC 1928 reply codes: 1=general failure, 2=connection not allowed by ruleset,
32+
* 3=network unreachable, 4=host unreachable, 5=connection refused, 6=TTL expired,
33+
* 7=command not supported, 8=address type not supported.
34+
*
35+
* @since 5.9
36+
*/
37+
public class MongoSocksProxyException extends MongoSocketOpenException {
38+
private static final long serialVersionUID = 1L;
39+
40+
@Nullable
41+
private final Integer proxyReplyCode;
42+
43+
/**
44+
* Construct an instance with no cause and no RFC 1928 reply code.
45+
*
46+
* @param message the message
47+
* @param serverAddress the server address
48+
*/
49+
public MongoSocksProxyException(final String message, final ServerAddress serverAddress) {
50+
this(message, serverAddress, null, null);
51+
}
52+
53+
/**
54+
* Construct an instance with a cause and no RFC 1928 reply code.
55+
*
56+
* @param message the message
57+
* @param serverAddress the server address
58+
* @param cause the cause
59+
*/
60+
public MongoSocksProxyException(final String message, final ServerAddress serverAddress, final Throwable cause) {
61+
this(message, serverAddress, cause, null);
62+
}
63+
64+
/**
65+
* Construct an instance with no cause and an RFC 1928 reply code from a parsed non-success
66+
* CONNECT reply.
67+
*
68+
* @param message the message
69+
* @param serverAddress the server address
70+
* @param proxyReplyCode the RFC 1928 reply code, or {@code null}
71+
*/
72+
public MongoSocksProxyException(final String message, final ServerAddress serverAddress,
73+
@Nullable final Integer proxyReplyCode) {
74+
super(message, serverAddress);
75+
this.proxyReplyCode = proxyReplyCode;
76+
}
77+
78+
/**
79+
* Construct an instance with a cause and an optional RFC 1928 reply code.
80+
*
81+
* @param message the message
82+
* @param serverAddress the server address
83+
* @param cause the cause, may be {@code null}
84+
* @param proxyReplyCode the RFC 1928 reply code, or {@code null}
85+
*/
86+
public MongoSocksProxyException(final String message, final ServerAddress serverAddress,
87+
@Nullable final Throwable cause, @Nullable final Integer proxyReplyCode) {
88+
super(message, serverAddress);
89+
this.proxyReplyCode = proxyReplyCode;
90+
if (cause != null) {
91+
initCause(cause);
92+
}
93+
}
94+
95+
/**
96+
* Returns the RFC 1928 reply code sent by the SOCKS5 proxy when a non-success CONNECT
97+
* reply was successfully parsed, or {@code null} otherwise.
98+
*
99+
* @return the RFC 1928 proxy reply code, or {@code null}
100+
*/
101+
@Nullable
102+
public Integer getProxyReplyCode() {
103+
return proxyReplyCode;
104+
}
105+
}

‎driver-core/src/main/com/mongodb/internal/connection/BackpressureErrorLabeler.java‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
import com.mongodb.MongoException;
2020
import com.mongodb.MongoSocketException;
21+
import com.mongodb.MongoSocksProxyException;
2122

2223
import javax.net.ssl.SSLHandshakeException;
2324
import javax.net.ssl.SSLPeerUnverifiedException;
@@ -76,19 +77,32 @@ static void applyLabelsIfEligible(final Throwable t) {
7677
return;
7778
}
7879
MongoSocketException socketException = (MongoSocketException) t;
80+
if (isSocksFailureNotEligibleForLabeling(socketException)) {
81+
return;
82+
}
7983
if (isDnsLookupFailure(socketException)) {
8084
return;
8185
}
8286
if (isTlsConfigurationError(socketException)) {
8387
return;
8488
}
85-
// TODO-BACKPRESSURE Nabil - Add SOCKS5 check once JAVA-6194 is introduced
86-
// async proxy error surfaces can be handled together — likely via a dedicated internal
87-
// exception thrown from the proxy code path.
8889
socketException.addLabel(MongoException.SYSTEM_OVERLOADED_ERROR_LABEL);
8990
socketException.addLabel(MongoException.RETRYABLE_ERROR_LABEL);
9091
}
9192

93+
private static boolean isSocksFailureNotEligibleForLabeling(final MongoSocketException t) {
94+
if (!(t instanceof MongoSocksProxyException)) {
95+
return false;
96+
}
97+
Integer replyCode = ((MongoSocksProxyException) t).getProxyReplyCode();
98+
if (replyCode == null) {
99+
return true;
100+
}
101+
return replyCode != SocksSocket.ServerReply.NET_UNREACHABLE.getReplyNumber()
102+
&& replyCode != SocksSocket.ServerReply.HOST_UNREACHABLE.getReplyNumber()
103+
&& replyCode != SocksSocket.ServerReply.CONN_REFUSED.getReplyNumber();
104+
}
105+
92106
private static boolean isDnsLookupFailure(final MongoSocketException t) {
93107
Throwable cause = t.getCause();
94108
while (cause != null) {

‎driver-core/src/main/com/mongodb/internal/connection/SocketStream.java‎

Lines changed: 63 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,18 @@
1616

1717
package com.mongodb.internal.connection;
1818

19+
import com.mongodb.MongoException;
20+
import com.mongodb.MongoInterruptedException;
1921
import com.mongodb.MongoSocketException;
2022
import com.mongodb.MongoSocketOpenException;
2123
import com.mongodb.MongoSocketReadException;
24+
import com.mongodb.MongoSocksProxyException;
2225
import com.mongodb.ServerAddress;
2326
import com.mongodb.connection.AsyncCompletionHandler;
2427
import com.mongodb.connection.ProxySettings;
2528
import com.mongodb.connection.SocketSettings;
2629
import com.mongodb.connection.SslSettings;
30+
import com.mongodb.lang.Nullable;
2731
import com.mongodb.spi.dns.InetAddressResolver;
2832
import org.bson.ByteBuf;
2933

@@ -38,6 +42,7 @@
3842
import java.net.SocketTimeoutException;
3943
import java.util.Iterator;
4044
import java.util.List;
45+
import java.util.Optional;
4146

4247
import static com.mongodb.assertions.Assertions.assertTrue;
4348
import static com.mongodb.assertions.Assertions.notNull;
@@ -79,13 +84,26 @@ public void open(final OperationContext operationContext) {
7984
socket = initializeSocket(operationContext);
8085
outputStream = socket.getOutputStream();
8186
inputStream = socket.getInputStream();
87+
} catch (MongoSocksProxyException e) {
88+
close();
89+
throw translateOr(e.getCause(), e);
8290
} catch (IOException e) {
8391
close();
84-
throw translateInterruptedException(e, "Interrupted while connecting")
85-
.orElseThrow(() -> new MongoSocketOpenException("Exception opening socket", getAddress(), e));
92+
throw translateOr(e, new MongoSocketOpenException("Exception opening socket", getAddress(), e));
8693
}
8794
}
8895

96+
/**
97+
* If {@code interruptCandidate} represents a thread interruption, returns the corresponding
98+
* {@link MongoInterruptedException}; otherwise returns {@code fallback}.
99+
*/
100+
private static MongoException translateOr(@Nullable final Throwable interruptCandidate,
101+
final MongoException fallback) {
102+
Optional<MongoInterruptedException> translated =
103+
translateInterruptedException(interruptCandidate, "Interrupted while connecting");
104+
return translated.isPresent() ? translated.get() : fallback;
105+
}
106+
89107
protected Socket initializeSocket(final OperationContext operationContext) throws IOException {
90108
ProxySettings proxySettings = settings.getProxySettings();
91109
if (proxySettings.isProxyEnabled()) {
@@ -119,15 +137,28 @@ private SSLSocket initializeSslSocketOverSocksProxy(final OperationContext opera
119137
final int serverPort = address.getPort();
120138

121139
SocksSocket socksProxy = new SocksSocket(settings.getProxySettings());
122-
configureSocket(socksProxy, operationContext, settings);
123-
InetSocketAddress inetSocketAddress = toSocketAddress(serverHost, serverPort);
124-
socksProxy.connect(inetSocketAddress, operationContext.getTimeoutContext().getConnectTimeoutMs());
125-
126-
SSLSocket sslSocket = (SSLSocket) sslSocketFactory.createSocket(socksProxy, serverHost, serverPort, true);
127-
//Even though Socks proxy connection is already established, TLS handshake has not been performed yet.
128-
//So it is possible to set SSL parameters before handshake is done.
129-
configureSslSocket(sslSocket, sslSettings, inetSocketAddress);
130-
return sslSocket;
140+
// Track the outermost socket layer to close on failure. Initially this is socksProxy;
141+
// once we wrap it into an SSLSocket, that becomes the outermost layer and closing it
142+
// tears down the underlying socksProxy as well.
143+
Socket toClose = socksProxy;
144+
try {
145+
configureSocket(socksProxy, operationContext, settings);
146+
InetSocketAddress inetSocketAddress = toSocketAddress(serverHost, serverPort);
147+
socksProxy.connect(inetSocketAddress, operationContext.getTimeoutContext().getConnectTimeoutMs());
148+
SSLSocket sslSocket = (SSLSocket) sslSocketFactory.createSocket(socksProxy, serverHost, serverPort, true);
149+
toClose = sslSocket;
150+
//Even though Socks proxy connection is already established, TLS handshake has not been performed yet.
151+
//So it is possible to set SSL parameters before handshake is done.
152+
configureSslSocket(sslSocket, sslSettings, inetSocketAddress);
153+
return sslSocket;
154+
} catch (IOException | RuntimeException e) {
155+
try {
156+
toClose.close();
157+
} catch (IOException closeException) {
158+
e.addSuppressed(closeException);
159+
}
160+
throw e;
161+
}
131162
}
132163

133164

@@ -141,17 +172,27 @@ private static InetSocketAddress toSocketAddress(final String serverHost, final
141172

142173
private Socket initializeSocketOverSocksProxy(final OperationContext operationContext) throws IOException {
143174
Socket createdSocket = socketFactory.createSocket();
144-
configureSocket(createdSocket, operationContext, settings);
145-
/*
146-
Wrap the configured socket with SocksSocket to add extra functionality.
147-
Reason for separate steps: We can't directly extend Java 11 methods within 'SocksSocket'
148-
to configure itself.
149-
*/
150-
SocksSocket socksProxy = new SocksSocket(createdSocket, settings.getProxySettings());
151-
152-
socksProxy.connect(toSocketAddress(address.getHost(), address.getPort()),
153-
operationContext.getTimeoutContext().getConnectTimeoutMs());
154-
return socksProxy;
175+
try {
176+
configureSocket(createdSocket, operationContext, settings);
177+
/*
178+
Wrap the configured socket with SocksSocket to add extra functionality.
179+
Reason for separate steps: We can't directly extend Java 11 methods within 'SocksSocket'
180+
to configure itself.
181+
*/
182+
SocksSocket socksProxy = new SocksSocket(createdSocket, settings.getProxySettings());
183+
socksProxy.connect(toSocketAddress(address.getHost(), address.getPort()),
184+
operationContext.getTimeoutContext().getConnectTimeoutMs());
185+
return socksProxy;
186+
} catch (IOException | RuntimeException e) {
187+
// SocksSocket.connect() closes itself on failure, but createdSocket may not yet
188+
// be owned by a SocksSocket (e.g. configureSocket threw). Close defensively;
189+
try {
190+
createdSocket.close();
191+
} catch (IOException closeException) {
192+
e.addSuppressed(closeException);
193+
}
194+
throw e;
195+
}
155196
}
156197

157198
@Override

0 commit comments

Comments
 (0)