This is an automated email from the ASF dual-hosted git repository.
Cole-Greer pushed a commit to branch 3.7-dev
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/3.7-dev by this push:
new d429ec0bd8 Cap and expire requests deferred during SASL authentication
d429ec0bd8 is described below
commit d429ec0bd8cba4885b12054b63a44fe14df97afc
Author: Guian Gumpac <[email protected]>
AuthorDate: Tue Sep 1 12:28:15 2026 -0700
Cap and expire requests deferred during SASL authentication
SaslAuthenticationHandler now bounds how many requests and how many
request bytes a channel may retain while its SASL authentication is
pending, and closes a channel whose authentication does not complete in
time. A request beyond a limit is answered UNAUTHORIZED rather than
retained, and can be resent once the channel is authenticated.
All three limits are configurable as authentication.maxDeferredRequests
(64), authentication.maxPreAuthRetainedBytes (2097152) and
authentication.preAuthTimeout (30000 ms). Each must be greater than
zero or Gremlin Server fails to start.
MAX_REQUEST_DEFERRABLE_DURATION is now the default for
authentication.preAuthTimeout, and its value moved from 5 to 30 seconds.
A request that a failed authentication attempt already answered is no
longer replayed when a later attempt on the same channel succeeds.
Assisted-by: Claude Code:claude-opus-5
---
CHANGELOG.asciidoc | 1 +
docs/src/dev/provider/index.asciidoc | 8 +-
docs/src/reference/gremlin-applications.asciidoc | 3 +
docs/src/upgrade/release-3.7.x.asciidoc | 38 +
.../apache/tinkerpop/gremlin/server/Settings.java | 37 +
.../server/handler/SaslAuthenticationHandler.java | 225 +++++-
.../tinkerpop/gremlin/server/handler/StateKey.java | 16 +
.../handler/WsGremlinBinaryRequestDecoder.java | 3 +
.../handler/WsGremlinCloseRequestDecoder.java | 3 +
.../handler/WsGremlinTextRequestDecoder.java | 3 +
.../server/GremlinServerAuthIntegrateTest.java | 60 +-
.../handler/SaslAuthenticationHandlerTest.java | 826 +++++++++++++++++++++
.../handler/WsGremlinRequestDecoderTest.java | 126 ++++
13 files changed, 1286 insertions(+), 63 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index f45d47a595..4e0bcae085 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -61,6 +61,7 @@
image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Fixed request isolation in HTTP and WebSocket authorization handlers.
* Fixed `gremlin-python` GraphBinary serialization of
`BigInteger`/`BigDecimal` negative boundary values (e.g. `-129`) that raised
`OverflowError`.
* Fixed `gremlin-go` GraphBinary serialization of zero
`BigInteger`/`BigDecimal` values, which were encoded with zero length and
rejected by Java servers.
+* Capped by count and by total size the requests `SaslAuthenticationHandler`
retains on a channel while its SASL authentication is pending, and closed the
channel when that authentication does not complete within 30 seconds, all three
tunable with the new `authentication.maxDeferredRequests`,
`authentication.maxPreAuthRetainedBytes` and `authentication.preAuthTimeout`
settings.
[[release-3-7-6]]
=== TinkerPop 3.7.6 (Release Date: April 1, 2026)
diff --git a/docs/src/dev/provider/index.asciidoc
b/docs/src/dev/provider/index.asciidoc
index 376bc22e83..98c0272e19 100644
--- a/docs/src/dev/provider/index.asciidoc
+++ b/docs/src/dev/provider/index.asciidoc
@@ -1332,15 +1332,17 @@ Other SASL mechanisms supported for selected clients
are listed in the
link:https://tinkerpop.apache.org/docs/x.y.z/reference/#security[security
section of the Gremlin Server reference documentation].
When authentication is enabled, an incoming request is intercepted before it
is evaluated by the `ScriptEngine`. The
-request is saved on the server and a `AUTHENTICATE` challenge response (status
code `407`) is returned to the client.
+request is saved on the server, unless it is too large to be held pending
authentication, and a `AUTHENTICATE`
+challenge response (status code `407`) is returned to the client.
The client will detect the `AUTHENTICATE` and respond with an `authentication`
for the `op` and an `arg` named `sasl`.
In case of the "PLAIN" SASL mechanism the `arg` contains the password. The
password should be either, an encoded
sequence of UTF-8 bytes, delimited by 0 (US-ASCII NUL), where the form is :
`<NUL>username<NUL>password`, or a Base64
encoded string of the former (which in this instance would be
`AHVzZXJuYW1lAHBhc3N3b3Jk`). Should Gremlin Server be
able to authenticate with the provided credentials, the server will return the
results of the original request as it
-normally does without authentication. If it cannot authenticate given the
challenge response from the client, it will
-return `UNAUTHORIZED` (status code `401`).
+normally does without authentication. A request that was not held, either
because it was too large or because an
+earlier failed authentication attempt already answered it, is not replayed and
must be resent. If it cannot
+authenticate given the challenge response from the client, it will return
`UNAUTHORIZED` (status code `401`).
NOTE: Gremlin Server does not support the "authorization identity" as
described in link:https://tools.ietf.org/html/rfc4616[RFC4616].
diff --git a/docs/src/reference/gremlin-applications.asciidoc
b/docs/src/reference/gremlin-applications.asciidoc
index 81c157247f..722cf561f9 100644
--- a/docs/src/reference/gremlin-applications.asciidoc
+++ b/docs/src/reference/gremlin-applications.asciidoc
@@ -1028,6 +1028,9 @@ The following table describes the various YAML
configuration options that Gremli
|authentication.authenticator |The fully qualified classname of an
`Authenticator` implementation to use. If this setting is not present, then
authentication is effectively disabled. |`AllowAllAuthenticator`
|authentication.authenticationHandler | The fully qualified classname of an
`AbstractAuthenticationHandler` implementation to use. If this setting is not
present, but the `authentication.authenticator` is, it will use that
authenticator with the default `AbstractAuthenticationHandler` implementation
for the specified `Channelizer` |_none_
|authentication.config |A `Map` of configuration settings to be passed to the
`Authenticator` when it is constructed. The settings available are dependent
on the implementation. |_none_
+|authentication.maxDeferredRequests |The number of requests a channel may
defer while its SASL authentication is pending. A request beyond this number is
answered with `UNAUTHORIZED` rather than deferred. Must be greater than zero.
This configuration only applies to the `SaslAuthenticationHandler` and its
subclasses. |64
+|authentication.maxPreAuthRetainedBytes |The total request bytes a channel may
have retained while its SASL authentication is pending, covering the request
that began authentication as well as the deferred ones. A request beyond this
total is answered with `UNAUTHORIZED` rather than retained. Must be greater
than zero. This configuration only applies to the `SaslAuthenticationHandler`
and its subclasses. |2097152
+|authentication.preAuthTimeout |Time in milliseconds a channel may stay
unauthenticated, covering the whole SASL handshake and not just the deferral
window, after which anything the channel still has pending is answered with
`UNAUTHORIZED` and the channel is closed. Must be greater than zero. This
configuration only applies to the `SaslAuthenticationHandler` and its
subclasses. |30000
|authorization.authorizer |The fully qualified classname of an `Authorizer`
implementation to use. |_none_
|authorization.config |A `Map` of configuration settings to be passed to the
`Authorizer` when it is constructed. The settings available are dependent on
the implementation. |_none_
|channelizer |The fully qualified classname of the `Channelizer`
implementation to use. A `Channelizer` is a "channel initializer" which
Gremlin Server uses to define the type of processing pipeline to use. By
allowing different `Channelizer` implementations, Gremlin Server can support
different communication protocols (e.g. WebSocket). |`WebSocketChannelizer`
diff --git a/docs/src/upgrade/release-3.7.x.asciidoc
b/docs/src/upgrade/release-3.7.x.asciidoc
index 380a70cddd..df795f61d6 100644
--- a/docs/src/upgrade/release-3.7.x.asciidoc
+++ b/docs/src/upgrade/release-3.7.x.asciidoc
@@ -274,6 +274,44 @@ registered by default. An unregistered custom strategy
produces an error such as
See:
link:https://tinkerpop.apache.org/docs/3.7.7/reference/#traversalstrategy-registration[TraversalStrategy
Registration]
+==== Pre-Authentication Request Limits
+
+Gremlin Server now bounds what a channel can make it retain while that
channel's SASL authentication is pending, and
+terminates a channel whose authentication does not complete. Only deployments
that configure an `Authenticator` are
+affected, as the handler is not in the pipeline for the default
`AllowAllAuthenticator`.
+
+A channel may defer at most 64 requests, and at most 2 MiB of total request
bytes covering both the request that began
+authentication and the deferred ones. A request beyond either cap is answered
with `UNAUTHORIZED` (status code `401`)
+rather than deferred, and can be resent once the channel is authenticated.
+
+A request larger than 2 MiB that arrives first on a channel still receives the
`AUTHENTICATE` challenge, but it is not
+held, so it is answered the same way. Note that `maxContentLength` defaults to
10 MiB, so a request the server
+otherwise accepts can be affected. Because drivers open pooled connections on
demand, the first request on a new
+connection is an ordinary user request, so this can appear intermittently for
workloads that send bulk data inside a
+query, such as a large `inject()` or a `within()` over many ids.
+
+`MAX_REQUEST_DEFERRABLE_DURATION` was raised from 5 seconds to 30 seconds, and
a channel that has not completed
+authentication within that duration is now closed. Previously the handler
never closed such a channel, and
+`idleConnectionTimeout` defaults to `0`, so nothing reaped it. A request that
a failed authentication attempt already
+answered is no longer replayed when a later attempt on the same channel
succeeds.
+
+All three limits are configurable, as `authentication.maxDeferredRequests`,
`authentication.maxPreAuthRetainedBytes`
+and `authentication.preAuthTimeout`, the last in milliseconds. Each must be
greater than zero and Gremlin Server fails
+to start otherwise. The defaults are the values described above, so a
configuration that sets none of them behaves as
+described.
+
+Each of the errors below is a resource limit rather than a credential failure,
so an operator seeing the `UNAUTHORIZED`
+status code they share need not investigate credentials. The messages show the
default limits.
+
+[cols="1,3,3", options="header"]
+|=========================================================
+|Status Code |Message |Remedy
+|`401` |`Too many requests were deferred pending authentication (64 maximum).`
|Resend once authenticated, or raise `authentication.maxDeferredRequests`.
+|`401` |`Too many bytes were retained pending authentication (2097152 bytes
maximum).` |Resend once authenticated, or raise
`authentication.maxPreAuthRetainedBytes`.
+|`401` |`Request is too large to hold pending authentication (2097152 bytes
maximum).` |Resend once authenticated, or raise
`authentication.maxPreAuthRetainedBytes`.
+|`401` |`Authentication did not finish in the allowed duration (30000 ms).`
|Complete the SASL handshake sooner, or raise `authentication.preAuthTimeout`.
+|=========================================================
+
=== Upgrading for Providers
==== Graph System Providers
diff --git
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java
index 210bee930b..b60f2f337e 100644
---
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java
+++
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java
@@ -31,6 +31,7 @@ import org.apache.tinkerpop.gremlin.server.authz.Authorizer;
import org.apache.tinkerpop.gremlin.server.channel.UnifiedChannelizer;
import org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer;
import
org.apache.tinkerpop.gremlin.server.handler.AbstractAuthenticationHandler;
+import org.apache.tinkerpop.gremlin.server.handler.SaslAuthenticationHandler;
import org.apache.tinkerpop.gremlin.server.handler.Session;
import org.apache.tinkerpop.gremlin.server.util.DefaultGraphManager;
import org.apache.tinkerpop.gremlin.server.util.LifeCycleHook;
@@ -491,6 +492,22 @@ public class Settings {
* Settings for the {@link Authenticator} implementation.
*/
public static class AuthenticationSettings {
+ /**
+ * Default for {@link #maxDeferredRequests}.
+ */
+ public static final int DEFAULT_MAX_DEFERRED_REQUESTS = 64;
+
+ /**
+ * Default for {@link #maxPreAuthRetainedBytes}.
+ */
+ public static final long DEFAULT_MAX_PRE_AUTH_RETAINED_BYTES = 2L *
1024 * 1024;
+
+ /**
+ * Default for {@link #preAuthTimeout}.
+ */
+ public static final long DEFAULT_PRE_AUTH_TIMEOUT =
+
SaslAuthenticationHandler.MAX_REQUEST_DEFERRABLE_DURATION.toMillis();
+
/**
* The fully qualified class name of the {@link Authenticator}
implementation. This class name will be
* used to load the implementation from the classpath. Defaults to
{@link AllowAllAuthenticator} when
@@ -510,6 +527,26 @@ public class Settings {
* {@link Authenticator} implementation for specifics on what
configurations are expected.
*/
public Map<String, Object> config = null;
+
+ /**
+ * Number of requests a channel may defer while its SASL
authentication is pending. Must be greater than zero.
+ * Only applies to {@link SaslAuthenticationHandler} and its
subclasses.
+ */
+ public int maxDeferredRequests = DEFAULT_MAX_DEFERRED_REQUESTS;
+
+ /**
+ * Total wire bytes a channel may have retained while its SASL
authentication is pending, covering the request
+ * that started it as well as the deferred ones. Deserialized, their
heap footprint is a multiple of this. Must
+ * be greater than zero. Only applies to {@link
SaslAuthenticationHandler} and its subclasses.
+ */
+ public long maxPreAuthRetainedBytes =
DEFAULT_MAX_PRE_AUTH_RETAINED_BYTES;
+
+ /**
+ * Time in milliseconds a channel may stay unauthenticated, covering
the whole handshake and not just the
+ * deferral window, after which it is closed. Must be greater than
zero. Only applies to
+ * {@link SaslAuthenticationHandler} and its subclasses.
+ */
+ public long preAuthTimeout = DEFAULT_PRE_AUTH_TIMEOUT;
}
/**
diff --git
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SaslAuthenticationHandler.java
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SaslAuthenticationHandler.java
index f22d465e29..50ac3b9fc5 100644
---
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SaslAuthenticationHandler.java
+++
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SaslAuthenticationHandler.java
@@ -22,6 +22,7 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.Attribute;
+import io.netty.util.concurrent.ScheduledFuture;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.tinkerpop.gremlin.server.GremlinServer;
@@ -44,8 +45,10 @@ import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Base64;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import java.util.function.Function;
/**
@@ -59,7 +62,11 @@ public class SaslAuthenticationHandler extends
AbstractAuthenticationHandler {
private static final Logger logger =
LoggerFactory.getLogger(SaslAuthenticationHandler.class);
private static final Base64.Decoder BASE64_DECODER = Base64.getDecoder();
private static final Base64.Encoder BASE64_ENCODER = Base64.getEncoder();
- public static final Duration MAX_REQUEST_DEFERRABLE_DURATION =
Duration.ofSeconds(5);
+ /**
+ * Default for {@code settings.authentication.preAuthTimeout}, how long a
channel may stay unauthenticated,
+ * covering the whole handshake, not just the deferral window.
+ */
+ public static final Duration MAX_REQUEST_DEFERRABLE_DURATION =
Duration.ofSeconds(30);
private static final Logger auditLogger =
LoggerFactory.getLogger(GremlinServer.AUDIT_LOGGER_NAME);
protected final Settings settings;
@@ -74,6 +81,23 @@ public class SaslAuthenticationHandler extends
AbstractAuthenticationHandler {
public SaslAuthenticationHandler(final Authenticator authenticator, final
Authorizer authorizer, final Settings settings) {
super(authenticator, authorizer);
+
+ // rejected here so that a bad configuration fails startup rather than
a request
+ if (settings.authentication.maxDeferredRequests < 1)
+ throw new IllegalStateException(String.format(
+ "authentication.maxDeferredRequests must be greater than
zero but was %s",
+ settings.authentication.maxDeferredRequests));
+
+ if (settings.authentication.maxPreAuthRetainedBytes < 1)
+ throw new IllegalStateException(String.format(
+ "authentication.maxPreAuthRetainedBytes must be greater
than zero but was %s",
+ settings.authentication.maxPreAuthRetainedBytes));
+
+ if (settings.authentication.preAuthTimeout < 1)
+ throw new IllegalStateException(String.format(
+ "authentication.preAuthTimeout must be greater than zero
but was %s",
+ settings.authentication.preAuthTimeout));
+
this.settings = settings;
}
@@ -91,15 +115,43 @@ public class SaslAuthenticationHandler extends
AbstractAuthenticationHandler {
final Attribute<Authenticator.SaslNegotiator> negotiator =
ctx.channel().attr(StateKey.NEGOTIATOR);
final Attribute<RequestMessage> request =
ctx.channel().attr(StateKey.REQUEST_MESSAGE);
final Attribute<Pair<LocalDateTime, List<RequestMessage>>>
deferredRequests = ctx.channel().attr(StateKey.DEFERRED_REQUEST_MESSAGES);
+ final Attribute<Long> retainedBytes =
ctx.channel().attr(StateKey.DEFERRED_REQUEST_BYTES);
+
+ final long maxRetainedBytes =
settings.authentication.maxPreAuthRetainedBytes;
if (negotiator.get() == null) {
+ final long incomingRequestSize =
incomingRequestSize(ctx.channel());
+
+ // the request is held until authentication completes, so on its
own it has to fit what may be retained
+ final boolean tooLargeToRetain = incomingRequestSize >
maxRetainedBytes;
+
+ if (tooLargeToRetain) {
+ logger.debug("Not retaining the {} byte request from {}
pending authentication - {} bytes maximum",
+ incomingRequestSize, ctx.channel().remoteAddress(),
maxRetainedBytes);
+ }
+
try {
// First time through so save the request and send an
AUTHENTICATE challenge with no data
negotiator.set(authenticator.newSaslNegotiator(getRemoteInetAddress(ctx)));
- request.set(requestMessage);
+
+ // retention starts here, so the deadline on it does too
+ armDeadline(ctx);
+
+ if (!tooLargeToRetain) {
+ request.set(requestMessage);
+ retainedBytes.set(incomingRequestSize);
+ }
+
final ResponseMessage authenticate =
ResponseMessage.build(requestMessage)
.code(ResponseStatusCode.AUTHENTICATE).create();
ctx.writeAndFlush(authenticate);
+
+ // answered after the challenge, so authentication can still
complete and the request be resent
+ if (tooLargeToRetain) {
+ ctx.writeAndFlush(ResponseMessage.build(requestMessage)
+ .statusMessage("Request is too large to hold
pending authentication (" + maxRetainedBytes + " bytes maximum).")
+ .code(ResponseStatusCode.UNAUTHORIZED).create());
+ }
} catch (Exception ex) {
// newSaslNegotiator can cause troubles - if we don't catch
and respond nicely the driver seems
// to hang until timeout which isn't so nice. treating this
like a server error as it means that
@@ -116,20 +168,45 @@ public class SaslAuthenticationHandler extends
AbstractAuthenticationHandler {
return;
} else if (!requestMessage.getOp().equals(Tokens.OPS_AUTHENTICATION)) {
// If authentication negotiation is pending, store subsequent
non-authentication requests for later processing
- deferredRequests.setIfAbsent(new
ImmutablePair<>(LocalDateTime.now(), new ArrayList<>()));
- deferredRequests.get().getValue().add(requestMessage);
+ final Pair<LocalDateTime, List<RequestMessage>> deferred =
deferredRequests.get();
- final Duration deferredDuration =
Duration.between(deferredRequests.get().getKey(), LocalDateTime.now());
+ // bounds what an unauthenticated channel can make the server
retain, by count and by size
+ final long alreadyRetainedBytes = retainedBytes.get() == null ? 0L
: retainedBytes.get();
+ final long incomingRequestSize =
incomingRequestSize(ctx.channel());
- if (deferredDuration.compareTo(MAX_REQUEST_DEFERRABLE_DURATION) >
0) {
- respondWithError(
- requestMessage,
- builder -> builder.statusMessage("Authentication did
not finish in the allowed duration (" + MAX_REQUEST_DEFERRABLE_DURATION + "s).")
- .code(ResponseStatusCode.UNAUTHORIZED),
- ctx);
+ final int maxDeferredRequests =
settings.authentication.maxDeferredRequests;
+
+ final String breachMessage;
+ if (deferred != null && deferred.getValue().size() >=
maxDeferredRequests) {
+ breachMessage = "Too many requests were deferred pending
authentication (" + maxDeferredRequests + " maximum).";
+ } else if (alreadyRetainedBytes + incomingRequestSize >
maxRetainedBytes) {
+ breachMessage = "Too many bytes were retained pending
authentication (" + maxRetainedBytes + " bytes maximum).";
+ } else {
+ breachMessage = null;
+ }
+
+ if (breachMessage != null) {
+ logger.debug("Rejecting the request from {} - {}",
ctx.channel().remoteAddress(), breachMessage);
+
+ // dropped rather than buffered for a peer that has stopped
reading
+ if (ctx.channel().isWritable()) {
+ ctx.writeAndFlush(ResponseMessage.build(requestMessage)
+
.statusMessage(breachMessage).code(ResponseStatusCode.UNAUTHORIZED).create());
+ }
+ return;
+ }
+
+ if (deferred == null) {
+ deferredRequests.set(new ImmutablePair<>(LocalDateTime.now(),
new ArrayList<>()));
+ } else if (Duration.between(deferred.getKey(),
LocalDateTime.now()).toMillis() > preAuthTimeout()) {
+ // answered here rather than deferred, so that one request id
cannot be answered twice
+ respondWithError(requestMessage, this::didNotFinishInTime,
ctx);
return;
}
+ deferredRequests.get().getValue().add(requestMessage);
+ retainedBytes.set(alreadyRetainedBytes + incomingRequestSize);
+
return;
} else if (!requestMessage.getArgs().containsKey(Tokens.ARGS_SASL)) {
// This is an authentication request that is missing a "sasl"
argument.
@@ -169,6 +246,7 @@ public class SaslAuthenticationHandler extends
AbstractAuthenticationHandler {
final org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser
user = negotiator.get().getAuthenticatedUser();
ctx.channel().attr(StateKey.AUTHENTICATED_USER).set(user);
+ cancelDeadline(ctx.channel());
// User name logged with the remote socket address and
authenticator classname for audit logging
if (settings.enableAuditLog) {
String address = ctx.channel().remoteAddress().toString();
@@ -177,16 +255,20 @@ public class SaslAuthenticationHandler extends
AbstractAuthenticationHandler {
auditLogger.info("User {} with address {} authenticated by {}",
user.getName(), address,
authClassParts[authClassParts.length - 1]);
}
+ final List<RequestMessage> deferred =
releaseDeferredRequests(ctx.channel());
+
// If we have got here we are authenticated so remove the handler
and pass
// the original message down the pipeline for processing
ctx.pipeline().remove(this);
final RequestMessage original = request.get();
- ctx.fireChannelRead(original);
- // Also send deferred requests if there are any down the pipeline
for processing
- if (deferredRequests.get() != null) {
-
deferredRequests.getAndSet(null).getValue().forEach(ctx::fireChannelRead);
+ // null when an earlier failed attempt already answered it
+ if (original != null) {
+ ctx.fireChannelRead(original);
}
+
+ // Also send deferred requests if there are any down the pipeline
for processing
+ deferred.forEach(ctx::fireChannelRead);
} catch (AuthenticationException ae) {
respondWithError(
requestMessage,
@@ -195,27 +277,114 @@ public class SaslAuthenticationHandler extends
AbstractAuthenticationHandler {
}
}
- private void respondWithError(final RequestMessage requestMessage, final
Function<ResponseMessage.Builder, ResponseMessage.Builder> buildResponse, final
ChannelHandlerContext ctx) {
- final Attribute<RequestMessage> originalRequest =
ctx.channel().attr(StateKey.REQUEST_MESSAGE);
- final Attribute<Pair<LocalDateTime, List<RequestMessage>>>
deferredRequests = ctx.channel().attr(StateKey.DEFERRED_REQUEST_MESSAGES);
+ /**
+ * Milliseconds a channel may stay unauthenticated.
+ */
+ private long preAuthTimeout() {
+ return settings.authentication.preAuthTimeout;
+ }
+
+ /**
+ * Schedules the task that ends an authentication that does not finish in
{@link #preAuthTimeout()}.
+ */
+ private void armDeadline(final ChannelHandlerContext ctx) {
+ final Channel channel = ctx.channel();
+ final Attribute<ScheduledFuture<?>> deadline =
channel.attr(StateKey.PREAUTH_DEADLINE);
+ // one task and one close listener per channel, however often a null
negotiator sends us back here
+ if (deadline.get() != null) return;
+
+ // so that the set below can never orphan a live task
+ cancelDeadline(channel);
+
+ deadline.set(channel.eventLoop().schedule(() -> expireDeadline(ctx),
+ preAuthTimeout(), TimeUnit.MILLISECONDS));
+ channel.closeFuture().addListener(future -> cancelDeadline(channel));
+ }
+
+ /**
+ * Cancels the deadline, if one is armed. Only authentication succeeding
and the channel closing get here.
+ */
+ private static void cancelDeadline(final Channel channel) {
+ final ScheduledFuture<?> deadline =
channel.attr(StateKey.PREAUTH_DEADLINE).getAndSet(null);
+
+ if (deadline != null) deadline.cancel(false);
+ }
+
+ /**
+ * Answers whatever the channel still has pending and closes it.
+ */
+ private void expireDeadline(final ChannelHandlerContext ctx) {
+ final Channel channel = ctx.channel();
+
+ if (channel.attr(StateKey.AUTHENTICATED_USER).get() != null ||
!channel.isActive()) return;
+
+ logger.debug("Closing the channel to {} - authentication did not
finish in {} ms",
+ channel.remoteAddress(), preAuthTimeout());
+
+ answerPendingRequests(this::didNotFinishInTime, ctx);
+
+ // unconditional, as a peer that never accepts the responses must not
be able to hold the channel open
+ ctx.close();
+ }
+
+ private ResponseMessage.Builder didNotFinishInTime(final
ResponseMessage.Builder builder) {
+ return builder.statusMessage("Authentication did not finish in the
allowed duration (" + preAuthTimeout() + " ms).")
+ .code(ResponseStatusCode.UNAUTHORIZED);
+ }
+
+ /**
+ * Size of the frame the request being processed was decoded from, or zero
when the transport recorded none.
+ */
+ private static long incomingRequestSize(final Channel channel) {
+ final Integer requestSize = channel.attr(StateKey.REQUEST_SIZE).get();
+
+ return requestSize == null ? 0L : requestSize;
+ }
+
+ /**
+ * Drops the running byte total, returning the requests the channel had
deferred.
+ */
+ private static List<RequestMessage> releaseDeferredRequests(final Channel
channel) {
+ final Pair<LocalDateTime, List<RequestMessage>> deferred =
+
channel.attr(StateKey.DEFERRED_REQUEST_MESSAGES).getAndSet(null);
+ channel.attr(StateKey.DEFERRED_REQUEST_BYTES).set(0L);
+
+ return deferred == null ? Collections.emptyList() :
deferred.getValue();
+ }
+
+ /**
+ * Answers the stashed and deferred requests, releasing both, plus {@code
requestMessage} unless it is an
+ * authentication request.
+ */
+ private void respondWithError(final RequestMessage requestMessage, final
Function<ResponseMessage.Builder, ResponseMessage.Builder> buildResponse, final
ChannelHandlerContext ctx) {
if (!requestMessage.getOp().equals(Tokens.OPS_AUTHENTICATION)) {
ctx.write(buildResponse.apply(ResponseMessage.build(requestMessage)).create());
}
- if (originalRequest.get() != null) {
-
ctx.write(buildResponse.apply(ResponseMessage.build(originalRequest.get())).create());
- }
+ answerPendingRequests(buildResponse, ctx);
+ }
+
+ /**
+ * Answers the stashed and deferred requests, releasing both.
+ */
+ private static void answerPendingRequests(final
Function<ResponseMessage.Builder, ResponseMessage.Builder> buildResponse, final
ChannelHandlerContext ctx) {
+ final Attribute<RequestMessage> originalRequest =
ctx.channel().attr(StateKey.REQUEST_MESSAGE);
- if (deferredRequests.get() != null) {
- deferredRequests
- .getAndSet(null).getValue().stream()
- .map(ResponseMessage::build)
- .map(buildResponse)
- .map(ResponseMessage.Builder::create)
- .forEach(ctx::write);
+ // cleared as it is answered so that one request id cannot be answered
twice
+ final RequestMessage stashedRequest = originalRequest.getAndSet(null);
+
+ if (stashedRequest != null) {
+
ctx.write(buildResponse.apply(ResponseMessage.build(stashedRequest)).create());
}
+ // this also drops the stashed request's share of the retained byte
total
+ releaseDeferredRequests(ctx.channel()).stream()
+ .map(ResponseMessage::build)
+ .map(buildResponse)
+ .map(ResponseMessage.Builder::create)
+ .forEach(ctx::write);
+
ctx.flush();
}
diff --git
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/StateKey.java
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/StateKey.java
index 2eb2bc8655..8a71dbaabb 100644
---
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/StateKey.java
+++
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/StateKey.java
@@ -25,6 +25,7 @@ import
org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser;
import org.apache.tinkerpop.gremlin.server.auth.Authenticator;
import org.apache.tinkerpop.gremlin.server.op.session.Session;
import io.netty.util.AttributeKey;
+import io.netty.util.concurrent.ScheduledFuture;
import java.time.LocalDateTime;
import java.util.List;
@@ -72,4 +73,19 @@ public final class StateKey {
* The key for the current {@link AuthenticatedUser}.
*/
public static final AttributeKey<AuthenticatedUser> AUTHENTICATED_USER =
AttributeKey.valueOf("authenticatedUser");
+
+ /**
+ * The key for the size in bytes of the frame the current request was
decoded from. Not public by design.
+ */
+ static final AttributeKey<Integer> REQUEST_SIZE =
AttributeKey.valueOf("requestSize");
+
+ /**
+ * The key for the running total of the sizes of the requests retained
pending authentication. Not public by design.
+ */
+ static final AttributeKey<Long> DEFERRED_REQUEST_BYTES =
AttributeKey.valueOf("deferredRequestBytes");
+
+ /**
+ * The key for the task that closes the channel if authentication does not
complete in time. Not public by design.
+ */
+ static final AttributeKey<ScheduledFuture<?>> PREAUTH_DEADLINE =
AttributeKey.valueOf("preAuthDeadline");
}
diff --git
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinBinaryRequestDecoder.java
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinBinaryRequestDecoder.java
index cd645de839..ea9a708016 100644
---
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinBinaryRequestDecoder.java
+++
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinBinaryRequestDecoder.java
@@ -53,6 +53,9 @@ public class WsGremlinBinaryRequestDecoder extends
MessageToMessageDecoder<Binar
@Override
protected void decode(final ChannelHandlerContext channelHandlerContext,
final BinaryWebSocketFrame frame, final List<Object> objects) throws Exception {
+ // recorded before anything reads from the frame as readableBytes() is
relative to the reader index
+
channelHandlerContext.channel().attr(StateKey.REQUEST_SIZE).set(frame.content().readableBytes());
+
final ByteBuf messageBytes = frame.content();
final byte len = messageBytes.readByte();
if (len <= 0) {
diff --git
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinCloseRequestDecoder.java
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinCloseRequestDecoder.java
index f73dab5be0..3324e09026 100644
---
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinCloseRequestDecoder.java
+++
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinCloseRequestDecoder.java
@@ -52,6 +52,9 @@ public class WsGremlinCloseRequestDecoder extends
MessageToMessageDecoder<CloseW
@Override
protected void decode(final ChannelHandlerContext channelHandlerContext,
final CloseWebSocketFrame frame, final List<Object> objects) throws Exception {
+ // recorded before anything reads from the frame as readableBytes() is
relative to the reader index
+
channelHandlerContext.channel().attr(StateKey.REQUEST_SIZE).set(frame.content().readableBytes());
+
final ByteBuf messageBytes = frame.content();
final byte len = messageBytes.readByte();
if (len <= 0) {
diff --git
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinTextRequestDecoder.java
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinTextRequestDecoder.java
index 79cf554dc3..3313177a5e 100644
---
a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinTextRequestDecoder.java
+++
b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinTextRequestDecoder.java
@@ -50,6 +50,9 @@ public class WsGremlinTextRequestDecoder extends
MessageToMessageDecoder<TextWeb
@Override
protected void decode(final ChannelHandlerContext channelHandlerContext,
final TextWebSocketFrame frame, final List<Object> objects) throws Exception {
+ // recorded before anything reads from the frame as readableBytes() is
relative to the reader index
+
channelHandlerContext.channel().attr(StateKey.REQUEST_SIZE).set(frame.content().readableBytes());
+
try {
// the default serializer must be a MessageTextSerializer instance
to be compatible with this decoder
final MessageTextSerializer<?> serializer =
(MessageTextSerializer<?>) select("application/json",
ServerSerializers.DEFAULT_TEXT_SERIALIZER);
diff --git
a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerAuthIntegrateTest.java
b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerAuthIntegrateTest.java
index 2df7863c8e..513ed4aec7 100644
---
a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerAuthIntegrateTest.java
+++
b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerAuthIntegrateTest.java
@@ -26,7 +26,6 @@ import
org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
import org.apache.tinkerpop.gremlin.driver.simple.WebSocketClient;
import org.apache.tinkerpop.gremlin.server.auth.SimpleAuthenticator;
-import org.apache.tinkerpop.gremlin.server.handler.SaslAuthenticationHandler;
import org.apache.tinkerpop.gremlin.structure.Property;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
@@ -58,6 +57,11 @@ import static org.junit.Assert.fail;
*/
public class GremlinServerAuthIntegrateTest extends
AbstractGremlinServerIntegrationTest {
+ /**
+ * Pre-authentication deadline for the one test that waits it out, kept
short to keep that test short.
+ */
+ private static final Duration SHORT_PRE_AUTH_TIMEOUT =
Duration.ofSeconds(2);
+
/**
* Configure specific Gremlin Server settings for specific tests.
*/
@@ -83,6 +87,9 @@ public class GremlinServerAuthIntegrateTest extends
AbstractGremlinServerIntegra
sslConfig.keyStorePassword = KEY_PASS;
settings.ssl = sslConfig;
break;
+ case
"shouldFailAuthenticateWithUnAuthenticatedRequestAfterMaxDeferrableDuration":
+ authSettings.preAuthTimeout =
SHORT_PRE_AUTH_TIMEOUT.toMillis();
+ break;
}
return settings;
@@ -177,38 +184,27 @@ public class GremlinServerAuthIntegrateTest extends
AbstractGremlinServerIntegra
@Test
public void
shouldFailAuthenticateWithUnAuthenticatedRequestAfterMaxDeferrableDuration()
throws Exception {
try (WebSocketClient client =
TestClientFactory.createWebSocketClient()) {
- // First request will initiate the authentication handshake
+ // First request will initiate the authentication handshake and be
held
// Subsequent requests will be deferred
- CompletableFuture<List<ResponseMessage>>
futureOfRequestWithinAuthDuration1 = client.submitAsync("");
- CompletableFuture<List<ResponseMessage>>
futureOfRequestWithinAuthDuration2 = client.submitAsync("");
- CompletableFuture<List<ResponseMessage>>
futureOfRequestWithinAuthDuration3 = client.submitAsync("");
-
- // After the maximum allowed deferred request duration,
- // any non-authenticated request will invalidate all requests with
429 error
- CompletableFuture<List<ResponseMessage>>
futureOfRequestSubmittedTooLate = CompletableFuture.runAsync(() -> {
- try {
-
Thread.sleep(SaslAuthenticationHandler.MAX_REQUEST_DEFERRABLE_DURATION.plus(Duration.ofSeconds(1)).toMillis());
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- }).thenCompose((__) -> {
- try {
- return client.submitAsync("");
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
-
- assertEquals(2, futureOfRequestWithinAuthDuration1.get().size());
- assertEquals(1, futureOfRequestWithinAuthDuration2.get().size());
- assertEquals(1, futureOfRequestWithinAuthDuration3.get().size());
- assertEquals(1, futureOfRequestSubmittedTooLate.get().size());
-
- assertEquals(ResponseStatusCode.AUTHENTICATE,
futureOfRequestWithinAuthDuration1.get().get(0).getStatus().getCode());
- assertEquals(ResponseStatusCode.UNAUTHORIZED,
futureOfRequestWithinAuthDuration1.get().get(1).getStatus().getCode());
- assertEquals(ResponseStatusCode.UNAUTHORIZED,
futureOfRequestWithinAuthDuration2.get().get(0).getStatus().getCode());
- assertEquals(ResponseStatusCode.UNAUTHORIZED,
futureOfRequestWithinAuthDuration3.get().get(0).getStatus().getCode());
- assertEquals(ResponseStatusCode.UNAUTHORIZED,
futureOfRequestSubmittedTooLate.get().get(0).getStatus().getCode());
+ final CompletableFuture<List<ResponseMessage>> firstRequest =
client.submitAsync("");
+ final CompletableFuture<List<ResponseMessage>> secondRequest =
client.submitAsync("");
+ final CompletableFuture<List<ResponseMessage>> thirdRequest =
client.submitAsync("");
+
+ // authentication is never completed, so the server answers
everything it held and closes the channel.
+ // waits are bounded because nothing more arrives on a closed
channel
+ final long timeoutMillis =
SHORT_PRE_AUTH_TIMEOUT.plus(Duration.ofSeconds(30)).toMillis();
+ final List<ResponseMessage> first =
firstRequest.get(timeoutMillis, TimeUnit.MILLISECONDS);
+ final List<ResponseMessage> second =
secondRequest.get(timeoutMillis, TimeUnit.MILLISECONDS);
+ final List<ResponseMessage> third =
thirdRequest.get(timeoutMillis, TimeUnit.MILLISECONDS);
+
+ assertEquals(2, first.size());
+ assertEquals(1, second.size());
+ assertEquals(1, third.size());
+
+ assertEquals(ResponseStatusCode.AUTHENTICATE,
first.get(0).getStatus().getCode());
+ assertEquals(ResponseStatusCode.UNAUTHORIZED,
first.get(1).getStatus().getCode());
+ assertEquals(ResponseStatusCode.UNAUTHORIZED,
second.get(0).getStatus().getCode());
+ assertEquals(ResponseStatusCode.UNAUTHORIZED,
third.get(0).getStatus().getCode());
}
}
diff --git
a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/SaslAuthenticationHandlerTest.java
b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/SaslAuthenticationHandlerTest.java
new file mode 100644
index 0000000000..ebe89b39f3
--- /dev/null
+++
b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/SaslAuthenticationHandlerTest.java
@@ -0,0 +1,826 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.server.handler;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoop;
+import io.netty.util.DefaultAttributeMap;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
+import io.netty.util.concurrent.ScheduledFuture;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.tinkerpop.gremlin.server.Settings;
+import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser;
+import org.apache.tinkerpop.gremlin.server.auth.AuthenticationException;
+import org.apache.tinkerpop.gremlin.server.auth.Authenticator;
+import org.apache.tinkerpop.gremlin.util.Tokens;
+import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
+import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
+import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class SaslAuthenticationHandlerTest {
+
+ // the size a request is given when the test does not care about the byte
budget
+ private static final int SMALL_REQUEST_SIZE = 200;
+
+ // a real AttributeMap backs the channel attributes the handler reads and
writes
+ private final DefaultAttributeMap attributes = new DefaultAttributeMap();
+ private final Channel channel = Mockito.mock(Channel.class);
+ private final ChannelHandlerContext ctx =
Mockito.mock(ChannelHandlerContext.class);
+ private final Authenticator authenticator =
Mockito.mock(Authenticator.class);
+ private final Authenticator.SaslNegotiator negotiator =
Mockito.mock(Authenticator.SaslNegotiator.class);
+ private final Settings settings = new Settings();
+ private final SaslAuthenticationHandler handler =
+ new SaslAuthenticationHandler(authenticator, null, settings);
+
+ // the limits the handler reads for this channel, left at their defaults
unless a test sets one
+ private final long budget =
settings.authentication.maxPreAuthRetainedBytes;
+ private final int maxDeferred =
settings.authentication.maxDeferredRequests;
+ private final long preAuthTimeout = settings.authentication.preAuthTimeout;
+
+ private final EventLoop eventLoop = Mockito.mock(EventLoop.class);
+ private final ChannelFuture closeFuture =
Mockito.mock(ChannelFuture.class);
+ private final ChannelFuture writeFuture =
Mockito.mock(ChannelFuture.class);
+ private final ScheduledFuture<?> deadlineFuture =
Mockito.mock(ScheduledFuture.class);
+
+ // what the handler handed to the event loop and to the close future, so
that the test can run them
+ private final List<Runnable> scheduledTasks = new ArrayList<>();
+ private final List<GenericFutureListener<Future<? super Void>>>
closeListeners = new ArrayList<>();
+
+ @Before
+ public void setupForEachTest() {
+ Mockito.when(channel.remoteAddress()).thenReturn(new
InetSocketAddress("127.0.0.1", 45678));
+ Mockito.doAnswer(invocation ->
attributes.attr(invocation.getArgument(0))).when(channel).attr(Mockito.any());
+ Mockito.when(channel.isActive()).thenReturn(true);
+ Mockito.when(channel.isWritable()).thenReturn(true);
+ Mockito.when(channel.eventLoop()).thenReturn(eventLoop);
+ Mockito.when(channel.closeFuture()).thenReturn(closeFuture);
+ Mockito.doAnswer(invocation -> {
+ scheduledTasks.add(invocation.getArgument(0));
+ return deadlineFuture;
+ }).when(eventLoop).schedule(Mockito.any(Runnable.class),
Mockito.anyLong(), Mockito.any(TimeUnit.class));
+ Mockito.doAnswer(invocation -> {
+ closeListeners.add(invocation.getArgument(0));
+ return closeFuture;
+ }).when(closeFuture).addListener(Mockito.any());
+ Mockito.when(ctx.channel()).thenReturn(channel);
+ Mockito.when(ctx.write(Mockito.any())).thenReturn(writeFuture);
+ Mockito.when(ctx.writeAndFlush(Mockito.any())).thenReturn(writeFuture);
+
Mockito.when(authenticator.newSaslNegotiator(Mockito.any(InetAddress.class))).thenReturn(negotiator);
+ }
+
+ @Test
+ public void shouldErrorOnlyTheRequestThatBreachedTheCountCap() throws
Exception {
+ final RequestMessage stashed = startAuthentication(SMALL_REQUEST_SIZE);
+
+ RequestMessage breaching = null;
+ for (int ix = 0; ix < maxDeferred + 1; ix++) {
+ breaching = deferRequest(SMALL_REQUEST_SIZE);
+ }
+
+ // the AUTHENTICATE challenge is followed by one error, for the
breaching request
+ final List<ResponseMessage> flushed = flushedResponses();
+ assertEquals(2, flushed.size());
+ assertEquals(ResponseStatusCode.AUTHENTICATE,
flushed.get(0).getStatus().getCode());
+
+ final ResponseMessage error = flushed.get(1);
+ assertEquals(breaching.getRequestId(), error.getRequestId());
+ assertEquals(ResponseStatusCode.UNAUTHORIZED,
error.getStatus().getCode());
+ assertTrue(error.getStatus().getMessage().contains(maxDeferred + "
maximum"));
+ Mockito.verify(ctx, Mockito.never()).write(Mockito.any());
+
+ // the list stays at the cap and the breaching request was never
appended
+ assertEquals(maxDeferred, deferredRequests().getValue().size());
+ assertFalse(deferredRequests().getValue().contains(breaching));
+ assertEquals((maxDeferred + 1) * (long) SMALL_REQUEST_SIZE,
retainedBytes());
+
+ // the request that started authentication is still stashed for
release once authentication completes
+ assertSame(stashed, attributes.attr(StateKey.REQUEST_MESSAGE).get());
+
+ // the channel is left open so authentication can still complete,
leaving the deadline as all that ends it
+ Mockito.verify(ctx, Mockito.never()).close();
+ assertDeadlineStillArmed();
+ }
+
+ @Test
+ public void shouldErrorOnlyTheRequestThatBreachedTheByteBudget() throws
Exception {
+ startAuthentication(SMALL_REQUEST_SIZE);
+
+ // two requests fill what is left of the budget and are far short of
the cap on the number of requests
+ final int half = (int) (budget - SMALL_REQUEST_SIZE) / 2;
+ deferRequest(half);
+ deferRequest(half);
+ assertEquals(budget, retainedBytes());
+
+ // the check includes the incoming request, otherwise one request
could pass the budget
+ final RequestMessage breaching = deferRequest(1);
+
+ final List<ResponseMessage> flushed = flushedResponses();
+ assertEquals(2, flushed.size());
+ final ResponseMessage error = flushed.get(1);
+ assertEquals(breaching.getRequestId(), error.getRequestId());
+ assertEquals(ResponseStatusCode.UNAUTHORIZED,
error.getStatus().getCode());
+ assertTrue(error.getStatus().getMessage().contains(budget + " bytes
maximum"));
+
+ // only the breaching request was rejected and the channel is left open
+ assertEquals(2, deferredRequests().getValue().size());
+ assertEquals(budget, retainedBytes());
+ Mockito.verify(ctx, Mockito.never()).write(Mockito.any());
+ Mockito.verify(ctx, Mockito.never()).close();
+
+ // refusing a request must not disarm the deadline, or breaching the
budget would hold the channel open
+ assertDeadlineStillArmed();
+ }
+
+ @Test
+ public void shouldChargeTheStashedRequestToTheByteBudget() throws
Exception {
+ final int stashSize = (int) budget - 100;
+ startAuthentication(stashSize);
+
+ // the request held in StateKey.REQUEST_MESSAGE counts from the moment
it is stashed
+ assertEquals(stashSize, retainedBytes());
+
+ // so only the remainder of the budget is left to defer into
+ final RequestMessage tooLarge = deferRequest(101);
+ assertNull(deferredRequests());
+
+ final List<ResponseMessage> flushed = flushedResponses();
+ assertEquals(2, flushed.size());
+ assertEquals(tooLarge.getRequestId(), flushed.get(1).getRequestId());
+ assertEquals(ResponseStatusCode.UNAUTHORIZED,
flushed.get(1).getStatus().getCode());
+
+ final RequestMessage fits = deferRequest(100);
+ assertEquals(1, deferredRequests().getValue().size());
+ assertSame(fits, deferredRequests().getValue().get(0));
+ assertEquals(budget, retainedBytes());
+ Mockito.verify(ctx, Mockito.never()).close();
+ }
+
+ @Test
+ public void
shouldChallengeButRetainNothingForAFirstRequestTooLargeToStash() throws
Exception {
+ final RequestMessage tooLarge =
startAuthenticationWithOversizedRequest();
+
+ // authentication still starts, so the challenge goes out first and
the request is answered after it
+ final List<ResponseMessage> flushed = flushedResponses();
+ assertEquals(2, flushed.size());
+ assertEquals(tooLarge.getRequestId(), flushed.get(0).getRequestId());
+ assertEquals(ResponseStatusCode.AUTHENTICATE,
flushed.get(0).getStatus().getCode());
+ assertEquals(tooLarge.getRequestId(), flushed.get(1).getRequestId());
+ assertEquals(ResponseStatusCode.UNAUTHORIZED,
flushed.get(1).getStatus().getCode());
+ assertTrue(flushed.get(1).getStatus().getMessage().contains("too
large"));
+
+ // it was answered rather than stashed, and nothing was retained for
the channel either
+ assertNull(attributes.attr(StateKey.REQUEST_MESSAGE).get());
+ assertNull(deferredRequests());
+ assertEquals(0L, retainedBytes());
+ Mockito.verify(ctx, Mockito.never()).write(Mockito.any());
+ Mockito.verify(ctx, Mockito.never()).close();
+ }
+
+ @Test
+ public void
shouldDeferANormalSizedRequestAfterAFirstRequestTooLargeToStashWasRejected()
throws Exception {
+ startAuthenticationWithOversizedRequest();
+
+ final RequestMessage deferred = deferRequest(SMALL_REQUEST_SIZE);
+
+ // only the deferred request is charged, as the rejected one was never
counted
+ assertEquals(1, deferredRequests().getValue().size());
+ assertSame(deferred, deferredRequests().getValue().get(0));
+ assertEquals(SMALL_REQUEST_SIZE, retainedBytes());
+
+ completeAuthentication();
+
+ final List<Object> fired = firedReads();
+ assertEquals(1, fired.size());
+ assertSame(deferred, fired.get(0));
+ Mockito.verify(ctx, Mockito.never()).write(Mockito.any());
+ assertNothingLeftToAnswer();
+ }
+
+ @Test
+ public void shouldAnswerTheStashedRequestExactlyOnceAcrossRepeatedErrors()
throws Exception {
+ final RequestMessage stashed = startAuthentication(SMALL_REQUEST_SIZE);
+ final RequestMessage deferred = deferRequest(SMALL_REQUEST_SIZE);
+
+ failAuthentication();
+
+ final List<ResponseMessage> errors = writtenResponses();
+ assertEquals(2, errors.size());
+ assertEquals(stashed.getRequestId(), errors.get(0).getRequestId());
+ assertEquals(deferred.getRequestId(), errors.get(1).getRequestId());
+ errors.forEach(error -> assertEquals(ResponseStatusCode.UNAUTHORIZED,
error.getStatus().getCode()));
+ assertNull("the stashed request must be cleared as it is answered",
+ attributes.attr(StateKey.REQUEST_MESSAGE).get());
+ assertNothingLeftToAnswer();
+
+ // an authentication request with no sasl argument takes
respondWithError again, with nothing left to answer
+ handler.channelRead(ctx,
RequestMessage.build(Tokens.OPS_AUTHENTICATION).create());
+
+ assertEquals(2, writtenResponses().size());
+ assertEquals(1L, answersFor(stashed));
+ Mockito.verify(ctx, Mockito.never()).close();
+ }
+
+ @Test
+ public void
shouldNotAnswerTheStashedRequestAgainWhenAuthenticationLaterSucceeds() throws
Exception {
+ final RequestMessage stashed = startAuthentication(SMALL_REQUEST_SIZE);
+ deferRequest(SMALL_REQUEST_SIZE);
+
+ failAuthentication();
+ assertNull(attributes.attr(StateKey.REQUEST_MESSAGE).get());
+
+ completeAuthentication();
+
+ // the failure already answered it, so nothing more goes out for it
and nothing null goes down the pipeline
+ assertEquals(1L, answersFor(stashed));
+ assertTrue(firedReads().isEmpty());
+ Mockito.verify(ctx, Mockito.never()).fireChannelRead(null);
+ assertNothingLeftToAnswer();
+ }
+
+ @Test
+ public void shouldRetainNothingOnceAuthenticationSucceeds() throws
Exception {
+ final RequestMessage stashed = startAuthentication(SMALL_REQUEST_SIZE);
+ final RequestMessage deferred = deferRequest(SMALL_REQUEST_SIZE);
+
+ completeAuthentication();
+
+ final List<Object> fired = firedReads();
+ assertEquals(2, fired.size());
+ assertSame(stashed, fired.get(0));
+ assertSame(deferred, fired.get(1));
+ Mockito.verify(ctx, Mockito.never()).write(Mockito.any());
+ assertNull(deferredRequests());
+ assertEquals(0L, retainedBytes());
+ }
+
+ @Test
+ public void shouldRetainNothingWhenTheSaslArgumentIsMissing() throws
Exception {
+ final RequestMessage stashed = startAuthentication(SMALL_REQUEST_SIZE);
+ final RequestMessage deferred = deferRequest(SMALL_REQUEST_SIZE);
+
+ handler.channelRead(ctx,
RequestMessage.build(Tokens.OPS_AUTHENTICATION).create());
+
+ // the authentication request itself is not answered, the stashed and
deferred ones are
+ final List<ResponseMessage> errors = writtenResponses();
+ assertEquals(2, errors.size());
+ assertEquals(stashed.getRequestId(), errors.get(0).getRequestId());
+ assertEquals(deferred.getRequestId(), errors.get(1).getRequestId());
+ errors.forEach(error -> assertEquals(ResponseStatusCode.UNAUTHORIZED,
error.getStatus().getCode()));
+ assertNothingLeftToAnswer();
+ }
+
+ @Test
+ public void shouldRetainNothingWhenTheSaslArgumentIsNotAString() throws
Exception {
+ final RequestMessage stashed = startAuthentication(SMALL_REQUEST_SIZE);
+ final RequestMessage deferred = deferRequest(SMALL_REQUEST_SIZE);
+
+ handler.channelRead(ctx,
RequestMessage.build(Tokens.OPS_AUTHENTICATION)
+ .addArg(Tokens.ARGS_SASL, 1234).create());
+
+ final List<ResponseMessage> errors = writtenResponses();
+ assertEquals(2, errors.size());
+ assertEquals(stashed.getRequestId(), errors.get(0).getRequestId());
+ assertEquals(deferred.getRequestId(), errors.get(1).getRequestId());
+ errors.forEach(error ->
+
assertEquals(ResponseStatusCode.REQUEST_ERROR_MALFORMED_REQUEST,
error.getStatus().getCode()));
+ assertNothingLeftToAnswer();
+ }
+
+ @Test
+ public void
shouldAnswerRatherThanDeferARequestOnceTheDeferrableDurationHasPassed() throws
Exception {
+ final RequestMessage stashed = startAuthentication(SMALL_REQUEST_SIZE);
+ final RequestMessage deferred = deferRequest(SMALL_REQUEST_SIZE);
+
+ // the deferral window opened longer ago than the allowed duration,
which the next arrival notices
+ attributes.attr(StateKey.DEFERRED_REQUEST_MESSAGES).set(new
ImmutablePair<>(
+
LocalDateTime.now().minus(Duration.ofMillis(preAuthTimeout)).minusSeconds(1),
+ deferredRequests().getValue()));
+
+ final RequestMessage late = deferRequest(SMALL_REQUEST_SIZE);
+
+ // the late request is answered as an arrival, never appended, so no
request id is answered twice
+ final List<ResponseMessage> errors = writtenResponses();
+ assertEquals(3, errors.size());
+ assertEquals(late.getRequestId(), errors.get(0).getRequestId());
+ assertEquals(stashed.getRequestId(), errors.get(1).getRequestId());
+ assertEquals(deferred.getRequestId(), errors.get(2).getRequestId());
+ errors.forEach(error -> assertEquals(ResponseStatusCode.UNAUTHORIZED,
error.getStatus().getCode()));
+ assertEquals(1L, answersFor(late));
+ assertEquals(1L, answersFor(stashed));
+ assertEquals(1L, answersFor(deferred));
+ assertNothingLeftToAnswer();
+
+ // the stale window answers rather than closes, so the deadline is
still what ends the channel
+ Mockito.verify(ctx, Mockito.never()).close();
+ assertDeadlineStillArmed();
+ }
+
+ @Test
+ public void shouldArmNothingWhenTheAuthenticatorIsNotReady() throws
Exception {
+
Mockito.when(authenticator.newSaslNegotiator(Mockito.any(InetAddress.class)))
+ .thenThrow(new IllegalStateException("not ready"));
+
+ final RequestMessage first =
RequestMessage.build(Tokens.OPS_EVAL).create();
+ attributes.attr(StateKey.REQUEST_SIZE).set(SMALL_REQUEST_SIZE);
+ handler.channelRead(ctx, first);
+
+ final List<ResponseMessage> errors = writtenResponses();
+ assertEquals(1, errors.size());
+ assertEquals(first.getRequestId(), errors.get(0).getRequestId());
+ assertEquals(ResponseStatusCode.SERVER_ERROR,
errors.get(0).getStatus().getCode());
+
+ // no negotiator means nothing is retained, so no deadline is left
running either
+ assertNull(attributes.attr(StateKey.NEGOTIATOR).get());
+ assertNull(deadline());
+ assertTrue(scheduledTasks.isEmpty());
+ assertNothingLeftToAnswer();
+ Mockito.verify(ctx, Mockito.never()).close();
+ }
+
+ @Test
+ public void shouldCountNothingForARequestTheTransportDidNotSize() throws
Exception {
+ // StateKey.REQUEST_SIZE is absent on transports that do not record it
+ assertNull(attributes.attr(StateKey.REQUEST_SIZE).get());
+
+ final RequestMessage first =
RequestMessage.build(Tokens.OPS_AUTHENTICATION).create();
+ handler.channelRead(ctx, first);
+
+ assertSame(first, attributes.attr(StateKey.REQUEST_MESSAGE).get());
+ assertEquals(0L, retainedBytes());
+
+ handler.channelRead(ctx,
RequestMessage.build(Tokens.OPS_EVAL).create());
+
+ // the request is still deferred, and still bounded by the cap on the
number of them
+ assertEquals(1, deferredRequests().getValue().size());
+ assertEquals(0L, retainedBytes());
+ assertEquals(1, flushedResponses().size());
+ }
+
+ @Test
+ public void shouldArmTheDeadlineWhenTheNegotiatorIsCreated() throws
Exception {
+ startAuthentication(SMALL_REQUEST_SIZE);
+
+ // retention begins with the stashed request, before anything is
deferred, so the deadline is already armed
+ assertSame(deadlineFuture, deadline());
+ assertNull(deferredRequests());
+ assertEquals(1, scheduledTasks.size());
+ Mockito.verify(eventLoop).schedule(Mockito.any(Runnable.class),
+ Mockito.eq(preAuthTimeout), Mockito.eq(TimeUnit.MILLISECONDS));
+ }
+
+ @Test
+ public void
shouldNotArmTheDeadlineAgainWhenTheDeferredRequestsAreRefilled() throws
Exception {
+ startAuthentication(SMALL_REQUEST_SIZE);
+ deferRequest(SMALL_REQUEST_SIZE);
+
+ // releases the deferred requests, so the next arrival opens a fresh
window on the same channel
+ failAuthentication();
+ assertNull(deferredRequests());
+
+ deferRequest(SMALL_REQUEST_SIZE);
+
+ assertEquals(1, deferredRequests().getValue().size());
+ assertEquals(1, scheduledTasks.size());
+ assertEquals(1, closeListeners.size());
+ assertDeadlineStillArmed();
+ }
+
+ @Test
+ public void shouldArmTheDeadlineOnceWhenTheNegotiatorKeepsComingBackNull()
throws Exception {
+
Mockito.when(authenticator.newSaslNegotiator(Mockito.any(InetAddress.class))).thenReturn(null);
+
+ for (int ix = 0; ix < 3; ix++) {
+ attributes.attr(StateKey.REQUEST_SIZE).set(SMALL_REQUEST_SIZE);
+ handler.channelRead(ctx,
RequestMessage.build(Tokens.OPS_AUTHENTICATION).create());
+ }
+
+ // the branch that creates the negotiator runs again for each request,
and must arm nothing more
+ assertNull(attributes.attr(StateKey.NEGOTIATOR).get());
+ assertEquals(3, flushedResponses().size());
+ assertEquals(1, scheduledTasks.size());
+ assertEquals(1, closeListeners.size());
+ assertDeadlineStillArmed();
+ }
+
+ /**
+ * The paths that refuse a request rather than fail an authentication one
keep the deadline armed in
+ * {@link #shouldErrorOnlyTheRequestThatBreachedTheCountCap},
+ * {@link #shouldErrorOnlyTheRequestThatBreachedTheByteBudget} and
+ * {@link
#shouldAnswerRatherThanDeferARequestOnceTheDeferrableDurationHasPassed}.
+ */
+ @Test
+ public void
shouldKeepTheDeadlineArmedThroughEveryFailedAuthenticationRequest() throws
Exception {
+ startAuthentication(SMALL_REQUEST_SIZE);
+ deferRequest(SMALL_REQUEST_SIZE);
+
+ // an authentication request with no sasl argument
+ handler.channelRead(ctx,
RequestMessage.build(Tokens.OPS_AUTHENTICATION).create());
+ assertEquals(2, writtenResponses().size());
+ assertDeadlineStillArmed();
+
+ // an authentication request whose sasl argument is not a String
+ deferRequest(SMALL_REQUEST_SIZE);
+ handler.channelRead(ctx,
RequestMessage.build(Tokens.OPS_AUTHENTICATION)
+ .addArg(Tokens.ARGS_SASL, 1234).create());
+ assertEquals(3, writtenResponses().size());
+ assertDeadlineStillArmed();
+
+ // an authentication request the negotiator rejects
+ deferRequest(SMALL_REQUEST_SIZE);
+ failAuthentication();
+ assertEquals(4, writtenResponses().size());
+ assertDeadlineStillArmed();
+
+ // none of them closed the channel either, so the deadline is all that
ends it
+ Mockito.verify(ctx, Mockito.never()).close();
+ }
+
+ @Test
+ public void shouldCancelTheDeadlineWhenAuthenticationSucceeds() throws
Exception {
+ startAuthentication(SMALL_REQUEST_SIZE);
+
+ completeAuthentication();
+
+ assertNull(deadline());
+ Mockito.verify(deadlineFuture).cancel(false);
+ }
+
+ @Test
+ public void shouldCancelTheDeadlineWhenTheChannelCloses() throws Exception
{
+ startAuthentication(SMALL_REQUEST_SIZE);
+ deferRequest(SMALL_REQUEST_SIZE);
+
+ closeChannel();
+
+ assertNull(deadline());
+ Mockito.verify(deadlineFuture).cancel(false);
+ }
+
+ @Test
+ public void
shouldAnswerEachPendingRequestExactlyOnceAndCloseWhenTheDeadlineExpires()
throws Exception {
+ final RequestMessage stashed = startAuthentication(SMALL_REQUEST_SIZE);
+ final RequestMessage first = deferRequest(SMALL_REQUEST_SIZE);
+ final RequestMessage second = deferRequest(SMALL_REQUEST_SIZE);
+
+ runDeadline();
+
+ // closed as soon as the responses are flushed, never gated on a write
completing
+ Mockito.verify(ctx).flush();
+ Mockito.verify(ctx).close();
+ Mockito.verify(writeFuture,
Mockito.never()).addListener(Mockito.any());
+
+ final List<ResponseMessage> errors = writtenResponses();
+ assertEquals(3, errors.size());
+ assertEquals(stashed.getRequestId(), errors.get(0).getRequestId());
+ assertEquals(first.getRequestId(), errors.get(1).getRequestId());
+ assertEquals(second.getRequestId(), errors.get(2).getRequestId());
+ errors.forEach(error -> assertEquals(ResponseStatusCode.UNAUTHORIZED,
error.getStatus().getCode()));
+ assertEquals(1L, answersFor(stashed));
+ assertEquals(1L, answersFor(first));
+ assertEquals(1L, answersFor(second));
+ assertNothingLeftToAnswer();
+ assertTrue(firedReads().isEmpty());
+ }
+
+ @Test
+ public void
shouldNotAnswerARequestAnErrorAlreadyAnsweredWhenTheDeadlineExpires() throws
Exception {
+ final RequestMessage stashed = startAuthentication(SMALL_REQUEST_SIZE);
+ final RequestMessage deferred = deferRequest(SMALL_REQUEST_SIZE);
+
+ failAuthentication();
+ assertEquals(2, writtenResponses().size());
+
+ runDeadline();
+
+ // both were answered and cleared by the failure, so the deadline has
nothing left to answer
+ assertEquals(2, writtenResponses().size());
+ assertEquals(1L, answersFor(stashed));
+ assertEquals(1L, answersFor(deferred));
+
+ // the channel is still unauthenticated, so it is still closed
+ Mockito.verify(ctx).close();
+ }
+
+ @Test
+ public void shouldNotCloseAnAuthenticatedChannelWhenTheDeadlineExpires()
throws Exception {
+ startAuthentication(SMALL_REQUEST_SIZE);
+ completeAuthentication();
+
+ // a task already running when the cancel came in still has to find
the channel authenticated and stop
+ runDeadline();
+
+ Mockito.verify(ctx, Mockito.never()).close();
+ Mockito.verify(ctx, Mockito.never()).write(Mockito.any());
+ }
+
+ @Test
+ public void shouldNotCloseAnInactiveChannelWhenTheDeadlineExpires() throws
Exception {
+ startAuthentication(SMALL_REQUEST_SIZE);
+ deferRequest(SMALL_REQUEST_SIZE);
+ Mockito.when(channel.isActive()).thenReturn(false);
+
+ runDeadline();
+
+ Mockito.verify(ctx, Mockito.never()).close();
+ Mockito.verify(ctx, Mockito.never()).write(Mockito.any());
+ }
+
+ @Test
+ public void shouldSkipTheRejectionWriteWhenTheChannelIsNotWritable()
throws Exception {
+ startAuthentication(SMALL_REQUEST_SIZE);
+
+ for (int ix = 0; ix < maxDeferred; ix++) {
+ deferRequest(SMALL_REQUEST_SIZE);
+ }
+
+ Mockito.when(channel.isWritable()).thenReturn(false);
+ final RequestMessage breaching = deferRequest(SMALL_REQUEST_SIZE);
+
+ // the rejection is dropped rather than buffered, so the AUTHENTICATE
challenge is all that went out
+ assertEquals(1, flushedResponses().size());
+ assertEquals(ResponseStatusCode.AUTHENTICATE,
flushedResponses().get(0).getStatus().getCode());
+ Mockito.verify(ctx, Mockito.never()).write(Mockito.any());
+
+ // the request is still refused rather than deferred
+ assertEquals(maxDeferred, deferredRequests().getValue().size());
+ assertFalse(deferredRequests().getValue().contains(breaching));
+ assertEquals((maxDeferred + 1) * (long) SMALL_REQUEST_SIZE,
retainedBytes());
+ Mockito.verify(ctx, Mockito.never()).close();
+ }
+
+ @Test
+ public void shouldCapTheDeferredRequestsAtTheConfiguredCount() throws
Exception {
+ settings.authentication.maxDeferredRequests = 2;
+
+ startAuthentication(SMALL_REQUEST_SIZE);
+ deferRequest(SMALL_REQUEST_SIZE);
+ deferRequest(SMALL_REQUEST_SIZE);
+
+ // the default of 64 would have deferred this one
+ final RequestMessage breaching = deferRequest(SMALL_REQUEST_SIZE);
+
+ final List<ResponseMessage> flushed = flushedResponses();
+ assertEquals(2, flushed.size());
+ assertEquals(breaching.getRequestId(), flushed.get(1).getRequestId());
+ assertEquals(ResponseStatusCode.UNAUTHORIZED,
flushed.get(1).getStatus().getCode());
+ assertTrue(flushed.get(1).getStatus().getMessage().contains("2
maximum"));
+ assertEquals(2, deferredRequests().getValue().size());
+ assertFalse(deferredRequests().getValue().contains(breaching));
+ }
+
+ @Test
+ public void shouldBudgetTheRetainedBytesAtTheConfiguredTotal() throws
Exception {
+ settings.authentication.maxPreAuthRetainedBytes = 500L;
+
+ startAuthentication(400);
+
+ // the default budget of 2 MiB would have deferred this one
+ final RequestMessage breaching = deferRequest(101);
+
+ final List<ResponseMessage> flushed = flushedResponses();
+ assertEquals(2, flushed.size());
+ assertEquals(breaching.getRequestId(), flushed.get(1).getRequestId());
+ assertEquals(ResponseStatusCode.UNAUTHORIZED,
flushed.get(1).getStatus().getCode());
+ assertTrue(flushed.get(1).getStatus().getMessage().contains("500 bytes
maximum"));
+ assertNull(deferredRequests());
+
+ // what is left of the smaller budget is still deferrable
+ final RequestMessage fits = deferRequest(100);
+ assertSame(fits, deferredRequests().getValue().get(0));
+ assertEquals(500L, retainedBytes());
+ }
+
+ @Test
+ public void shouldArmTheDeadlineWithTheConfiguredTimeout() throws
Exception {
+ settings.authentication.preAuthTimeout = 1234L;
+
+ startAuthentication(SMALL_REQUEST_SIZE);
+
+ Mockito.verify(eventLoop).schedule(Mockito.any(Runnable.class),
+ Mockito.eq(1234L), Mockito.eq(TimeUnit.MILLISECONDS));
+
+ runDeadline();
+
+ assertEquals(1, writtenResponses().size());
+
assertTrue(writtenResponses().get(0).getStatus().getMessage().contains("1234
ms"));
+ }
+
+ @Test
+ public void shouldRefuseAMaxDeferredRequestsThatIsNotPositive() {
+ assertRejectedAtConstruction(s -> s.authentication.maxDeferredRequests
= 0,
+ "authentication.maxDeferredRequests");
+ assertRejectedAtConstruction(s -> s.authentication.maxDeferredRequests
= -1,
+ "authentication.maxDeferredRequests");
+ }
+
+ @Test
+ public void shouldRefuseAMaxPreAuthRetainedBytesThatIsNotPositive() {
+ assertRejectedAtConstruction(s ->
s.authentication.maxPreAuthRetainedBytes = 0L,
+ "authentication.maxPreAuthRetainedBytes");
+ assertRejectedAtConstruction(s ->
s.authentication.maxPreAuthRetainedBytes = -1L,
+ "authentication.maxPreAuthRetainedBytes");
+ }
+
+ @Test
+ public void shouldRefuseAPreAuthTimeoutThatIsNotPositive() {
+ assertRejectedAtConstruction(s -> s.authentication.preAuthTimeout = 0L,
+ "authentication.preAuthTimeout");
+ assertRejectedAtConstruction(s -> s.authentication.preAuthTimeout =
-1L,
+ "authentication.preAuthTimeout");
+ }
+
+ /**
+ * Asserts that the handler cannot be constructed once {@code
invalidSetting} has been applied, so that the server
+ * fails to start rather than a request.
+ */
+ private void assertRejectedAtConstruction(final Consumer<Settings>
invalidSetting, final String settingName) {
+ final Settings invalid = new Settings();
+ invalidSetting.accept(invalid);
+
+ try {
+ new SaslAuthenticationHandler(authenticator, null, invalid);
+ fail("a value that is not positive for " + settingName + " must
not be accepted");
+ } catch (IllegalStateException ex) {
+ assertTrue(ex.getMessage(), ex.getMessage().contains(settingName +
" must be greater than zero"));
+ }
+ }
+
+ /**
+ * Sends a first request so that later non-authentication requests take
the deferral path, returning the request
+ * the handler stashes in {@link StateKey#REQUEST_MESSAGE}.
+ */
+ private RequestMessage startAuthentication(final int requestSize) throws
Exception {
+ final RequestMessage first =
RequestMessage.build(Tokens.OPS_AUTHENTICATION).create();
+ attributes.attr(StateKey.REQUEST_SIZE).set(requestSize);
+ handler.channelRead(ctx, first);
+
+ assertNotNull(attributes.attr(StateKey.NEGOTIATOR).get());
+
+ return first;
+ }
+
+ /**
+ * Sends a first request too large to be retained, which still starts
authentication.
+ */
+ private RequestMessage startAuthenticationWithOversizedRequest() throws
Exception {
+ final RequestMessage first =
RequestMessage.build(Tokens.OPS_EVAL).create();
+ attributes.attr(StateKey.REQUEST_SIZE).set((int) budget + 1);
+ handler.channelRead(ctx, first);
+
+ assertNotNull(attributes.attr(StateKey.NEGOTIATOR).get());
+
+ return first;
+ }
+
+ /**
+ * Sends a request that takes the deferral path, sized as the decoders
size it.
+ */
+ private RequestMessage deferRequest(final int requestSize) throws
Exception {
+ final RequestMessage request =
RequestMessage.build(Tokens.OPS_EVAL).create();
+ attributes.attr(StateKey.REQUEST_SIZE).set(requestSize);
+ handler.channelRead(ctx, request);
+
+ return request;
+ }
+
+ /**
+ * Sends a SASL response that the negotiator accepts, releasing the
deferred requests down the pipeline.
+ */
+ private void completeAuthentication() throws Exception {
+ // doReturn() rather than when() so that it can follow
failAuthentication() in the one test
+
Mockito.when(ctx.pipeline()).thenReturn(Mockito.mock(ChannelPipeline.class));
+ Mockito.doReturn(new
byte[0]).when(negotiator).evaluateResponse(Mockito.any());
+ Mockito.doReturn(true).when(negotiator).isComplete();
+ Mockito.doReturn(new
AuthenticatedUser("stephen")).when(negotiator).getAuthenticatedUser();
+
+ sendSaslResponse();
+ }
+
+ /**
+ * Sends a SASL response that the negotiator rejects.
+ */
+ private void failAuthentication() throws Exception {
+ Mockito.doThrow(new AuthenticationException("Username and/or password
are incorrect"))
+ .when(negotiator).evaluateResponse(Mockito.any());
+
+ sendSaslResponse();
+ }
+
+ private void sendSaslResponse() throws Exception {
+ attributes.attr(StateKey.REQUEST_SIZE).set(SMALL_REQUEST_SIZE);
+ handler.channelRead(ctx,
RequestMessage.build(Tokens.OPS_AUTHENTICATION)
+ .addArg(Tokens.ARGS_SASL,
Base64.getEncoder().encodeToString(new byte[0])).create());
+ }
+
+ private void assertNothingLeftToAnswer() {
+ assertNull(attributes.attr(StateKey.REQUEST_MESSAGE).get());
+ assertNull(deferredRequests());
+ assertEquals(0L, retainedBytes());
+ }
+
+ private void assertDeadlineStillArmed() {
+ assertSame("the deadline must still be armed", deadlineFuture,
deadline());
+ Mockito.verify(deadlineFuture,
Mockito.never()).cancel(Mockito.anyBoolean());
+ }
+
+ /**
+ * Runs the task the handler scheduled, failing rather than doing nothing
when there is not exactly one of them.
+ */
+ private void runDeadline() {
+ assertEquals("exactly one deadline task must have been scheduled", 1,
scheduledTasks.size());
+ scheduledTasks.get(0).run();
+ }
+
+ /**
+ * Fires the listener the handler registered on the close future, failing
when there is not exactly one of them.
+ */
+ private void closeChannel() throws Exception {
+ assertEquals("exactly one close listener must have been registered",
1, closeListeners.size());
+ closeListeners.get(0).operationComplete(closeFuture);
+ }
+
+ private ScheduledFuture<?> deadline() {
+ return attributes.attr(StateKey.PREAUTH_DEADLINE).get();
+ }
+
+ /**
+ * Number of terminal responses written for {@code request}. The
AUTHENTICATE challenge is not one of them and
+ * leaves through {@code writeAndFlush()} rather than {@code write()}.
+ */
+ private long answersFor(final RequestMessage request) {
+ return writtenResponses().stream()
+ .filter(response ->
request.getRequestId().equals(response.getRequestId()))
+ .count();
+ }
+
+ private Pair<LocalDateTime, List<RequestMessage>> deferredRequests() {
+ return attributes.attr(StateKey.DEFERRED_REQUEST_MESSAGES).get();
+ }
+
+ private long retainedBytes() {
+ final Long bytes =
attributes.attr(StateKey.DEFERRED_REQUEST_BYTES).get();
+ return bytes == null ? 0L : bytes;
+ }
+
+ private List<ResponseMessage> flushedResponses() {
+ final ArgumentCaptor<Object> captor =
ArgumentCaptor.forClass(Object.class);
+ Mockito.verify(ctx,
Mockito.atLeast(0)).writeAndFlush(captor.capture());
+ return
captor.getAllValues().stream().map(ResponseMessage.class::cast).collect(Collectors.toList());
+ }
+
+ private List<ResponseMessage> writtenResponses() {
+ final ArgumentCaptor<Object> captor =
ArgumentCaptor.forClass(Object.class);
+ Mockito.verify(ctx, Mockito.atLeast(0)).write(captor.capture());
+ return
captor.getAllValues().stream().map(ResponseMessage.class::cast).collect(Collectors.toList());
+ }
+
+ private List<Object> firedReads() {
+ final ArgumentCaptor<Object> captor =
ArgumentCaptor.forClass(Object.class);
+ Mockito.verify(ctx,
Mockito.atLeast(0)).fireChannelRead(captor.capture());
+ return captor.getAllValues();
+ }
+}
diff --git
a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinRequestDecoderTest.java
b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinRequestDecoderTest.java
new file mode 100644
index 0000000000..bf62c1d697
--- /dev/null
+++
b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/WsGremlinRequestDecoderTest.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.server.handler;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import org.apache.tinkerpop.gremlin.util.MessageSerializer;
+import org.apache.tinkerpop.gremlin.util.Tokens;
+import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
+import org.apache.tinkerpop.gremlin.util.ser.GraphBinaryMessageSerializerV1;
+import org.apache.tinkerpop.gremlin.util.ser.GraphSONMessageSerializerV3;
+import org.junit.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Covers the frame size the WebSocket decoders record for {@link
StateKey#REQUEST_SIZE}.
+ */
+public class WsGremlinRequestDecoderTest {
+
+ private final GraphBinaryMessageSerializerV1 binarySerializer = new
GraphBinaryMessageSerializerV1();
+ private final GraphSONMessageSerializerV3 textSerializer = new
GraphSONMessageSerializerV3();
+
+ @Test
+ public void shouldRecordTheFullFrameSizeOfABinaryFrame() throws Exception {
+ final RequestMessage request = RequestMessage.build(Tokens.OPS_EVAL)
+ .addArg(Tokens.ARGS_GREMLIN, "g.V()").create();
+ final ByteBuf content =
binarySerializer.serializeRequestAsBinary(request,
UnpooledByteBufAllocator.DEFAULT);
+ final int frameSize = content.readableBytes();
+
+ final EmbeddedChannel channel = new EmbeddedChannel(new
WsGremlinBinaryRequestDecoder(serializers(binarySerializer)));
+ try {
+ channel.writeInbound(new BinaryWebSocketFrame(content));
+
+ assertEquals(request.getRequestId(), ((RequestMessage)
channel.readInbound()).getRequestId());
+ assertEquals(frameSize, requestSize(channel));
+
+ // not the remainder that is left once the mime type length byte
has been read
+ assertTrue(frameSize > 1);
+ assertNotEquals(frameSize - 1, requestSize(channel));
+ } finally {
+ channel.finishAndReleaseAll();
+ }
+ }
+
+ @Test
+ public void shouldRecordTheFullFrameSizeOfACloseFrame() throws Exception {
+ final RequestMessage request =
RequestMessage.build(Tokens.OPS_CLOSE).create();
+ final ByteBuf content =
binarySerializer.serializeRequestAsBinary(request,
UnpooledByteBufAllocator.DEFAULT);
+ final int frameSize = content.readableBytes();
+
+ final EmbeddedChannel channel = new EmbeddedChannel(new
WsGremlinCloseRequestDecoder(serializers(binarySerializer)));
+ try {
+ channel.writeInbound(new CloseWebSocketFrame(true, 0, content));
+
+ assertEquals(request.getRequestId(), ((RequestMessage)
channel.readInbound()).getRequestId());
+ assertEquals(frameSize, requestSize(channel));
+
+ // not the remainder that is left once the mime type length byte
has been read
+ assertTrue(frameSize > 1);
+ assertNotEquals(frameSize - 1, requestSize(channel));
+ } finally {
+ channel.finishAndReleaseAll();
+ }
+ }
+
+ @Test
+ public void shouldRecordTheFullFrameSizeOfATextFrame() throws Exception {
+ // the non-ascii argument makes the frame longer than the character
count of its text
+ final RequestMessage request = RequestMessage.build(Tokens.OPS_EVAL)
+ .addArg(Tokens.ARGS_GREMLIN, "g.V().has('name','é')").create();
+ final String text = textSerializer.serializeRequestAsString(request,
UnpooledByteBufAllocator.DEFAULT);
+ final int frameSize = text.getBytes(StandardCharsets.UTF_8).length;
+
+ final EmbeddedChannel channel = new EmbeddedChannel(new
WsGremlinTextRequestDecoder(serializers(textSerializer)));
+ try {
+ channel.writeInbound(new TextWebSocketFrame(text));
+
+ assertEquals(request.getRequestId(), ((RequestMessage)
channel.readInbound()).getRequestId());
+ assertEquals(frameSize, requestSize(channel));
+ assertNotEquals(text.length(), requestSize(channel));
+ } finally {
+ channel.finishAndReleaseAll();
+ }
+ }
+
+ private static int requestSize(final EmbeddedChannel channel) {
+ return channel.attr(StateKey.REQUEST_SIZE).get();
+ }
+
+ private static Map<String, MessageSerializer<?>> serializers(final
MessageSerializer<?> serializer) {
+ final Map<String, MessageSerializer<?>> serializers = new HashMap<>();
+
+ for (final String mimeType : serializer.mimeTypesSupported()) {
+ serializers.put(mimeType, serializer);
+ }
+
+ return serializers;
+ }
+}
