Add ability for Watcher's webhook actions to send additional header (#…

…93426)

This commit builds on #92576 by adding the `xpack.notification.webhook.host_token_pairs` keystore setting to allow an additional token to be sent when Watcher performs a webhook request. This includes both the regular `webhook` action as well as the `email` action that uses an `attachment` parameter to a url (which internally uses a webhook to retrieve the attachment).

These settings can both by reloaded by updating the keystore and using the reload secure settings API.

- `xpack.notification.webhook.host_token_pairs` is a comma-separated list of `<host>:<port>=<token>` pairs for which the header should be sent. For example, if this contains `localhost:1234=token1,aoeu.com:9182=token2` then the token will only be added to the headers for webhook requests to the `localhost` and `aoeu.com` hosts on the 1234 and 9182 ports with tokens `token1` and `token2` respectively.

Also added is a cluster setting (non-dynamic) — `xpack.notification.webhook.additional_token_enabled` that determines whether the token should be sent. It defaults to `false`.

docs/changelog/93426.yaml

@@ -0,0 +1,5 @@
+pr: 93426
+summary: Add ability for Watcher's webhook actions to send additional header
+area: Watcher
+type: enhancement
+issues: []

server/src/main/resources/org/elasticsearch/bootstrap/test-framework.policy

@@ -126,4 +126,6 @@ grant codeBase "${codebase.netty-transport}" {
 grant {
   permission java.net.SocketPermission "127.0.0.1", "connect, resolve";
   permission java.nio.file.LinkPermission "symbolic";
+  // needed for keystore tests
+  permission java.lang.RuntimePermission "accessUserInformation";
 };

test/framework/src/main/java/org/elasticsearch/test/InternalTestCluster.java

@@ -1775,6 +1775,10 @@ private void rebuildUnicastHostFiles(List<NodeAndClient> newNodes) {
         }
     }
 
+    public Collection<Path> configPaths() {
+        return nodes.values().stream().map(nac -> nac.node.getEnvironment().configFile()).toList();
+    }
+
     private void stopNodesAndClient(NodeAndClient nodeAndClient) throws IOException {
         stopNodesAndClients(Collections.singleton(nodeAndClient));
     }

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/actions/webhook/WebhookTokenIntegrationTests.java

@@ -0,0 +1,142 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.watcher.actions.webhook;
+
+import org.elasticsearch.action.admin.cluster.node.reload.NodesReloadSecureSettingsAction;
+import org.elasticsearch.action.admin.cluster.node.reload.NodesReloadSecureSettingsRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.common.settings.KeyStoreWrapper;
+import org.elasticsearch.common.settings.MockSecureSettings;
+import org.elasticsearch.common.settings.SecureString;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.http.MockRequest;
+import org.elasticsearch.test.http.MockResponse;
+import org.elasticsearch.test.http.MockWebServer;
+import org.elasticsearch.transport.netty4.Netty4Plugin;
+import org.elasticsearch.xpack.core.watcher.history.WatchRecord;
+import org.elasticsearch.xpack.core.watcher.support.xcontent.XContentSource;
+import org.elasticsearch.xpack.core.watcher.transport.actions.put.PutWatchRequestBuilder;
+import org.elasticsearch.xpack.watcher.actions.ActionBuilders;
+import org.elasticsearch.xpack.watcher.common.http.BasicAuth;
+import org.elasticsearch.xpack.watcher.common.http.HttpMethod;
+import org.elasticsearch.xpack.watcher.common.http.HttpRequestTemplate;
+import org.elasticsearch.xpack.watcher.common.text.TextTemplate;
+import org.elasticsearch.xpack.watcher.condition.InternalAlwaysCondition;
+import org.elasticsearch.xpack.watcher.notification.WebhookService;
+import org.elasticsearch.xpack.watcher.test.AbstractWatcherIntegrationTestCase;
+import org.junit.After;
+import org.junit.Before;
+
+import java.nio.file.Path;
+import java.util.Collection;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
+import static org.elasticsearch.xpack.watcher.client.WatchSourceBuilders.watchBuilder;
+import static org.elasticsearch.xpack.watcher.input.InputBuilders.simpleInput;
+import static org.elasticsearch.xpack.watcher.test.WatcherTestUtils.xContentSource;
+import static org.elasticsearch.xpack.watcher.trigger.TriggerBuilders.schedule;
+import static org.elasticsearch.xpack.watcher.trigger.schedule.Schedules.interval;
+import static org.hamcrest.Matchers.anyOf;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+
+public class WebhookTokenIntegrationTests extends AbstractWatcherIntegrationTestCase {
+
+    private MockWebServer webServer = new MockWebServer();
+
+    @Override
+    protected boolean addMockHttpTransport() {
+        return false; // enable http
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return CollectionUtils.appendToCopy(super.nodePlugins(), Netty4Plugin.class); // for http
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        Settings.Builder builder = Settings.builder();
+        builder.put(super.nodeSettings(nodeOrdinal, otherSettings));
+        builder.put(WebhookService.SETTING_WEBHOOK_TOKEN_ENABLED.getKey(), true);
+        MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString(WebhookService.SETTING_WEBHOOK_HOST_TOKEN_PAIRS.getKey(), "localhost:0=oldtoken");
+        builder.setSecureSettings(secureSettings);
+        return builder.build();
+    }
+
+    @Before
+    public void startWebservice() throws Exception {
+        webServer.start();
+    }
+
+    @After
+    public void stopWebservice() throws Exception {
+        webServer.close();
+    }
+
+    public void testWebhook() throws Exception {
+        String localServer = "localhost:" + webServer.getPort();
+        logger.info("--> updating keystore token hosts to: {}", localServer);
+        Path configPath = internalCluster().configPaths().stream().findFirst().orElseThrow();
+        try (KeyStoreWrapper ksw = KeyStoreWrapper.bootstrap(configPath, () -> new SecureString("".toCharArray()))) {
+            ksw.setString(WebhookService.SETTING_WEBHOOK_HOST_TOKEN_PAIRS.getKey(), (localServer + "=token1234").toCharArray());
+            ksw.save(configPath, "".toCharArray());
+        }
+        // Reload the keystore to load the new settings
+        NodesReloadSecureSettingsRequest reloadReq = new NodesReloadSecureSettingsRequest();
+        reloadReq.setSecureStorePassword(new SecureString("".toCharArray()));
+        client().execute(NodesReloadSecureSettingsAction.INSTANCE, reloadReq).get();
+
+        webServer.enqueue(new MockResponse().setResponseCode(200).setBody("body"));
+        HttpRequestTemplate.Builder builder = HttpRequestTemplate.builder("localhost", webServer.getPort())
+            .path(new TextTemplate("/test/_id"))
+            .putParam("param1", new TextTemplate("value1"))
+            .putParam("watch_id", new TextTemplate("_id"))
+            .body(new TextTemplate("_body"))
+            .auth(new BasicAuth("user", "pass".toCharArray()))
+            .method(HttpMethod.POST);
+
+        new PutWatchRequestBuilder(client(), "_id").setSource(
+            watchBuilder().trigger(schedule(interval("5s")))
+                .input(simpleInput("key", "value"))
+                .condition(InternalAlwaysCondition.INSTANCE)
+                .addAction("_id", ActionBuilders.webhookAction(builder))
+        ).get();
+
+        timeWarp().trigger("_id");
+        refresh();
+
+        assertWatchWithMinimumPerformedActionsCount("_id", 1, false);
+        assertThat(webServer.requests(), hasSize(1));
+        MockRequest req = webServer.requests().get(0);
+        assertThat(
+            webServer.requests().get(0).getUri().getQuery(),
+            anyOf(equalTo("watch_id=_id&param1=value1"), equalTo("param1=value1&watch_id=_id"))
+        );
+        assertThat("token header should be set", req.getHeader(WebhookService.TOKEN_HEADER_NAME), equalTo("token1234"));
+
+        assertThat(webServer.requests().get(0).getBody(), is("_body"));
+
+        SearchResponse response = searchWatchRecords(b -> QueryBuilders.termQuery(WatchRecord.STATE.getPreferredName(), "executed"));
+
+        assertNoFailures(response);
+        XContentSource source = xContentSource(response.getHits().getAt(0).getSourceRef());
+        String body = source.getValue("result.actions.0.webhook.response.body");
+        assertThat(body, notNullValue());
+        assertThat(body, is("body"));
+        Number status = source.getValue("result.actions.0.webhook.response.status");
+        assertThat(status, notNullValue());
+        assertThat(status.intValue(), is(200));
+    }
+}

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/Watcher.java

@@ -598,6 +598,7 @@ public List<Setting<?>> getSettings() {
         settings.addAll(JiraService.getSettings());
         settings.addAll(PagerDutyService.getSettings());
         settings.addAll(ReportingAttachmentParser.getSettings());
+        settings.addAll(WebhookService.getSettings());
 
         // http settings
         settings.addAll(HttpSettings.getSettings());

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/common/http/HttpRequest.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.xpack.core.watcher.support.WatcherUtils;
 import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherXContentParser;
+import org.elasticsearch.xpack.watcher.notification.WebhookService;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -166,7 +167,7 @@ public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params toX
             if (WatcherParams.hideSecrets(toXContentParams)) {
                 builder.field(Field.HEADERS.getPreferredName(), sanitizeHeaders(headers));
             } else {
-                builder.field(Field.HEADERS.getPreferredName(), headers);
+                builder.field(Field.HEADERS.getPreferredName(), sanitizeInternalHeaders(headers));
             }
         }
         if (auth != null) {
@@ -195,6 +196,10 @@ public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params toX
         return builder.endObject();
     }
 
+    /**
+     * Sanitize both internal (see {@link #sanitizeInternalHeaders(Map)} and
+     * user-added sensitive headers that should not be shown.
+     */
     private Map<String, String> sanitizeHeaders(Map<String, String> headers) {
         String authorizationHeader = headers.containsKey("Authorization") ? "Authorization" : null;
         if (authorizationHeader == null) {
@@ -205,7 +210,22 @@ private Map<String, String> sanitizeHeaders(Map<String, String> headers) {
         }
         Map<String, String> sanitizedHeaders = new HashMap<>(headers);
         sanitizedHeaders.put(authorizationHeader, WatcherXContentParser.REDACTED_PASSWORD);
-        return sanitizedHeaders;
+        return sanitizeInternalHeaders(sanitizedHeaders);
+    }
+
+    /**
+     * Sanitize headers that the user may not have added, but were automatically
+     * added by Elasticsearch.
+     */
+    private Map<String, String> sanitizeInternalHeaders(Map<String, String> headers) {
+        // Redact the additional webhook password, if present.
+        if (headers.containsKey(WebhookService.TOKEN_HEADER_NAME)) {
+            Map<String, String> sanitizedHeaders = new HashMap<>(headers);
+            sanitizedHeaders.put(WebhookService.TOKEN_HEADER_NAME, WatcherXContentParser.REDACTED_PASSWORD);
+            return sanitizedHeaders;
+        } else {
+            return headers;
+        }
     }
 
     @Override
@@ -268,6 +288,13 @@ public static Builder builder(String host, int port) {
         return new Builder(host, port);
     }
 
+    /**
+     * Create a new builder modeled on this HttpRequest
+     */
+    public Builder copy() {
+        return new Builder(this);
+    }
+
     static Builder builder() {
         return new Builder();
     }
@@ -402,6 +429,21 @@ private Builder(String host, int port) {
             this.port = port;
         }
 
+        private Builder(HttpRequest original) {
+            this.host = original.host;
+            this.port = original.port;
+            this.scheme = original.scheme;
+            this.method = original.method;
+            this.path = original.path;
+            this.params = new HashMap<>(original.params);
+            this.headers = new HashMap<>(original.headers);
+            this.auth = original.auth;
+            this.body = original.body;
+            this.connectionTimeout = original.connectionTimeout;
+            this.readTimeout = original.readTimeout;
+            this.proxy = original.proxy;
+        }
+
         private Builder() {}
 
         public Builder scheme(Scheme scheme) {

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/notification/NotificationService.java

@@ -142,19 +142,20 @@ private Set<String> getAccountNames(Settings settings) {
         return settings.getByPrefix(getNotificationsAccountPrefix()).names();
     }
 
-    private @Nullable String getDefaultAccountName(Settings settings) {
+    @Nullable
+    protected String getDefaultAccountName(Settings settings) {
         return settings.get("xpack.notification." + type + ".default_account");
     }
 
-    private Map<String, LazyInitializable<Account, SettingsException>> createAccounts(
+    protected Map<String, LazyInitializable<Account, SettingsException>> createAccounts(
         Settings settings,
         Set<String> accountNames,
         BiFunction<String, Settings, Account> accountFactory
     ) {
         final Map<String, LazyInitializable<Account, SettingsException>> accounts = new HashMap<>();
         for (final String accountName : accountNames) {
             final Settings accountSettings = settings.getAsSettings(getNotificationsAccountPrefix() + accountName);
-            accounts.put(accountName, new LazyInitializable<>(() -> { return accountFactory.apply(accountName, accountSettings); }));
+            accounts.put(accountName, new LazyInitializable<>(() -> accountFactory.apply(accountName, accountSettings)));
         }
         return Collections.unmodifiableMap(accounts);
     }