Watcher: Add support for scheme in proxy configuration (elastic/x-pack-elasticsearch#3614)

This adds support to allow different schemes in a proxy being used
compared to what the actual request requires. So if your proxy runs via
HTTP, but the endpoint you want to connect to uses HTTPS, this is now
possible to configure the proxy explicitely.

Also a small unit test for parsing this has been added.

relates elastic/x-pack-elasticsearch#3596

Original commit: elastic/x-pack-elasticsearch@176f7cdf0e

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/common/http/HttpClient.java

@@ -65,6 +65,7 @@ public class HttpClient extends AbstractComponent {
     private final CloseableHttpClient client;
     private final Integer proxyPort;
     private final String proxyHost;
+    private final String proxyScheme;
     private final TimeValue defaultConnectionTimeout;
     private final TimeValue defaultReadTimeout;
     private final ByteSizeValue maxResponseSize;
@@ -78,6 +79,7 @@ public class HttpClient extends AbstractComponent {
 
         // proxy setup
         this.proxyHost = HttpSettings.PROXY_HOST.get(settings);
+        this.proxyScheme = HttpSettings.PROXY_SCHEME.exists(settings) ? HttpSettings.PROXY_SCHEME.get(settings) : null;
         this.proxyPort = HttpSettings.PROXY_PORT.get(settings);
         if (proxyPort != 0 && Strings.hasText(proxyHost)) {
             logger.info("Using default proxy for http input and slack/hipchat/pagerduty/webhook actions [{}:{}]", proxyHost, proxyPort);
@@ -139,10 +141,14 @@ public class HttpClient extends AbstractComponent {
 
         // proxy
         if (request.proxy != null && request.proxy.equals(HttpProxy.NO_PROXY) == false) {
-            HttpHost proxy = new HttpHost(request.proxy.getHost(), request.proxy.getPort(), request.scheme.scheme());
+            // if a proxy scheme is configured use this, but fall back to the same than the request in case there was no special
+            // configuration given
+            String scheme = request.proxy.getScheme() != null ? request.proxy.getScheme().scheme() : request.scheme.scheme();
+            HttpHost proxy = new HttpHost(request.proxy.getHost(), request.proxy.getPort(), scheme);
             config.setProxy(proxy);
         } else if (proxyPort != null && Strings.hasText(proxyHost)) {
-            HttpHost proxy = new HttpHost(proxyHost, proxyPort, request.scheme.scheme());
+            String scheme = proxyScheme != null ? proxyScheme : request.scheme.scheme();
+            HttpHost proxy = new HttpHost(proxyHost, proxyPort, scheme);
             config.setProxy(proxy);
         }
 

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/common/http/HttpProxy.java

@@ -22,34 +22,37 @@ import java.net.Proxy;
 import java.net.UnknownHostException;
 import java.util.Objects;
 
-public class HttpProxy implements ToXContentFragment, Streamable {
+public class HttpProxy implements ToXContentFragment {
 
-    public static final HttpProxy NO_PROXY = new HttpProxy(null, null);
+    public static final HttpProxy NO_PROXY = new HttpProxy(null, null, null);
+
+    private static final ParseField HOST = new ParseField("host");
+    private static final ParseField PORT = new ParseField("port");
+    private static final ParseField SCHEME = new ParseField("scheme");
 
     private String host;
     private Integer port;
+    private Scheme scheme;
 
     public HttpProxy(String host, Integer port) {
         this.host = host;
         this.port = port;
     }
 
-    @Override
+    public HttpProxy(String host, Integer port, Scheme scheme) {
-    public void readFrom(StreamInput in) throws IOException {
+        this.host = host;
-        host = in.readOptionalString();
+        this.port = port;
-        port = in.readOptionalVInt();
+        this.scheme = scheme;
-    }
-
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        out.writeOptionalString(host);
-        out.writeOptionalVInt(port);
     }
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         if (Strings.hasText(host) && port != null) {
-            builder.startObject("proxy").field("host", host).field("port", port).endObject();
+            builder.startObject("proxy").field("host", host).field("port", port);
+            if (scheme != null) {
+                builder.field("scheme", scheme.scheme());
+            }
+            builder.endObject();
         }
         return builder;
     }
@@ -62,12 +65,8 @@ public class HttpProxy implements ToXContentFragment, Streamable {
         return port;
     }
 
-    public Proxy proxy() throws UnknownHostException {
+    public Scheme getScheme() {
-        if (Strings.hasText(host) && port != null) {
+        return scheme;
-            return new Proxy(Proxy.Type.HTTP, new InetSocketAddress(InetAddress.getByName(host), port));
-        }
-
-        return Proxy.NO_PROXY;
     }
 
     @Override
@@ -77,12 +76,12 @@ public class HttpProxy implements ToXContentFragment, Streamable {
 
         HttpProxy that = (HttpProxy) o;
 
-        return Objects.equals(port, that.port) && Objects.equals(host, that.host);
+        return Objects.equals(port, that.port) && Objects.equals(host, that.host) && Objects.equals(scheme, that.scheme);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(host, port);
+        return Objects.hash(host, port, scheme);
     }
 
 
@@ -91,13 +90,16 @@ public class HttpProxy implements ToXContentFragment, Streamable {
         String currentFieldName = null;
         String host = null;
         Integer port = null;
+        Scheme scheme = null;
 
         while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
             if (token == XContentParser.Token.FIELD_NAME) {
                 currentFieldName = parser.currentName();
-            } else if (Field.HOST.match(currentFieldName)) {
+            } else if (HOST.match(currentFieldName)) {
                 host = parser.text();
-            } else if (Field.PORT.match(currentFieldName)) {
+            } else if (SCHEME.match(currentFieldName)) {
+                scheme = Scheme.parse(parser.text());
+            } else if (PORT.match(currentFieldName)) {
                 port = parser.intValue();
                 if (port <= 0 || port >= 65535) {
                     throw new ElasticsearchParseException("Proxy port must be between 1 and 65534, but was " + port);
@@ -109,11 +111,6 @@ public class HttpProxy implements ToXContentFragment, Streamable {
             throw new ElasticsearchParseException("Proxy must contain 'port' and 'host' field");
         }
 
-        return new HttpProxy(host, port);
+        return new HttpProxy(host, port, scheme);
-    }
-
-    public interface Field {
-        ParseField HOST = new ParseField("host");
-        ParseField PORT = new ParseField("port");
     }
 }

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/common/http/HttpSettings.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.watcher.common.http;
 
 import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
@@ -23,22 +24,24 @@ public class HttpSettings {
     private static final TimeValue DEFAULT_CONNECTION_TIMEOUT = DEFAULT_READ_TIMEOUT;
 
     static final Setting<TimeValue> READ_TIMEOUT = Setting.timeSetting("xpack.http.default_read_timeout",
-            DEFAULT_READ_TIMEOUT, Setting.Property.NodeScope);
+            DEFAULT_READ_TIMEOUT, Property.NodeScope);
     static final Setting<TimeValue> CONNECTION_TIMEOUT = Setting.timeSetting("xpack.http.default_connection_timeout",
-            DEFAULT_CONNECTION_TIMEOUT, Setting.Property.NodeScope);
+            DEFAULT_CONNECTION_TIMEOUT, Property.NodeScope);
 
-    static final String PROXY_HOST_KEY = "xpack.http.proxy.host";
+    private static final String PROXY_HOST_KEY = "xpack.http.proxy.host";
-    static final String PROXY_PORT_KEY = "xpack.http.proxy.port";
+    private static final String PROXY_PORT_KEY = "xpack.http.proxy.port";
-    static final String SSL_KEY_PREFIX = "xpack.http.ssl.";
+    private static final String PROXY_SCHEME_KEY = "xpack.http.proxy.scheme";
+    private static final String SSL_KEY_PREFIX = "xpack.http.ssl.";
 
-    static final Setting<String> PROXY_HOST = Setting.simpleString(PROXY_HOST_KEY, Setting.Property.NodeScope);
+    static final Setting<String> PROXY_HOST = Setting.simpleString(PROXY_HOST_KEY, Property.NodeScope);
-    static final Setting<Integer> PROXY_PORT = Setting.intSetting(PROXY_PORT_KEY, 0, 0, 0xFFFF, Setting.Property.NodeScope);
+    static final Setting<String> PROXY_SCHEME = Setting.simpleString(PROXY_SCHEME_KEY, (v, s) -> Scheme.parse(v), Property.NodeScope);
+    static final Setting<Integer> PROXY_PORT = Setting.intSetting(PROXY_PORT_KEY, 0, 0, 0xFFFF, Property.NodeScope);
 
     static final Setting<ByteSizeValue> MAX_HTTP_RESPONSE_SIZE = Setting.byteSizeSetting("xpack.http.max_response_size",
             new ByteSizeValue(10, ByteSizeUnit.MB),   // default
             new ByteSizeValue(1, ByteSizeUnit.BYTES), // min
             new ByteSizeValue(50, ByteSizeUnit.MB),   // max
-            Setting.Property.NodeScope);
+            Property.NodeScope);
 
     private static final SSLConfigurationSettings SSL = SSLConfigurationSettings.withPrefix(SSL_KEY_PREFIX);
 
@@ -49,6 +52,7 @@ public class HttpSettings {
         settings.add(CONNECTION_TIMEOUT);
         settings.add(PROXY_HOST);
         settings.add(PROXY_PORT);
+        settings.add(PROXY_SCHEME);
         settings.add(MAX_HTTP_RESPONSE_SIZE);
         return settings;
     }

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/common/http/HttpClientTests.java

@@ -323,6 +323,50 @@ public class HttpClientTests extends ESTestCase {
         }
     }
 
+    public void testProxyCanHaveDifferentSchemeThanRequest() throws Exception {
+        // this test fakes a proxy server that sends a response instead of forwarding it to the mock web server
+        // on top of that the proxy request is HTTPS but the real request is HTTP only
+        MockSecureSettings serverSecureSettings = new MockSecureSettings();
+        // We can't use the client created above for the server since it is only a truststore
+        serverSecureSettings.setString("xpack.ssl.keystore.secure_password", "testnode");
+        Settings serverSettings = Settings.builder()
+                .put("xpack.ssl.keystore.path", getDataPath("/org/elasticsearch/xpack/security/keystore/testnode.jks"))
+                .setSecureSettings(serverSecureSettings)
+                .build();
+        TestsSSLService sslService = new TestsSSLService(serverSettings, environment);
+
+        try (MockWebServer proxyServer = new MockWebServer(sslService.sslContext(), false)) {
+            proxyServer.enqueue(new MockResponse().setResponseCode(200).setBody("fullProxiedContent"));
+            proxyServer.start();
+
+            Path resource = getDataPath("/org/elasticsearch/xpack/security/keystore/truststore-testnode-only.jks");
+            MockSecureSettings secureSettings = new MockSecureSettings();
+            secureSettings.setString("xpack.http.ssl.truststore.secure_password", "truststore-testnode-only");
+            Settings settings = Settings.builder()
+                    .put(HttpSettings.PROXY_HOST.getKey(), "localhost")
+                    .put(HttpSettings.PROXY_PORT.getKey(), proxyServer.getPort())
+                    .put(HttpSettings.PROXY_SCHEME.getKey(), "https")
+                    .put("xpack.http.ssl.truststore.path", resource.toString())
+                    .setSecureSettings(secureSettings)
+                    .build();
+
+            HttpClient httpClient = new HttpClient(settings, authRegistry, new SSLService(settings, environment));
+
+            HttpRequest.Builder requestBuilder = HttpRequest.builder("localhost", webServer.getPort())
+                    .method(HttpMethod.GET)
+                    .scheme(Scheme.HTTP)
+                    .path("/");
+
+            HttpResponse response = httpClient.execute(requestBuilder.build());
+            assertThat(response.status(), equalTo(200));
+            assertThat(response.body().utf8ToString(), equalTo("fullProxiedContent"));
+
+            // ensure we hit the proxyServer and not the webserver
+            assertThat(webServer.requests(), hasSize(0));
+            assertThat(proxyServer.requests(), hasSize(1));
+        }
+    }
+
     public void testThatProxyCanBeOverriddenByRequest() throws Exception {
         // this test fakes a proxy server that sends a response instead of forwarding it to the mock web server
         try (MockWebServer proxyServer = new MockWebServer()) {
@@ -331,12 +375,13 @@ public class HttpClientTests extends ESTestCase {
             Settings settings = Settings.builder()
                     .put(HttpSettings.PROXY_HOST.getKey(), "localhost")
                     .put(HttpSettings.PROXY_PORT.getKey(), proxyServer.getPort() + 1)
+                    .put(HttpSettings.PROXY_HOST.getKey(), "https")
                     .build();
             HttpClient httpClient = new HttpClient(settings, authRegistry, new SSLService(settings, environment));
 
             HttpRequest.Builder requestBuilder = HttpRequest.builder("localhost", webServer.getPort())
                     .method(HttpMethod.GET)
-                    .proxy(new HttpProxy("localhost", proxyServer.getPort()))
+                    .proxy(new HttpProxy("localhost", proxyServer.getPort(), Scheme.HTTP))
                     .path("/");
 
             HttpResponse response = httpClient.execute(requestBuilder.build());

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/common/http/HttpProxyTests.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.watcher.common.http;
+
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.test.ESTestCase;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+
+public class HttpProxyTests extends ESTestCase {
+
+    public void testParser() throws Exception {
+        int port = randomIntBetween(1, 65000);
+        String host = randomAlphaOfLength(10);
+        XContentBuilder builder = jsonBuilder().startObject().field("host", host).field("port", port);
+        boolean isSchemeConfigured = randomBoolean();
+        String scheme = null;
+        if (isSchemeConfigured) {
+            scheme = randomFrom(Scheme.values()).scheme();
+            builder.field("scheme", scheme);
+        }
+        builder.endObject();
+        try (XContentParser parser = XContentFactory.xContent(XContentType.JSON)
+                .createParser(NamedXContentRegistry.EMPTY, builder.bytes())) {
+            parser.nextToken();
+            HttpProxy proxy = HttpProxy.parse(parser);
+            assertThat(proxy.getHost(), is(host));
+            assertThat(proxy.getPort(), is(port));
+            if (isSchemeConfigured) {
+                assertThat(proxy.getScheme().scheme(), is(scheme));
+            } else {
+                assertThat(proxy.getScheme(), is(nullValue()));
+            }
+        }
+    }
+
+    public void testParserValidScheme() throws Exception {
+        XContentBuilder builder = jsonBuilder().startObject()
+                .field("host", "localhost").field("port", 12345).field("scheme", "invalid")
+                .endObject();
+        try (XContentParser parser = XContentFactory.xContent(XContentType.JSON)
+                .createParser(NamedXContentRegistry.EMPTY, builder.bytes())) {
+            parser.nextToken();
+            expectThrows(IllegalArgumentException.class, () -> HttpProxy.parse(parser));
+        }
+    }
+
+    public void testParserValidPortRange() throws Exception {
+        XContentBuilder builder = jsonBuilder().startObject()
+                .field("host", "localhost").field("port", -1)
+                .endObject();
+        try (XContentParser parser = XContentFactory.xContent(XContentType.JSON)
+                .createParser(NamedXContentRegistry.EMPTY, builder.bytes())) {
+            parser.nextToken();
+            expectThrows(ElasticsearchParseException.class, () -> HttpProxy.parse(parser));
+        }
+    }
+
+    public void testParserNoHost() throws Exception {
+        XContentBuilder builder = jsonBuilder().startObject()
+                .field("port", -1)
+                .endObject();
+        try (XContentParser parser = XContentFactory.xContent(XContentType.JSON)
+                .createParser(NamedXContentRegistry.EMPTY, builder.bytes())) {
+            parser.nextToken();
+            expectThrows(ElasticsearchParseException.class, () -> HttpProxy.parse(parser));
+        }
+    }
+
+    public void testParserNoPort() throws Exception {
+        XContentBuilder builder = jsonBuilder().startObject()
+                .field("host", "localhost")
+                .endObject();
+        try (XContentParser parser = XContentFactory.xContent(XContentType.JSON)
+                .createParser(NamedXContentRegistry.EMPTY, builder.bytes())) {
+            parser.nextToken();
+            expectThrows(ElasticsearchParseException.class, () -> HttpProxy.parse(parser));
+        }
+    }
+
+    public void testToXContent() throws Exception {
+        try (XContentBuilder builder = jsonBuilder()) {
+            builder.startObject();
+            HttpProxy proxy = new HttpProxy("localhost", 3128);
+            proxy.toXContent(builder, ToXContent.EMPTY_PARAMS);
+            builder.endObject();
+            assertThat(builder.string(), is("{\"proxy\":{\"host\":\"localhost\",\"port\":3128}}"));
+        }
+
+        try (XContentBuilder builder = jsonBuilder()) {
+            builder.startObject();
+            HttpProxy httpsProxy = new HttpProxy("localhost", 3128, Scheme.HTTPS);
+            httpsProxy.toXContent(builder, ToXContent.EMPTY_PARAMS);
+            builder.endObject();
+            assertThat(builder.string(), is("{\"proxy\":{\"host\":\"localhost\",\"port\":3128,\"scheme\":\"https\"}}"));
+        }
+    }
+}