This is an automated email from the ASF dual-hosted git repository. kenhuuu pushed a commit to branch 3.7-dev in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit c90ac5df947d3c86ca05e3de85f67d32ce4b9023 Author: Ken Hu <[email protected]> AuthorDate: Mon Aug 31 20:10:35 2026 -0700 Ensure buffer release in HTTP handlers CTR Assisted-by: Codex:gpt-5.6-sol --- CHANGELOG.asciidoc | 1 + .../handler/HttpBasicAuthorizationHandler.java | 63 ++-- .../server/handler/HttpGremlinEndpointHandler.java | 346 +++++++++++---------- .../gremlin/server/handler/HttpHandlerUtil.java | 19 +- .../handler/HttpBasicAuthorizationHandlerTest.java | 67 ++++ .../handler/HttpGremlinEndpointHandlerTest.java | 116 +++++++ .../server/handler/HttpHandlerUtilTest.java | 44 +++ 7 files changed, 448 insertions(+), 208 deletions(-) diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc index 49f4187c12..f45d47a595 100644 --- a/CHANGELOG.asciidoc +++ b/CHANGELOG.asciidoc @@ -50,6 +50,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima * Removed the Mono dependency from the `gremlin-dotnet` build/release process, using `dotnet pack`/`dotnet nuget push` instead of `mono nuget.exe`. * Expanded `gremlin-python` CI matrix to test against Python 3.9, 3.10, 3.11, 3.12, and 3.13. * Add Node 26 support for `gremlin-javascript` and `gremlint`. +* Fixed Gremlin Server HTTP requests to release their buffers when request processing fails unexpectedly. * Fixed a panic in `gremlin-go` `PartitionStrategy` when `ReadPartitions` was left unset. * Fixed `gremlin-python` `ProductiveByStrategy` to pass through the `productiveKeys` argument, which was previously accepted but never serialized to the server. * Deprecated `ProductiveByStrategy` which was introduced as a temporary way to mimic pre-3.5.0 null processing behavior. diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpBasicAuthorizationHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpBasicAuthorizationHandler.java index 736b11dc4c..048a1c11b7 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpBasicAuthorizationHandler.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpBasicAuthorizationHandler.java @@ -59,46 +59,49 @@ public class HttpBasicAuthorizationHandler extends ChannelInboundHandlerAdapter public void channelRead(final ChannelHandlerContext ctx, final Object msg) { if (msg instanceof FullHttpMessage){ final FullHttpMessage request = (FullHttpMessage) msg; - final boolean keepAlive = HttpUtil.isKeepAlive(request); - final RequestMessage requestMessage; try { - requestMessage = HttpHandlerUtil.getRequestMessageFromHttpRequest((FullHttpRequest) request); - } catch (IllegalArgumentException iae) { - HttpHandlerUtil.sendError(ctx, BAD_REQUEST, iae.getMessage(), keepAlive); - return; - } - - final AuthenticatedUser channelUser = ctx.channel().attr(StateKey.AUTHENTICATED_USER).get(); - // channelUser is null when using the AllowAllAuthenticator - final AuthenticatedUser user = null == channelUser ? - AuthenticatedUser.ANONYMOUS_USER : channelUser; - try { - authorizer.authorize(user, requestMessage); - ctx.fireChannelRead(request); - } catch (AuthorizationException ex) { // Expected: users can alternate between allowed and disallowed requests - String address = ctx.channel().remoteAddress().toString(); - if (address.startsWith("/") && address.length() > 1) address = address.substring(1); - final String script; + final boolean keepAlive = HttpUtil.isKeepAlive(request); + final RequestMessage requestMessage; try { - script = HttpHandlerUtil.getRequestMessageFromHttpRequest((FullHttpRequest) request).getArgOrDefault(Tokens.ARGS_GREMLIN, ""); + requestMessage = HttpHandlerUtil.getRequestMessageFromHttpRequest((FullHttpRequest) request); } catch (IllegalArgumentException iae) { - HttpHandlerUtil.sendError(ctx, BAD_REQUEST, requestMessage.getRequestId(), iae.getMessage(), keepAlive); + HttpHandlerUtil.sendError(ctx, BAD_REQUEST, iae.getMessage(), keepAlive); return; } - auditLogger.info("User {} with address {} attempted an unauthorized http request: {}", - user.getName(), address, script); - final String message = String.format("No authorization for script [%s] - check permissions.", script); - HttpHandlerUtil.sendError(ctx, UNAUTHORIZED, requestMessage.getRequestId(), message, keepAlive); - ReferenceCountUtil.release(msg); - } catch (Exception ex) { - final String message = String.format( - "%s is not ready to handle requests - unknown error", authorizer.getClass().getSimpleName()); - HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestMessage.getRequestId(), message, keepAlive); + + final AuthenticatedUser channelUser = ctx.channel().attr(StateKey.AUTHENTICATED_USER).get(); + // channelUser is null when using the AllowAllAuthenticator + final AuthenticatedUser user = null == channelUser ? + AuthenticatedUser.ANONYMOUS_USER : channelUser; + try { + authorizer.authorize(user, requestMessage); + ctx.fireChannelRead(ReferenceCountUtil.retain(request)); + } catch (AuthorizationException ex) { // Expected: users can alternate between allowed and disallowed requests + String address = ctx.channel().remoteAddress().toString(); + if (address.startsWith("/") && address.length() > 1) address = address.substring(1); + final String script; + try { + script = HttpHandlerUtil.getRequestMessageFromHttpRequest((FullHttpRequest) request).getArgOrDefault(Tokens.ARGS_GREMLIN, ""); + } catch (IllegalArgumentException iae) { + HttpHandlerUtil.sendError(ctx, BAD_REQUEST, requestMessage.getRequestId(), iae.getMessage(), keepAlive); + return; + } + auditLogger.info("User {} with address {} attempted an unauthorized http request: {}", + user.getName(), address, script); + final String message = String.format("No authorization for script [%s] - check permissions.", script); + HttpHandlerUtil.sendError(ctx, UNAUTHORIZED, requestMessage.getRequestId(), message, keepAlive); + } catch (Exception ex) { + final String message = String.format( + "%s is not ready to handle requests - unknown error", authorizer.getClass().getSimpleName()); + HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestMessage.getRequestId(), message, keepAlive); + } + } finally { ReferenceCountUtil.release(msg); } } else { logger.warn("{} only processes FullHttpMessage instances - received {} - channel closing", this.getClass().getSimpleName(), msg.getClass()); + ReferenceCountUtil.release(msg); ctx.close(); } } diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java index 0aa424d3bc..1554aa53e8 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java @@ -129,195 +129,197 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(final ChannelHandlerContext ctx, final Object msg) { - if (msg instanceof FullHttpRequest) { - final FullHttpRequest req = (FullHttpRequest) msg; - final boolean keepAlive = HttpUtil.isKeepAlive(req); - - if ("/favicon.ico".equals(req.uri())) { - HttpHandlerUtil.sendError(ctx, NOT_FOUND, "Gremlin Server doesn't have a favicon.ico", keepAlive); - ReferenceCountUtil.release(msg); - return; - } + boolean release = msg instanceof FullHttpRequest; + try { + if (msg instanceof FullHttpRequest) { + final FullHttpRequest req = (FullHttpRequest) msg; + final boolean keepAlive = HttpUtil.isKeepAlive(req); + + if ("/favicon.ico".equals(req.uri())) { + HttpHandlerUtil.sendError(ctx, NOT_FOUND, "Gremlin Server doesn't have a favicon.ico", keepAlive); + return; + } - if (HttpUtil.is100ContinueExpected(req)) { - ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE)); - } + if (HttpUtil.is100ContinueExpected(req)) { + ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE)); + } - if (req.method() != GET && req.method() != POST) { - HttpHandlerUtil.sendError(ctx, METHOD_NOT_ALLOWED, METHOD_NOT_ALLOWED.toString(), keepAlive); - ReferenceCountUtil.release(msg); - return; - } + if (req.method() != GET && req.method() != POST) { + HttpHandlerUtil.sendError(ctx, METHOD_NOT_ALLOWED, METHOD_NOT_ALLOWED.toString(), keepAlive); + return; + } - final RequestMessage requestMessage; - try { - requestMessage = HttpHandlerUtil.getRequestMessageFromHttpRequest(req, serializers); - } catch (IllegalArgumentException|SerializationException ex) { - HttpHandlerUtil.sendError(ctx, BAD_REQUEST, ex.getMessage(), keepAlive); - ReferenceCountUtil.release(msg); - return; - } + final RequestMessage requestMessage; + try { + requestMessage = HttpHandlerUtil.getRequestMessageFromHttpRequest(req, serializers); + } catch (IllegalArgumentException|SerializationException ex) { + HttpHandlerUtil.sendError(ctx, BAD_REQUEST, ex.getMessage(), keepAlive); + return; + } - final UUID requestId = requestMessage.getRequestId(); - final String acceptMime = Optional.ofNullable(req.headers().get(HttpHeaderNames.ACCEPT)).orElse("application/json"); - final Pair<String, MessageTextSerializer<?>> serializer = chooseSerializer(acceptMime); - if (null == serializer) { - HttpHandlerUtil.sendError(ctx, BAD_REQUEST, requestId, String.format("no serializer for requested Accept header: %s", acceptMime), - keepAlive); - ReferenceCountUtil.release(msg); - return; - } + final UUID requestId = requestMessage.getRequestId(); + final String acceptMime = Optional.ofNullable(req.headers().get(HttpHeaderNames.ACCEPT)).orElse("application/json"); + final Pair<String, MessageTextSerializer<?>> serializer = chooseSerializer(acceptMime); + if (null == serializer) { + HttpHandlerUtil.sendError(ctx, BAD_REQUEST, requestId, String.format("no serializer for requested Accept header: %s", acceptMime), + keepAlive); + return; + } - final String origin = req.headers().get(HttpHeaderNames.ORIGIN); + final String origin = req.headers().get(HttpHeaderNames.ORIGIN); - // not using the req anywhere below here - assume it is safe to release at this point. - ReferenceCountUtil.release(msg); + // not using the req anywhere below here - assume it is safe to release at this point. + ReferenceCountUtil.release(msg); + release = false; - try { - logger.debug("Processing request containing script [{}] and bindings of [{}] on {}", + try { + logger.debug("Processing request containing script [{}] and bindings of [{}] on {}", requestMessage.getArgOrDefault(Tokens.ARGS_GREMLIN, ""), requestMessage.getArgOrDefault(Tokens.ARGS_BINDINGS, Collections.emptyMap()), Thread.currentThread().getName()); - if (settings.enableAuditLog) { - AuthenticatedUser user = ctx.channel().attr(StateKey.AUTHENTICATED_USER).get(); - if (null == user) { // This is expected when using the AllowAllAuthenticator - user = AuthenticatedUser.ANONYMOUS_USER; + if (settings.enableAuditLog) { + AuthenticatedUser user = ctx.channel().attr(StateKey.AUTHENTICATED_USER).get(); + if (null == user) { // This is expected when using the AllowAllAuthenticator + user = AuthenticatedUser.ANONYMOUS_USER; + } + String address = ctx.channel().remoteAddress().toString(); + if (address.startsWith("/") && address.length() > 1) address = address.substring(1); + auditLogger.info("User {} with address {} requested: {}", user.getName(), address, + requestMessage.getArgOrDefault(Tokens.ARGS_GREMLIN, "")); } - String address = ctx.channel().remoteAddress().toString(); - if (address.startsWith("/") && address.length() > 1) address = address.substring(1); - auditLogger.info("User {} with address {} requested: {}", user.getName(), address, - requestMessage.getArgOrDefault(Tokens.ARGS_GREMLIN, "")); - } - final ChannelPromise promise = ctx.channel().newPromise(); - final AtomicReference<Object> resultHolder = new AtomicReference<>(); - promise.addListener(future -> { - // if failed then the error was already written back to the client as part of the eval future - // processing of the exception - if (future.isSuccess()) { - logger.debug("Preparing HTTP response for request with script [{}] and bindings of [{}] with result of [{}] on [{}]", - requestMessage.getArgOrDefault(Tokens.ARGS_GREMLIN, ""), - requestMessage.getArgOrDefault(Tokens.ARGS_BINDINGS, Collections.emptyMap()), - resultHolder.get(), Thread.currentThread().getName()); - final FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, (ByteBuf) resultHolder.get()); - response.headers().set(HttpHeaderNames.CONTENT_TYPE, serializer.getValue0()); - - // handle cors business - if (origin != null) response.headers().set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, origin); - - HttpHandlerUtil.sendAndCleanupConnection(ctx, keepAlive, response); + final ChannelPromise promise = ctx.channel().newPromise(); + final AtomicReference<Object> resultHolder = new AtomicReference<>(); + promise.addListener(future -> { + // if failed then the error was already written back to the client as part of the eval future + // processing of the exception + if (future.isSuccess()) { + logger.debug("Preparing HTTP response for request with script [{}] and bindings of [{}] with result of [{}] on [{}]", + requestMessage.getArgOrDefault(Tokens.ARGS_GREMLIN, ""), + requestMessage.getArgOrDefault(Tokens.ARGS_BINDINGS, Collections.emptyMap()), + resultHolder.get(), Thread.currentThread().getName()); + final FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, (ByteBuf) resultHolder.get()); + response.headers().set(HttpHeaderNames.CONTENT_TYPE, serializer.getValue0()); + + // handle cors business + if (origin != null) response.headers().set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, origin); + + HttpHandlerUtil.sendAndCleanupConnection(ctx, keepAlive, response); + } + }); + + final Timer.Context timerContext = evalOpTimer.time(); + + final Bindings bindings; + try { + bindings = createBindings(requestMessage.getArgOrDefault(Tokens.ARGS_BINDINGS, Collections.emptyMap()), + requestMessage.getArgOrDefault(Tokens.ARGS_ALIASES, Collections.emptyMap())); + } catch (IllegalStateException iae) { + HttpHandlerUtil.sendError(ctx, BAD_REQUEST, requestId, iae.getMessage(), keepAlive); + return; } - }); - - final Timer.Context timerContext = evalOpTimer.time(); - final Bindings bindings; - try { - bindings = createBindings(requestMessage.getArgOrDefault(Tokens.ARGS_BINDINGS, Collections.emptyMap()), - requestMessage.getArgOrDefault(Tokens.ARGS_ALIASES, Collections.emptyMap())); - } catch (IllegalStateException iae) { - HttpHandlerUtil.sendError(ctx, BAD_REQUEST, requestId, iae.getMessage(), keepAlive); - ReferenceCountUtil.release(msg); - return; - } - - // provide a transform function to serialize to message - this will force serialization to occur - // in the same thread as the eval. after the CompletableFuture is returned from the eval the result - // is ready to be written as a ByteBuf directly to the response. nothing should be blocking here. - final CompletableFuture<Object> evalFuture = gremlinExecutor.eval( - requestMessage.getArg(Tokens.ARGS_GREMLIN), requestMessage.getArg(Tokens.ARGS_LANGUAGE), bindings, - requestMessage.getArgOrDefault(Tokens.ARGS_EVAL_TIMEOUT, null), - FunctionUtils.wrapFunction(o -> { - // stopping the timer here is roughly equivalent to where the timer would have been stopped for - // this metric in other contexts. we just want to measure eval time not serialization time. - timerContext.stop(); - - logger.debug("Transforming result of request with script [{}] and bindings of [{}] with result of [{}] on [{}]", - requestMessage.getArg(Tokens.ARGS_GREMLIN), - requestMessage.getArg(Tokens.ARGS_BINDINGS), o, Thread.currentThread().getName()); - - final Optional<String> mp = requestMessage.getArg(Tokens.ARGS_GREMLIN) instanceof String - ? GremlinScriptChecker.parse(requestMessage.getArg(Tokens.ARGS_GREMLIN)).getMaterializeProperties() - : Optional.empty(); - - // need to replicate what TraversalOpProcessor does with the bytecode op. it converts - // results to Traverser so that GLVs can handle the results. don't quite get the same - // benefit here because the bulk has to be 1 since we've already resolved the result, - // but at least http is compatible - final List<Object> results = requestMessage.getOp().equals(Tokens.OPS_BYTECODE) ? - (List<Object>) IteratorUtils.asList(o).stream().map(r -> new DefaultRemoteTraverser<Object>(r, 1)).collect(Collectors.toList()) : - IteratorUtils.asList(o); - - if (mp.isPresent() && mp.get().equals(Tokens.MATERIALIZE_PROPERTIES_TOKENS)) { - final Object firstElement = results.get(0); - - if (firstElement instanceof Element) { - for (int i = 0; i < results.size(); i++) - results.set(i, ReferenceFactory.detach((Element) results.get(i))); - } else if (firstElement instanceof AbstractTraverser) { - for (final Object item : results) - ((AbstractTraverser) item).detach(); - } - } - - final ResponseMessage responseMessage = ResponseMessage.build(requestId) - .code(ResponseStatusCode.SUCCESS) - .result(results).create(); - - // http server is sessionless and must handle commit on transactions. the commit occurs - // before serialization to be consistent with how things work for websocket based - // communication. this means that failed serialization does not mean that you won't get - // a commit to the database - attemptCommit(requestMessage.getArg(Tokens.ARGS_ALIASES), graphManager, settings.strictTransactionManagement); - - try { - return Unpooled.wrappedBuffer(serializer.getValue1().serializeResponseAsBinary(responseMessage, ctx.alloc())); - } catch (Exception ex) { - logger.warn(String.format("Error during serialization for %s", responseMessage), ex); - - // creating a new SerializationException will clear the cause which will allow the - // future to report a better error message. if the cause is present, then - // GremlinExecutor will prefer the cause and we'll get a low level Jackson sort of - // error in the response. - if (ex instanceof SerializationException) { - throw new SerializationException(String.format( - "Could not serialize the result with %s - %s", - serializer.getValue0(), - ex.getMessage())); + // provide a transform function to serialize to message - this will force serialization to occur + // in the same thread as the eval. after the CompletableFuture is returned from the eval the result + // is ready to be written as a ByteBuf directly to the response. nothing should be blocking here. + final CompletableFuture<Object> evalFuture = gremlinExecutor.eval( + requestMessage.getArg(Tokens.ARGS_GREMLIN), requestMessage.getArg(Tokens.ARGS_LANGUAGE), bindings, + requestMessage.getArgOrDefault(Tokens.ARGS_EVAL_TIMEOUT, null), + FunctionUtils.wrapFunction(o -> { + // stopping the timer here is roughly equivalent to where the timer would have been stopped for + // this metric in other contexts. we just want to measure eval time not serialization time. + timerContext.stop(); + + logger.debug("Transforming result of request with script [{}] and bindings of [{}] with result of [{}] on [{}]", + requestMessage.getArg(Tokens.ARGS_GREMLIN), + requestMessage.getArg(Tokens.ARGS_BINDINGS), o, Thread.currentThread().getName()); + + final Optional<String> mp = requestMessage.getArg(Tokens.ARGS_GREMLIN) instanceof String + ? GremlinScriptChecker.parse(requestMessage.getArg(Tokens.ARGS_GREMLIN)).getMaterializeProperties() + : Optional.empty(); + + // need to replicate what TraversalOpProcessor does with the bytecode op. it converts + // results to Traverser so that GLVs can handle the results. don't quite get the same + // benefit here because the bulk has to be 1 since we've already resolved the result, + // but at least http is compatible + final List<Object> results = requestMessage.getOp().equals(Tokens.OPS_BYTECODE) ? + (List<Object>) IteratorUtils.asList(o).stream().map(r -> new DefaultRemoteTraverser<Object>(r, 1)).collect(Collectors.toList()) : + IteratorUtils.asList(o); + + if (mp.isPresent() && mp.get().equals(Tokens.MATERIALIZE_PROPERTIES_TOKENS)) { + final Object firstElement = results.get(0); + + if (firstElement instanceof Element) { + for (int i = 0; i < results.size(); i++) + results.set(i, ReferenceFactory.detach((Element) results.get(i))); + } else if (firstElement instanceof AbstractTraverser) { + for (final Object item : results) + ((AbstractTraverser) item).detach(); + } } - throw ex; - } - })); - - evalFuture.exceptionally(t -> { - if (t.getMessage() != null) - HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, t.getMessage(), Optional.of(t), keepAlive); - else - HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, String.format("Error encountered evaluating script: %s", - requestMessage.getArg(Tokens.ARGS_GREMLIN)) - , Optional.of(t), keepAlive); - promise.setFailure(t); - return null; - }); - - evalFuture.thenAcceptAsync(r -> { - // now that the eval/serialization is done in the same thread - complete the promise so we can - // write back the HTTP response on the same thread as the original request - resultHolder.set(r); - promise.setSuccess(); - }, gremlinExecutor.getExecutorService()); - } catch (Exception ex) { - // send the error response here and don't rely on exception caught because it might not have the - // context on whether to close the connection or not, based on keepalive. - final Throwable t = ExceptionHelper.getRootCause(ex); - if (t instanceof TooLongFrameException) { - HttpHandlerUtil.sendError(ctx, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, requestId, t.getMessage() + " - increase the maxContentLength", keepAlive); - } else if (t != null){ - HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, t.getMessage(), keepAlive); - } else { - HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, ex.getMessage(), keepAlive); + final ResponseMessage responseMessage = ResponseMessage.build(requestId) + .code(ResponseStatusCode.SUCCESS) + .result(results).create(); + + // http server is sessionless and must handle commit on transactions. the commit occurs + // before serialization to be consistent with how things work for websocket based + // communication. this means that failed serialization does not mean that you won't get + // a commit to the database + attemptCommit(requestMessage.getArg(Tokens.ARGS_ALIASES), graphManager, settings.strictTransactionManagement); + + try { + return Unpooled.wrappedBuffer(serializer.getValue1().serializeResponseAsBinary(responseMessage, ctx.alloc())); + } catch (Exception ex) { + logger.warn(String.format("Error during serialization for %s", responseMessage), ex); + + // creating a new SerializationException will clear the cause which will allow the + // future to report a better error message. if the cause is present, then + // GremlinExecutor will prefer the cause and we'll get a low level Jackson sort of + // error in the response. + if (ex instanceof SerializationException) { + throw new SerializationException(String.format( + "Could not serialize the result with %s - %s", + serializer.getValue0(), + ex.getMessage())); + } + + throw ex; + } + })); + + evalFuture.exceptionally(t -> { + if (t.getMessage() != null) + HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, t.getMessage(), Optional.of(t), keepAlive); + else + HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, String.format("Error encountered evaluating script: %s", + requestMessage.getArg(Tokens.ARGS_GREMLIN)) + , Optional.of(t), keepAlive); + promise.setFailure(t); + return null; + }); + + evalFuture.thenAcceptAsync(r -> { + // now that the eval/serialization is done in the same thread - complete the promise so we can + // write back the HTTP response on the same thread as the original request + resultHolder.set(r); + promise.setSuccess(); + }, gremlinExecutor.getExecutorService()); + } catch (Exception ex) { + // send the error response here and don't rely on exception caught because it might not have the + // context on whether to close the connection or not, based on keepalive. + final Throwable t = ExceptionHelper.getRootCause(ex); + if (t instanceof TooLongFrameException) { + HttpHandlerUtil.sendError(ctx, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, requestId, t.getMessage() + " - increase the maxContentLength", keepAlive); + } else if (t != null){ + HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, t.getMessage(), keepAlive); + } else { + HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, ex.getMessage(), keepAlive); + } } } + } finally { + if (release) + ReferenceCountUtil.release(msg); } } diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtil.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtil.java index 7b1036cc30..62a0066e6e 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtil.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtil.java @@ -107,19 +107,26 @@ public class HttpHandlerUtil { final ByteBuf buffer = request.content(); - // additional validation for header - final int first = buffer.readByte(); - // payload can be plain json or can start with additional header with content type. - // if first character is not "{" (0x7b) then need to verify is correct serializer selected. + if (!buffer.isReadable()) + throw new IllegalArgumentException("Request body is empty."); + + // Driver payloads may start with [unsigned MIME length][MIME bytes], followed by the serialized request. + // Plain GraphSON starts directly with '{', so peek without advancing until the payload form is known. + final int first = buffer.getUnsignedByte(buffer.readerIndex()); if (first != 0x7b) { + // Include the length byte itself when verifying that the complete MIME header is available. + if (buffer.readableBytes() < first + 1) + throw new IllegalArgumentException("Request body is shorter than the mime type header."); + + // Remove the MIME header so the serializer receives only the RequestMessage payload. + buffer.skipBytes(1); final byte[] bytes = new byte[first]; buffer.readBytes(bytes); final String mimeType = new String(bytes, StandardCharsets.UTF_8); if (Arrays.stream(serializer.mimeTypesSupported()).noneMatch(t -> t.equals(mimeType))) throw new IllegalArgumentException("Mime type mismatch. Value in content-type header is not equal payload header."); - } else - buffer.resetReaderIndex(); + } return serializer.deserializeRequest(buffer); } diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpBasicAuthorizationHandlerTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpBasicAuthorizationHandlerTest.java index 2e0e6e6b34..992785fa83 100644 --- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpBasicAuthorizationHandlerTest.java +++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpBasicAuthorizationHandlerTest.java @@ -18,18 +18,25 @@ */ package org.apache.tinkerpop.gremlin.server.handler; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.http.DefaultFullHttpRequest; +import io.netty.handler.codec.http.DefaultHttpHeaders; import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.FullHttpResponse; import io.netty.handler.codec.http.HttpMethod; import io.netty.handler.codec.http.HttpVersion; import io.netty.handler.codec.http.QueryStringEncoder; +import io.netty.util.CharsetUtil; +import io.netty.util.ReferenceCountUtil; import org.apache.tinkerpop.gremlin.process.traversal.Bytecode; import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser; import org.apache.tinkerpop.gremlin.server.authz.AuthorizationException; import org.apache.tinkerpop.gremlin.server.authz.Authorizer; 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.junit.Test; import java.util.Map; @@ -40,11 +47,31 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; public class HttpBasicAuthorizationHandlerTest { + private final Authorizer authorizer = new Authorizer() { + @Override + public void setup(final Map<String, Object> config) throws AuthorizationException { + } + + @Override + public Bytecode authorize(final AuthenticatedUser user, final Bytecode bytecode, + final Map<String, String> aliases) throws AuthorizationException { + return bytecode; + } + + @Override + public void authorize(final AuthenticatedUser user, final RequestMessage msg) throws AuthorizationException { + } + }; + @Test public void shouldHandleRejectedRequestWithTheUserFromItsChannel() throws Exception { final CountDownLatch firstRequestInAuthorizer = new CountDownLatch(1); @@ -141,4 +168,44 @@ public class HttpBasicAuthorizationHandlerTest { return nameCalls.get(); } } + + @Test + public void shouldReleaseRejectedGraphBinaryRequest() throws Exception { + final RequestMessage requestMessage = RequestMessage.build("eval").addArg("gremlin", "g.V()").create(); + final FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/", + new GraphBinaryMessageSerializerV1().serializeRequestAsBinary(requestMessage, ByteBufAllocator.DEFAULT)); + final EmbeddedChannel channel = new EmbeddedChannel(new HttpBasicAuthorizationHandler(authorizer)); + FullHttpResponse response = null; + try { + assertFalse(channel.writeInbound(request)); + + assertEquals(0, request.refCnt()); + response = channel.readOutbound(); + assertNotNull(response); + assertEquals(BAD_REQUEST, response.status()); + } finally { + ReferenceCountUtil.release(response); + channel.finishAndReleaseAll(); + } + } + + @Test + public void shouldTransferAuthorizedRequestOwnership() { + final ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(); + buffer.writeCharSequence("{\"gremlin\":\"g.V()\"}", CharsetUtil.UTF_8); + final FullHttpRequest request = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, HttpMethod.POST, "/", buffer, new DefaultHttpHeaders(), new DefaultHttpHeaders()); + final EmbeddedChannel channel = new EmbeddedChannel(new HttpBasicAuthorizationHandler(authorizer)); + FullHttpRequest forwarded = null; + try { + assertTrue(channel.writeInbound(request)); + + assertEquals(1, request.refCnt()); + forwarded = channel.readInbound(); + assertSame(request, forwarded); + } finally { + ReferenceCountUtil.release(forwarded); + channel.finishAndReleaseAll(); + } + } } diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandlerTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandlerTest.java new file mode 100644 index 0000000000..75573580f1 --- /dev/null +++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandlerTest.java @@ -0,0 +1,116 @@ +/* + * 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.ByteBufAllocator; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.DefaultFullHttpRequest; +import io.netty.handler.codec.http.DefaultHttpHeaders; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.FullHttpResponse; +import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.HttpVersion; +import io.netty.util.CharsetUtil; +import io.netty.util.ReferenceCountUtil; +import org.apache.tinkerpop.gremlin.util.MessageSerializer; +import org.apache.tinkerpop.gremlin.util.ser.GraphBinaryMessageSerializerV1; +import org.apache.tinkerpop.gremlin.util.ser.SerTokens; +import org.junit.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; + +public class HttpGremlinEndpointHandlerTest { + + @Test + public void shouldReleaseRequestWithEmptySerializedBody() { + assertMalformedSerializedRequestReleased(new byte[0]); + } + + @Test + public void shouldReleaseRequestWithUnsignedMimeTypeLength() { + final byte[] payload = new byte[129]; + payload[0] = (byte) 0x80; + + assertMalformedSerializedRequestReleased(payload); + } + + @Test + public void shouldReleaseRequestWithTruncatedMimeTypeHeader() { + assertMalformedSerializedRequestReleased(new byte[] { 5, 'a', 'b' }); + } + + @Test + public void shouldReleaseRequestWhenRuntimeExceptionIsThrown() { + final ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(); + buffer.writeCharSequence("{\"gremlin\":\"g.V()\"}", CharsetUtil.UTF_8); + + final DefaultHttpHeaders headers = new DefaultHttpHeaders(); + headers.add(HttpHeaderNames.CONTENT_TYPE, "application/json"); + headers.add(HttpHeaderNames.ACCEPT, "application/json;q=invalid"); + final FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/", + buffer, headers, new DefaultHttpHeaders()); + final HttpGremlinEndpointHandler handler = new HttpGremlinEndpointHandler( + Collections.emptyMap(), null, null, null); + + try { + handler.channelRead(null, request); + fail("NumberFormatException expected"); + } catch (NumberFormatException ex) { + assertEquals("For input string: \"invalid\"", ex.getMessage()); + } + + assertEquals(0, request.refCnt()); + } + + private void assertMalformedSerializedRequestReleased(final byte[] payload) { + final ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(payload.length); + buffer.writeBytes(payload); + + final DefaultHttpHeaders headers = new DefaultHttpHeaders(); + headers.add(HttpHeaderNames.CONTENT_TYPE, SerTokens.MIME_GRAPHBINARY_V1); + final FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/", + buffer, headers, new DefaultHttpHeaders()); + + final Map<String, MessageSerializer<?>> serializers = new HashMap<>(); + serializers.put(SerTokens.MIME_GRAPHBINARY_V1, new GraphBinaryMessageSerializerV1()); + final EmbeddedChannel channel = new EmbeddedChannel( + new HttpGremlinEndpointHandler(serializers, null, null, null)); + FullHttpResponse response = null; + try { + channel.writeInbound(request); + + assertEquals(0, request.refCnt()); + response = channel.readOutbound(); + assertNotNull(response); + assertEquals(BAD_REQUEST, response.status()); + } finally { + ReferenceCountUtil.release(response); + channel.finishAndReleaseAll(); + } + } +} diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtilTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtilTest.java index 8ef5b05e6b..e679021556 100644 --- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtilTest.java +++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtilTest.java @@ -48,6 +48,7 @@ import java.util.UUID; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.samePropertyValuesAs; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; public class HttpHandlerUtilTest { @@ -83,6 +84,26 @@ public class HttpHandlerUtilTest { } } + @Test + public void shouldRejectEmptySerializedRequest() { + assertInvalidSerializedRequest(new byte[0], "Request body is empty."); + } + + @Test + public void shouldTreatSerializedRequestHeaderLengthAsUnsigned() { + final byte[] payload = new byte[129]; + payload[0] = (byte) 0x80; + + assertInvalidSerializedRequest(payload, + "Mime type mismatch. Value in content-type header is not equal payload header."); + } + + @Test + public void shouldRejectTruncatedSerializedRequestHeader() { + assertInvalidSerializedRequest(new byte[] { 5, 'a', 'b' }, + "Request body is shorter than the mime type header."); + } + @Test public void shouldCorrectlyDeserializeRequestMessage() throws SerializationException { final RequestMessage request = RequestMessage.build(Tokens.OPS_BYTECODE) @@ -162,4 +183,27 @@ public class HttpHandlerUtilTest { assertEquals("gremlin-groovy", deserialized.getArg(Tokens.ARGS_LANGUAGE)); } } + + private void assertInvalidSerializedRequest(final byte[] payload, final String expectedMessage) { + final ByteBuf buffer = allocator.buffer(payload.length); + buffer.writeBytes(payload); + + final HttpHeaders headers = new DefaultHttpHeaders(); + headers.add(HttpHeaderNames.CONTENT_TYPE, SerTokens.MIME_GRAPHBINARY_V1); + final FullHttpRequest httpRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/", + buffer, headers, new DefaultHttpHeaders()); + + final Map<String, MessageSerializer<?>> serializers = new HashMap<>(); + serializers.put(SerTokens.MIME_GRAPHBINARY_V1, graphBinarySerializer); + + try { + HttpHandlerUtil.getRequestMessageFromHttpRequest(httpRequest, serializers); + fail("IllegalArgumentException expected"); + } catch (Exception ex) { + assertTrue(ex instanceof IllegalArgumentException); + assertEquals(expectedMessage, ex.getMessage()); + } finally { + httpRequest.release(); + } + } }
