diff --git a/gremlin-dotnet/docker-compose.yml b/gremlin-dotnet/docker-compose.yml index d68175d0e30..baa31e1d067 100644 --- a/gremlin-dotnet/docker-compose.yml +++ b/gremlin-dotnet/docker-compose.yml @@ -52,6 +52,7 @@ services: - SOCKET_SERVER_VERSION=${GREMLIN_SERVER} ports: - "45943:45943" + - "45944:45944" gremlin-dotnet-integration-tests: container_name: gremlin-dotnet-integration-tests @@ -67,6 +68,7 @@ services: - GREMLIN_SERVER_PORT=45940 - GREMLIN_SECURE_SERVER_PORT=45941 - GREMLIN_SOCKET_SERVER_URL=http://gremlin-socket-server-test-dotnet:45943/gremlin + - GREMLIN_SOCKET_SERVER_PROXY_URL=http://gremlin-socket-server-test-dotnet:45944 - VERTEX_LABEL=dotnet-example working_dir: /gremlin-dotnet command: > diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ProxyBehaviorIntegrationTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ProxyBehaviorIntegrationTests.cs new file mode 100644 index 00000000000..5653cabbafe --- /dev/null +++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ProxyBehaviorIntegrationTests.cs @@ -0,0 +1,108 @@ +#region License + +/* + * 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. + */ + +#endregion + +using System; +using System.Net; +using System.Net.Http; +using System.Text.Json; +using System.Threading.Tasks; +using Gremlin.Net.Driver; +using Xunit; + +namespace Gremlin.Net.IntegrationTest.Driver +{ + public class ProxyBehaviorIntegrationTests + { + private static readonly string Host = GetHost(); + private static readonly string ProxyUrl = GetProxyUrl(); + + private static string GetHost() + { + var url = Environment.GetEnvironmentVariable("GREMLIN_SOCKET_SERVER_URL"); + if (string.IsNullOrEmpty(url)) return "localhost"; + try + { + return new Uri(url).Host; + } + catch + { + return "localhost"; + } + } + + private static string GetProxyUrl() + { + var url = Environment.GetEnvironmentVariable("GREMLIN_SOCKET_SERVER_PROXY_URL"); + return string.IsNullOrEmpty(url) ? "http://localhost:45944" : url; + } + + private static async Task ResetProxyAsync(HttpClient httpClient) + { + using var response = await httpClient.PostAsync($"{ProxyUrl}/__reset", null); + response.EnsureSuccessStatusCode(); + } + + private static async Task GetRecordedAsync(HttpClient httpClient) + { + var json = await httpClient.GetStringAsync($"{ProxyUrl}/__recorded"); + return JsonSerializer.Deserialize(json) ?? Array.Empty(); + } + + [Fact] + public async Task ShouldRouteSocketServerTrafficThroughProxy() + { + using var httpClient = new HttpClient(); + await ResetProxyAsync(httpClient); + + var server = new GremlinServer(Host, SocketServerConstants.Port); + using (var client = new GremlinClient(server, + connectionSettings: new ConnectionSettings { Proxy = new WebProxy(ProxyUrl) })) + { + var resultSet = await client.SubmitAsync(SocketServerConstants.GremlinSingleVertex); + var results = await resultSet.ToListAsync(); + Assert.Single(results); + } + + var recorded = await GetRecordedAsync(httpClient); + Assert.Contains(recorded, t => t.EndsWith(":45943")); + } + + [Fact] + public async Task ShouldNotRecordWhenNoProxy() + { + using var httpClient = new HttpClient(); + await ResetProxyAsync(httpClient); + + var server = new GremlinServer(Host, SocketServerConstants.Port); + using (var client = new GremlinClient(server)) + { + var resultSet = await client.SubmitAsync(SocketServerConstants.GremlinSingleVertex); + var results = await resultSet.ToListAsync(); + Assert.Single(results); + } + + var recorded = await GetRecordedAsync(httpClient); + Assert.DoesNotContain(recorded, t => t.EndsWith(":45943")); + } + } +} diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java index bab50ede53c..04cea7823d1 100644 --- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java +++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java @@ -29,6 +29,9 @@ import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslHandler; import io.netty.handler.proxy.HttpProxyHandler; + +import java.net.InetSocketAddress; +import java.net.SocketAddress; import io.netty.handler.timeout.IdleStateHandler; import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException; import org.apache.tinkerpop.gremlin.driver.handler.GremlinResponseHandler; @@ -147,9 +150,10 @@ protected void initChannel(final SocketChannel socketChannel) { // TLS handshake. final ProxyOptions proxy = cluster.getProxy(); if (proxy != null) { + final SocketAddress proxyAddress = resolveProxyAddress(proxy.getAddress()); final HttpProxyHandler proxyHandler = proxy.hasCredentials() - ? new HttpProxyHandler(proxy.getAddress(), proxy.getUsername(), proxy.getPassword()) - : new HttpProxyHandler(proxy.getAddress()); + ? new HttpProxyHandler(proxyAddress, proxy.getUsername(), proxy.getPassword()) + : new HttpProxyHandler(proxyAddress); pipeline.addLast(PIPELINE_PROXY_HANDLER, proxyHandler); } @@ -189,6 +193,21 @@ public void connected() { } } } + + /** + * Resolves an unresolved proxy {@link SocketAddress} so that Netty's {@code HttpProxyHandler} can connect to + * it. A resolved address (or a non-{@link InetSocketAddress}) is returned unchanged. + */ + private static SocketAddress resolveProxyAddress(final SocketAddress address) { + if (address instanceof InetSocketAddress) { + final InetSocketAddress inet = (InetSocketAddress) address; + if (inet.isUnresolved()) { + return new InetSocketAddress(inet.getHostString(), inet.getPort()); + } + } + return address; + } + } /** diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ProxyBehaviorIntegrateTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ProxyBehaviorIntegrateTest.java new file mode 100644 index 00000000000..0da00b9d530 --- /dev/null +++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ProxyBehaviorIntegrateTest.java @@ -0,0 +1,143 @@ +/* + * 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.driver; + +import org.apache.tinkerpop.gremlin.socket.server.RecordingProxyServer; +import org.apache.tinkerpop.gremlin.socket.server.SimpleTestServer; +import org.apache.tinkerpop.gremlin.socket.server.SocketServerConstants; +import org.apache.tinkerpop.gremlin.socket.server.TestHttpServerInitializer; +import org.apache.tinkerpop.gremlin.structure.Vertex; +import org.apache.tinkerpop.gremlin.util.ser.GraphBinaryMessageSerializerV4; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.util.List; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.core.IsInstanceOf.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * Proves that {@link Cluster.Builder#proxy(ProxyOptions)} routes driver traffic through an HTTP proxy. + *

+ * A {@link SimpleTestServer} (the real Gremlin socket server) runs on {@link SocketServerConstants#PORT} + * and an in-process {@link RecordingProxyServer} runs on {@link SocketServerConstants#PROXY_PORT}. The + * proxy records the {@code host:port} target of every tunnel it establishes and exposes a small control + * API ({@code GET /__recorded}, {@code POST /__reset}) that the tests query with the JDK11 + * {@link java.net.http.HttpClient}. + */ +public class ProxyBehaviorIntegrateTest { + + private static final int PORT = SocketServerConstants.PORT; + private static final int PROXY_PORT = SocketServerConstants.PROXY_PORT; + + private static SimpleTestServer server; + private static RecordingProxyServer proxyServer; + private static HttpClient httpClient; + + @BeforeClass + public static void setUp() throws InterruptedException { + server = new SimpleTestServer(PORT); + server.start(new TestHttpServerInitializer()); + + proxyServer = new RecordingProxyServer(PROXY_PORT); + proxyServer.start(); + + httpClient = HttpClient.newHttpClient(); + } + + @AfterClass + public static void tearDown() { + if (server != null) server.stop(); + if (proxyServer != null) proxyServer.stop(); + } + + private static Cluster.Builder buildCluster() { + return Cluster.build("localhost") + .validationRequest(SocketServerConstants.GREMLIN_SINGLE_VERTEX) + .port(PORT) + .serializer(new GraphBinaryMessageSerializerV4()); + } + + private static String getRecorded() throws Exception { + final HttpRequest request = HttpRequest.newBuilder() + .uri(URI.create("http://localhost:" + PROXY_PORT + "/__recorded")) + .GET() + .build(); + return httpClient.send(request, HttpResponse.BodyHandlers.ofString()).body(); + } + + private static void resetRecorded() throws Exception { + final HttpRequest request = HttpRequest.newBuilder() + .uri(URI.create("http://localhost:" + PROXY_PORT + "/__reset")) + .POST(HttpRequest.BodyPublishers.noBody()) + .build(); + httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + } + + @Test + public void shouldRouteTrafficThroughProxy() throws Exception { + resetRecorded(); + + final Cluster cluster = buildCluster() + .proxy(ProxyOptions.create("localhost", PROXY_PORT)) + .create(); + try { + final Client client = cluster.connect(); + final List results = client.submit(SocketServerConstants.GREMLIN_SINGLE_VERTEX).all().get(); + assertEquals(1, results.size()); + assertThat(results.get(0).getObject(), instanceOf(Vertex.class)); + + // the driver issues an HTTP CONNECT, so the proxy should have recorded the tunnel to the + // socket server. Match on the ":45943" substring to stay tolerant of the host form. + final String recorded = getRecorded(); + assertTrue("Expected the proxy to have recorded the tunnel to :" + PORT + " but was: " + recorded, + recorded.contains(":" + PORT)); + } finally { + cluster.close(); + } + } + + @Test + public void shouldNotRecordWithoutProxy() throws Exception { + resetRecorded(); + + final Cluster cluster = buildCluster().create(); + try { + final Client client = cluster.connect(); + final List results = client.submit(SocketServerConstants.GREMLIN_SINGLE_VERTEX).all().get(); + assertEquals(1, results.size()); + assertThat(results.get(0).getObject(), instanceOf(Vertex.class)); + + // without a proxy configured the driver connects directly, so nothing should be recorded. + final String recorded = getRecorded(); + assertFalse("Expected no tunnel to :" + PORT + " to be recorded but was: " + recorded, + recorded.contains(":" + PORT)); + } finally { + cluster.close(); + } + } +} diff --git a/gremlin-go/docker-compose.yml b/gremlin-go/docker-compose.yml index 5df15618269..21c915bbab2 100644 --- a/gremlin-go/docker-compose.yml +++ b/gremlin-go/docker-compose.yml @@ -52,6 +52,7 @@ services: - SOCKET_SERVER_VERSION=${GREMLIN_SERVER} ports: - "45943:45943" + - "45944:45944" gremlin-go-integration-tests: container_name: gremlin-go-integration-tests @@ -65,6 +66,7 @@ services: - GREMLIN_SERVER_URL=http://gremlin-server-test:45940/gremlin - GREMLIN_SERVER_BASIC_AUTH_URL=https://gremlin-server-test:45941/gremlin - GREMLIN_SOCKET_SERVER_URL=http://gremlin-socket-server-test-go:45943/gremlin + - GREMLIN_SOCKET_SERVER_PROXY_URL=http://gremlin-socket-server-test-go:45944 - RUN_INTEGRATION_TESTS=true - RUN_INTEGRATION_WITH_ALIAS_TESTS=true - RUN_BASIC_AUTH_INTEGRATION_TESTS=true diff --git a/gremlin-go/driver/client_behavior_test.go b/gremlin-go/driver/client_behavior_test.go index f274c0ad2c5..b358b064041 100644 --- a/gremlin-go/driver/client_behavior_test.go +++ b/gremlin-go/driver/client_behavior_test.go @@ -21,7 +21,11 @@ package gremlingo import ( "context" + "encoding/json" "fmt" + "net/http" + "net/url" + "strings" "sync" "testing" "time" @@ -35,6 +39,105 @@ func socketServerURL() string { fmt.Sprintf("http://localhost:%d/gremlin", socketServerPort)) } +func socketServerProxyURL() string { + return getEnvOrDefaultString("GREMLIN_SOCKET_SERVER_PROXY_URL", + "http://localhost:45944") +} + +// resetProxyRecording clears the proxy's recorded targets via POST /__reset. +func resetProxyRecording(t *testing.T) { + t.Helper() + proxyURL := socketServerProxyURL() + resp, err := http.Post(proxyURL+"/__reset", "application/json", nil) + if err != nil { + t.Skipf("Recording proxy not available: %v", err) + } + resp.Body.Close() +} + +// recordedProxyTargets fetches the recorded "host:port" targets via GET /__recorded. +func recordedProxyTargets(t *testing.T) []string { + t.Helper() + proxyURL := socketServerProxyURL() + resp, err := http.Get(proxyURL + "/__recorded") + if err != nil { + t.Skipf("Recording proxy not available: %v", err) + } + defer resp.Body.Close() + + var recorded []string + if err := json.NewDecoder(resp.Body).Decode(&recorded); err != nil { + t.Fatalf("failed to decode recorded proxy targets: %v", err) + } + return recorded +} + +// anyTargetHasPortSuffix reports whether any recorded target ends with ":", +// matching on port suffix only so the host portion is ignored. +func anyTargetHasPortSuffix(targets []string, port int) bool { + suffix := fmt.Sprintf(":%d", port) + for _, target := range targets { + if strings.HasSuffix(target, suffix) { + return true + } + } + return false +} + +// TestProxyRoutesSocketServerTraffic verifies that traffic submitted through a +// client configured with a Proxy is routed to the socket server via the shared +// recording proxy, and that the proxy records the socket server target. +func TestProxyRoutesSocketServerTraffic(t *testing.T) { + resetProxyRecording(t) + + parsedProxyURL, err := url.Parse(socketServerProxyURL()) + require.NoError(t, err) + + client, err := NewClient(socketServerURL(), func(settings *ClientSettings) { + settings.Proxy = func(*http.Request) (*url.URL, error) { + return parsedProxyURL, nil + } + }) + require.NoError(t, err) + defer client.Close() + + rs, err := client.Submit(gremlinSingleVertex) + require.NoError(t, err) + results, err := rs.All() + require.NoError(t, err) + assert.Equal(t, 1, len(results)) + + targets := recordedProxyTargets(t) + assert.True(t, anyTargetHasPortSuffix(targets, socketServerPort), + "expected at least one recorded target ending with :%d, got %v", socketServerPort, targets) +} + +// TestProxyNegativeControl verifies that a client connecting directly to the +// socket server (no Proxy configured) does not route through the recording +// proxy, so the proxy records no socket server target. +func TestProxyNegativeControl(t *testing.T) { + resetProxyRecording(t) + + client, err := NewClient(socketServerURL(), func(settings *ClientSettings) { + // Force a direct connection (no proxy) regardless of ambient proxy env vars. + settings.Proxy = func(*http.Request) (*url.URL, error) { + return nil, nil + } + }) + require.NoError(t, err) + defer client.Close() + + rs, err := client.Submit(gremlinSingleVertex) + require.NoError(t, err) + results, err := rs.All() + require.NoError(t, err) + assert.Equal(t, 1, len(results)) + + targets := recordedProxyTargets(t) + assert.False(t, anyTargetHasPortSuffix(targets, socketServerPort), + "expected no recorded target ending with :%d, got %v", socketServerPort, targets) +} + func newSocketServerClient(t *testing.T, configurations ...func(*ClientSettings)) *Client { t.Helper() url := socketServerURL() diff --git a/gremlin-js/gremlin-javascript/docker-compose.yml b/gremlin-js/gremlin-javascript/docker-compose.yml index 6b90238a308..4d2a1d97c57 100644 --- a/gremlin-js/gremlin-javascript/docker-compose.yml +++ b/gremlin-js/gremlin-javascript/docker-compose.yml @@ -52,6 +52,7 @@ services: - SOCKET_SERVER_VERSION=${GREMLIN_SERVER} ports: - "45943:45943" + - "45944:45944" gremlin-js-integration-tests: container_name: gremlin-js-integration-tests @@ -68,6 +69,7 @@ services: - DOCKER_ENVIRONMENT=true - GREMLIN_SERVER_URL=http://gremlin-server-test-js:45940/gremlin - GREMLIN_SOCKET_SERVER_URL=http://gremlin-socket-server-test-js:45943/gremlin + - GREMLIN_SOCKET_SERVER_PROXY_URL=http://gremlin-socket-server-test-js:45944 - VERTEX_LABEL=javascript-example - IO_TEST_DIRECTORY=/workspace/gremlin-js/gremlin-javascript/gremlin-test/graphbinary/ - NPM_CONFIG_CACHE=/tmp/npm-cache diff --git a/gremlin-js/gremlin-javascript/test/integration/proxy-behavior-tests.js b/gremlin-js/gremlin-javascript/test/integration/proxy-behavior-tests.js new file mode 100644 index 00000000000..7d410c41fe5 --- /dev/null +++ b/gremlin-js/gremlin-javascript/test/integration/proxy-behavior-tests.js @@ -0,0 +1,52 @@ +/* + * 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. + */ + +import assert from 'assert'; + +import Client from '../../lib/driver/client.js'; + +import { GREMLIN_SINGLE_VERTEX } from './socket-server-constants.js'; + +const url = process.env.GREMLIN_SOCKET_SERVER_URL || 'http://localhost:45943/gremlin'; +const proxyUrl = process.env.GREMLIN_SOCKET_SERVER_PROXY_URL || 'http://localhost:45944'; + +describe('proxy behavior', function () { + this.timeout(30000); + + it('routes socket server traffic through the configured proxy', async function () { + await fetch(proxyUrl + '/__reset', { method: 'POST' }); + const client = new Client(url, { traversalSource: 'g', proxy: proxyUrl }); + await client.open?.(); + const result = await client.submit(GREMLIN_SINGLE_VERTEX); + assert.strictEqual(result.length, 1); + const recorded = await (await fetch(proxyUrl + '/__recorded')).json(); + assert.ok(recorded.some((t) => t.endsWith(':45943')), JSON.stringify(recorded)); + await client.close(); + }); + + it('does not record when no proxy is configured', async function () { + await fetch(proxyUrl + '/__reset', { method: 'POST' }); + const client = new Client(url, { traversalSource: 'g' }); + const result = await client.submit(GREMLIN_SINGLE_VERTEX); + assert.strictEqual(result.length, 1); + const recorded = await (await fetch(proxyUrl + '/__recorded')).json(); + assert.ok(!recorded.some((t) => t.endsWith(':45943'))); + await client.close(); + }); +}); diff --git a/gremlin-python/docker-compose.yml b/gremlin-python/docker-compose.yml index 2876bbf048a..795905d424e 100644 --- a/gremlin-python/docker-compose.yml +++ b/gremlin-python/docker-compose.yml @@ -52,6 +52,7 @@ services: - SOCKET_SERVER_VERSION=${GREMLIN_SERVER} ports: - "45943:45943" + - "45944:45944" gremlin-python-integration-tests: container_name: gremlin-python-integration-tests @@ -66,6 +67,7 @@ services: - GREMLIN_SERVER_URL=http://gremlin-server-test-python:{}/gremlin - GREMLIN_SERVER_BASIC_AUTH_URL=https://gremlin-server-test-python:{}/gremlin - GREMLIN_SOCKET_SERVER_URL=http://gremlin-socket-server-test-python:45943/gremlin + - GREMLIN_SOCKET_SERVER_PROXY_URL=http://gremlin-socket-server-test-python:45944 - IO_TEST_DIRECTORY=/python_app/gremlin-test/graphbinary/ - PYTEST_ARGS=${PYTEST_ARGS:-} - RADISH_ARGS=${RADISH_ARGS:-} diff --git a/gremlin-python/src/main/python/tests/integration/driver/test_client_behavior.py b/gremlin-python/src/main/python/tests/integration/driver/test_client_behavior.py index 013280e39fc..18eef5edf8e 100644 --- a/gremlin-python/src/main/python/tests/integration/driver/test_client_behavior.py +++ b/gremlin-python/src/main/python/tests/integration/driver/test_client_behavior.py @@ -18,8 +18,10 @@ # import asyncio +import json import os import time +import urllib.request from concurrent.futures import ThreadPoolExecutor, as_completed import pytest @@ -43,6 +45,7 @@ ) url = os.environ.get('GREMLIN_SOCKET_SERVER_URL', 'http://localhost:{}/gremlin'.format(PORT)) +proxy_url = os.environ.get('GREMLIN_SOCKET_SERVER_PROXY_URL', 'http://localhost:45944') @pytest.fixture(scope="module") @@ -50,7 +53,8 @@ def socket_server_client(): try: client = Client(url, 'g') # Verify connectivity - client.submit(GREMLIN_SINGLE_VERTEX).all().result() + result = client.submit(GREMLIN_SINGLE_VERTEX).all().result() + assert len(result) == 1 except Exception: pytest.skip("Socket server is not available at {}".format(url)) yield client @@ -195,3 +199,34 @@ def submit_close(): assert len(vertex_results) == 5 assert len(close_errors) == 5 + + +def test_proxy_routes_socket_server_traffic(): + # Reset the recording proxy, route a client through it, and confirm the + # proxy observed traffic to the socket server on port 45943. + urllib.request.urlopen(urllib.request.Request(proxy_url + '/__reset', method='POST')) + + client = Client(url, 'g', proxy=proxy_url) + try: + result = client.submit(GREMLIN_SINGLE_VERTEX).all().result() + assert len(result) == 1 + + recorded = json.loads(urllib.request.urlopen(proxy_url + '/__recorded').read().decode()) + assert any(target.endswith(':45943') for target in recorded) + finally: + client.close() + + +def test_proxy_negative_control(): + # Negative control: without the proxy kwarg the client connects directly, + # so the recording proxy should observe no traffic to the socket server. + urllib.request.urlopen(urllib.request.Request(proxy_url + '/__reset', method='POST')) + + client = Client(url, 'g') + try: + client.submit(GREMLIN_SINGLE_VERTEX).all().result() + + recorded = json.loads(urllib.request.urlopen(proxy_url + '/__recorded').read().decode()) + assert not any(target.endswith(':45943') for target in recorded) + finally: + client.close() diff --git a/gremlin-tools/gremlin-socket-server/Dockerfile b/gremlin-tools/gremlin-socket-server/Dockerfile index 3010f33cd64..45c650d1055 100644 --- a/gremlin-tools/gremlin-socket-server/Dockerfile +++ b/gremlin-tools/gremlin-socket-server/Dockerfile @@ -32,5 +32,6 @@ COPY ${SOCKET_SERVER_DIR}/libs/ /opt/gremlin-socket-server/libs/ WORKDIR /opt/gremlin-socket-server EXPOSE 45943 +EXPOSE 45944 ENTRYPOINT ["java","-jar","gremlin-socket-server.jar"] diff --git a/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/RecordingProxyServer.java b/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/RecordingProxyServer.java new file mode 100644 index 00000000000..04f33429b42 --- /dev/null +++ b/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/RecordingProxyServer.java @@ -0,0 +1,366 @@ +/* + * 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.socket.server; + +import io.netty.bootstrap.Bootstrap; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.FullHttpResponse; +import io.netty.handler.codec.http.HttpClientCodec; +import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpServerCodec; +import io.netty.handler.codec.http.HttpUtil; +import io.netty.util.ReferenceCountUtil; + +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1; + +/** + * A self-contained Netty HTTP proxy used for cross-GLV proxy testing. It records the + * {@code host:port} target of every request it proxies and exposes a small control API so tests can + * prove that traffic actually transited the proxy. + *

+ * Three kinds of requests are supported: + *

    + *
  • HTTP CONNECT tunneling (used by Java/Netty and JS/undici clients): the target is + * recorded, a 200 "Connection Established" is returned and the pipeline is reconfigured to relay + * raw bytes bidirectionally between the client and a freshly opened upstream connection.
    • + *
  • Absolute-URI forward requests (used by Go/Python/.NET for http targets): the target + * authority is recorded and the request is forwarded to that target over a new upstream HTTP + * connection with the response relayed back to the client.
    • + *
  • Origin-form control requests: {@code GET /__recorded} returns a JSON array of the + * recorded targets, {@code POST /__reset} clears the recorded list. Any other control path + * returns 404.
    • + *
+ */ +public class RecordingProxyServer { + + private final int port; + private final List recordedTargets = new CopyOnWriteArrayList<>(); + private EventLoopGroup bossGroup; + private EventLoopGroup workerGroup; + + public RecordingProxyServer(final int port) { + this.port = port; + } + + public Channel start() throws InterruptedException { + bossGroup = new NioEventLoopGroup(1); + workerGroup = new NioEventLoopGroup(); + final ServerBootstrap b = new ServerBootstrap(); + b.group(bossGroup, workerGroup) + .channel(NioServerSocketChannel.class) + .childHandler(new ChannelInitializer() { + @Override + protected void initChannel(final SocketChannel ch) { + ch.pipeline().addLast("codec", new HttpServerCodec()); + ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536)); + ch.pipeline().addLast("handler", new ProxyFrontendHandler(recordedTargets)); + } + }); + return b.bind(port).sync().channel(); + } + + public void stop() { + bossGroup.shutdownGracefully(); + workerGroup.shutdownGracefully(); + } + + /** + * Handles the client-facing side of the proxy: CONNECT tunneling, absolute-URI forwarding and + * the origin-form control API. + */ + private static class ProxyFrontendHandler extends SimpleChannelInboundHandler { + + private final List recordedTargets; + + ProxyFrontendHandler(final List recordedTargets) { + this.recordedTargets = recordedTargets; + } + + @Override + protected void channelRead0(final ChannelHandlerContext ctx, final FullHttpRequest request) { + final String uri = request.uri(); + if (HttpMethod.CONNECT.equals(request.method())) { + handleConnect(ctx, uri); + } else if (uri.startsWith("/")) { + // origin-form -> control API + handleControl(ctx, request); + } else { + // absolute-form -> forward to the target origin + handleForward(ctx, request); + } + } + + // (a) CONNECT tunneling: record host:port, reply 200, then relay raw bytes. + private void handleConnect(final ChannelHandlerContext ctx, final String target) { + recordedTargets.add(target); + + final int colon = target.lastIndexOf(':'); + final String host = target.substring(0, colon); + final int targetPort = Integer.parseInt(target.substring(colon + 1)); + + final Channel clientChannel = ctx.channel(); + // Pause reads on the client until the tunnel is fully wired up. + clientChannel.config().setAutoRead(false); + + final Bootstrap b = new Bootstrap(); + b.group(clientChannel.eventLoop()) + .channel(NioSocketChannel.class) + .option(ChannelOption.AUTO_READ, false) + .handler(new ChannelInitializer() { + @Override + protected void initChannel(final Channel ch) { + ch.pipeline().addLast(new RelayHandler(clientChannel)); + } + }); + + final ChannelFuture connectFuture = b.connect(host, targetPort); + final Channel upstreamChannel = connectFuture.channel(); + connectFuture.addListener((ChannelFutureListener) future -> { + if (!future.isSuccess()) { + sendControlResponse(clientChannel, HttpResponseStatus.BAD_GATEWAY, "", "text/plain"); + closeOnFlush(clientChannel); + return; + } + final FullHttpResponse established = new DefaultFullHttpResponse( + HTTP_1_1, new HttpResponseStatus(200, "Connection Established")); + established.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0); + clientChannel.writeAndFlush(established).addListener((ChannelFutureListener) writeFuture -> { + if (!writeFuture.isSuccess()) { + closeOnFlush(upstreamChannel); + closeOnFlush(clientChannel); + return; + } + // Switch to a raw byte relay. Add the relay before removing the HTTP handlers + // (codec last) so any bytes buffered by the codec flow into the relay. + clientChannel.pipeline().addLast(new RelayHandler(upstreamChannel)); + clientChannel.pipeline().remove("handler"); + clientChannel.pipeline().remove("aggregator"); + clientChannel.pipeline().remove("codec"); + clientChannel.config().setAutoRead(true); + upstreamChannel.config().setAutoRead(true); + clientChannel.read(); + upstreamChannel.read(); + }); + }); + } + + // (b) Absolute-URI forward: record host:port, forward request, relay response back. + private void handleForward(final ChannelHandlerContext ctx, final FullHttpRequest request) { + final URI parsed; + try { + parsed = new URI(request.uri()); + } catch (Exception e) { + sendControlResponse(ctx.channel(), HttpResponseStatus.BAD_REQUEST, "", "text/plain"); + return; + } + + final String host = parsed.getHost(); + final int targetPort = parsed.getPort(); + + recordedTargets.add(host + ":" + targetPort); + + // Build an origin-form request for the upstream. Copy synchronously since the inbound + // request is released once channelRead0 returns. + String pathAndQuery = parsed.getRawPath(); + if (pathAndQuery == null || pathAndQuery.isEmpty()) { + pathAndQuery = "/"; + } + if (parsed.getRawQuery() != null) { + pathAndQuery = pathAndQuery + "?" + parsed.getRawQuery(); + } + + final FullHttpRequest forwardRequest = request.copy(); + forwardRequest.setUri(pathAndQuery); + forwardRequest.headers().set(HttpHeaderNames.HOST, host + ":" + targetPort); + + final Channel clientChannel = ctx.channel(); + final String upstreamHost = host; + final int upstreamPort = targetPort; + + final Bootstrap b = new Bootstrap(); + b.group(clientChannel.eventLoop()) + .channel(NioSocketChannel.class) + .handler(new ChannelInitializer() { + @Override + protected void initChannel(final Channel ch) { + ch.pipeline().addLast(new HttpClientCodec()); + ch.pipeline().addLast(new HttpObjectAggregator(65536)); + ch.pipeline().addLast(new UpstreamResponseHandler(clientChannel)); + } + }); + + b.connect(upstreamHost, upstreamPort).addListener((ChannelFutureListener) future -> { + if (!future.isSuccess()) { + forwardRequest.release(); + sendControlResponse(clientChannel, HttpResponseStatus.BAD_GATEWAY, "", "text/plain"); + return; + } + future.channel().writeAndFlush(forwardRequest); + }); + } + + // (c) Origin-form control API. + private void handleControl(final ChannelHandlerContext ctx, final FullHttpRequest request) { + final String uri = request.uri(); + if (HttpMethod.GET.equals(request.method()) && "/__recorded".equals(uri)) { + sendControlResponse(ctx.channel(), HttpResponseStatus.OK, toJsonArray(recordedTargets), + "application/json"); + } else if (HttpMethod.POST.equals(request.method()) && "/__reset".equals(uri)) { + recordedTargets.clear(); + sendControlResponse(ctx.channel(), HttpResponseStatus.OK, "", "text/plain"); + } else { + sendControlResponse(ctx.channel(), HttpResponseStatus.NOT_FOUND, "", "text/plain"); + } + } + + @Override + public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) { + closeOnFlush(ctx.channel()); + } + + /** + * Builds a JSON array of strings by hand to avoid pulling in a JSON dependency. + */ + private static String toJsonArray(final List targets) { + final StringBuilder sb = new StringBuilder("["); + final List snapshot = new ArrayList<>(targets); + for (int i = 0; i < snapshot.size(); i++) { + if (i > 0) { + sb.append(","); + } + sb.append("\"").append(escapeJson(snapshot.get(i))).append("\""); + } + sb.append("]"); + return sb.toString(); + } + + private static String escapeJson(final String value) { + return value.replace("\\", "\\\\").replace("\"", "\\\""); + } + } + + /** + * Writes a simple full HTTP response with the given body and content type. + */ + private static void sendControlResponse(final Channel channel, final HttpResponseStatus status, + final String body, final String contentType) { + final ByteBuf content = Unpooled.copiedBuffer(body, StandardCharsets.UTF_8); + final FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, status, content); + response.headers().set(HttpHeaderNames.CONTENT_TYPE, contentType + "; charset=UTF-8"); + HttpUtil.setContentLength(response, content.readableBytes()); + channel.writeAndFlush(response); + } + + private static void closeOnFlush(final Channel channel) { + if (channel != null && channel.isActive()) { + channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE); + } + } + + /** + * Relays raw bytes to a peer channel and mirrors close/error events so both ends of a tunnel are + * torn down together. + */ + private static class RelayHandler extends ChannelInboundHandlerAdapter { + + private final Channel relayChannel; + + RelayHandler(final Channel relayChannel) { + this.relayChannel = relayChannel; + } + + @Override + public void channelRead(final ChannelHandlerContext ctx, final Object msg) { + if (relayChannel.isActive()) { + relayChannel.writeAndFlush(msg).addListener((ChannelFutureListener) future -> { + if (future.isSuccess()) { + ctx.channel().read(); + } else { + future.channel().close(); + } + }); + } else { + ReferenceCountUtil.release(msg); + } + } + + @Override + public void channelInactive(final ChannelHandlerContext ctx) { + closeOnFlush(relayChannel); + } + + @Override + public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) { + closeOnFlush(ctx.channel()); + } + } + + /** + * Relays a forwarded HTTP response from the upstream origin back to the client, then closes the + * upstream connection. + */ + private static class UpstreamResponseHandler extends SimpleChannelInboundHandler { + + private final Channel clientChannel; + + UpstreamResponseHandler(final Channel clientChannel) { + this.clientChannel = clientChannel; + } + + @Override + protected void channelRead0(final ChannelHandlerContext ctx, final FullHttpResponse response) { + if (clientChannel.isActive()) { + clientChannel.writeAndFlush(response.retain()); + } + ctx.close(); + } + + @Override + public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) { + ctx.close(); + closeOnFlush(clientChannel); + } + } +} diff --git a/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/SocketServerConstants.java b/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/SocketServerConstants.java index 74289110208..a276a75704e 100644 --- a/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/SocketServerConstants.java +++ b/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/SocketServerConstants.java @@ -26,6 +26,8 @@ public final class SocketServerConstants { public static final int PORT = 45943; + public static final int PROXY_PORT = 45944; + public static final String GREMLIN_SINGLE_VERTEX = "server_single_vertex"; public static final String GREMLIN_CLOSE_CONNECTION = "server_close_connection"; public static final String GREMLIN_VERTEX_THEN_CLOSE = "server_vertex_then_close"; diff --git a/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/SocketServerRunner.java b/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/SocketServerRunner.java index 5f6d958cc0c..becdc3d6381 100644 --- a/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/SocketServerRunner.java +++ b/gremlin-tools/gremlin-socket-server/src/main/java/org/apache/tinkerpop/gremlin/socket/server/SocketServerRunner.java @@ -28,7 +28,11 @@ public class SocketServerRunner { public static void main(final String[] args) throws InterruptedException { final SimpleTestServer server = new SimpleTestServer(SocketServerConstants.PORT); final Channel channel = server.start(new TestHttpServerInitializer()); - while (channel.isOpen()) { + + final RecordingProxyServer proxyServer = new RecordingProxyServer(SocketServerConstants.PROXY_PORT); + final Channel proxyChannel = proxyServer.start(); + + while (channel.isOpen() || proxyChannel.isOpen()) { Thread.sleep(1000); } }