Watcher: Never return credentials after watch creation... (elastic/x-pack-elasticsearch#3581)

... yet support updates. This commit introduces a few changes of how
watches are put.

The GET Watch API will never return credentials like basic auth
passwords, but a placeholder instead now. If the watcher is enabled to
encrypt sensitive settings, then the original encrypted value is
returned otherwise a "::es_redacted::" place holder.

There have been several Put Watch API changes.

The API now internally uses the Update API and versioning. This has
several implications. First if no version is supplied, we assume an
initial creation. This will work as before, however if a credential is
marked as redacted we will reject storing the watch, so users do not
accidentally store the wrong watch.

The watch xcontent parser now has an additional methods to tell the
caller if redacted passwords have been found. Based on this information
an error can be thrown.

If the user now wants to store a watch that contains a password marked
as redacted, this password will not be part of the toXContent
representation of the watch and in combinatination with update request
the existing password will be merged in. If the encrypted password is
supplied this one will be stored.

The serialization for GetWatchResponse/PutWatchRequest has changed.
The version checks for this will be put into the 6.x branch.

The Watcher UI now needs specify the version, when it wants to store a
watch. This also prevents last-write-wins scenarios and is the reason
why the put/get watch response now contains the internal version.

relates elastic/x-pack-elasticsearch#3089

Original commit: elastic/x-pack-elasticsearch@bb63be9f79

docs/en/rest-api/watcher/ack-watch.asciidoc

@@ -86,6 +86,7 @@ The action state of a newly-created watch is `awaits_successful_execution`:
 --------------------------------------------------
 {
   "found": true,
+  "_version": 1,
   "_id": "my_watch",
   "status": {
     "version": 1,
@@ -129,6 +130,7 @@ and the action is now in `ackable` state:
 {
   "found": true,
   "_id": "my_watch",
+  "_version": 2,
   "status": {
     "version": 2,
     "actions": {
@@ -177,6 +179,7 @@ GET _xpack/watcher/watch/my_watch
 {
   "found": true,
   "_id": "my_watch",
+  "_version": 3,
   "status": {
     "version": 3,
     "actions": {

docs/en/rest-api/watcher/activate-watch.asciidoc

@@ -41,6 +41,7 @@ GET _xpack/watcher/watch/my_watch
 {
   "found": true,
   "_id": "my_watch",
+  "_version": 1,
   "status": {
     "state" : {
       "active" : false,

docs/en/rest-api/watcher/deactivate-watch.asciidoc

@@ -40,6 +40,7 @@ GET _xpack/watcher/watch/my_watch
 {
   "found": true,
   "_id": "my_watch",
+  "_version": 1,
   "status": {
     "state" : {
       "active" : true,

docs/en/rest-api/watcher/get-watch.asciidoc

@@ -41,6 +41,7 @@ Response:
 {
   "found": true,
   "_id": "my_watch",
+  "_version": 1,
   "status": { <1>
     "version": 1,
     "state": {

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/client/WatchSourceBuilder.java

@@ -21,6 +21,7 @@ import org.elasticsearch.xpack.core.watcher.condition.Condition;
 import org.elasticsearch.xpack.core.watcher.input.Input;
 import org.elasticsearch.xpack.core.watcher.input.none.NoneInput;
 import org.elasticsearch.xpack.core.watcher.support.Exceptions;
+import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.core.watcher.support.xcontent.XContentSource;
 import org.elasticsearch.xpack.core.watcher.transform.Transform;
 import org.elasticsearch.xpack.core.watcher.trigger.Trigger;
@@ -175,7 +176,8 @@ public class WatchSourceBuilder implements ToXContentObject {
      */
     public final BytesReference buildAsBytes(XContentType contentType) {
         try {
-            return XContentHelper.toXContent(this, contentType, false);
+            WatcherParams params = WatcherParams.builder().hideSecrets(false).build();
+            return XContentHelper.toXContent(this, contentType,  params,false);
         } catch (Exception e) {
             throw new ElasticsearchException("Failed to build ToXContent", e);
         }

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/crypto/CryptoService.java

@@ -39,7 +39,7 @@ public class CryptoService extends AbstractComponent {
     public static final String KEY_ALGO = "HmacSHA512";
     public static final int KEY_SIZE = 1024;
 
-    static final String ENCRYPTED_TEXT_PREFIX = "::es_encrypted::";
+    public static final String ENCRYPTED_TEXT_PREFIX = "::es_encrypted::";
 
     // the encryption used in this class was picked when Java 7 was still the min. supported
     // version. The use of counter mode was chosen to simplify the need to deal with padding

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/support/xcontent/WatcherParams.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.core.watcher.support.xcontent;
 
 import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.xpack.core.watcher.watch.Watch;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -17,9 +18,9 @@ public class WatcherParams extends ToXContent.DelegatingMapParams {
 
     public static final WatcherParams HIDE_SECRETS = WatcherParams.builder().hideSecrets(true).build();
 
-    static final String HIDE_SECRETS_KEY = "hide_secrets";
+    private static final String HIDE_SECRETS_KEY = "hide_secrets";
-    public static final String HIDE_HEADERS = "hide_headers";
+    private static final String HIDE_HEADERS = "hide_headers";
-    static final String DEBUG_KEY = "debug";
+    private static final String DEBUG_KEY = "debug";
 
     public static boolean hideSecrets(ToXContent.Params params) {
         return wrap(params).hideSecrets();
@@ -29,19 +30,23 @@ public class WatcherParams extends ToXContent.DelegatingMapParams {
         return wrap(params).debug();
     }
 
+    public static boolean hideHeaders(ToXContent.Params params) {
+        return wrap(params).hideHeaders();
+    }
+
     private WatcherParams(Map<String, String> params, ToXContent.Params delegate) {
         super(params, delegate);
     }
 
-    public boolean hideSecrets() {
+    private boolean hideSecrets() {
-        return paramAsBoolean(HIDE_SECRETS_KEY, false);
+        return paramAsBoolean(HIDE_SECRETS_KEY, true);
     }
 
-    public boolean debug() {
+    private boolean debug() {
         return paramAsBoolean(DEBUG_KEY, false);
     }
 
-    public boolean hideHeaders() {
+    private boolean hideHeaders() {
         return paramAsBoolean(HIDE_HEADERS, true);
     }
 
@@ -83,6 +88,11 @@ public class WatcherParams extends ToXContent.DelegatingMapParams {
             return this;
         }
 
+        public Builder includeStatus(boolean includeStatus) {
+            params.put(Watch.INCLUDE_STATUS_KEY, String.valueOf(includeStatus));
+            return this;
+        }
+
         public Builder put(String key, Object value) {
             params.put(key, String.valueOf(value));
             return this;

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/support/xcontent/WatcherXContentParser.java

@@ -7,6 +7,7 @@ package org.elasticsearch.xpack.core.watcher.support.xcontent;
 
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.xcontent.DeprecationHandler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
@@ -33,41 +34,47 @@ import java.util.Map;
  */
 public class WatcherXContentParser implements XContentParser {
 
-    public static Secret secret(XContentParser parser) throws IOException {
+    public static final String REDACTED_PASSWORD = "::es_redacted::";
-        char[] chars = parser.text().toCharArray();
-        if (parser instanceof WatcherXContentParser) {
-            WatcherXContentParser watcherParser = (WatcherXContentParser) parser;
-            if (watcherParser.cryptoService != null) {
-                chars = watcherParser.cryptoService.encrypt(chars);
-            }
-        }
-        return new Secret(chars);
-    }
 
     public static Secret secretOrNull(XContentParser parser) throws IOException {
         String text = parser.textOrNull();
         if (text == null) {
             return null;
         }
-        char[] chars = parser.text().toCharArray();
+
-        if (parser instanceof WatcherXContentParser) {
+        char[] chars = text.toCharArray();
-            WatcherXContentParser watcherParser = (WatcherXContentParser) parser;
+        boolean isEncryptedAlready = text.startsWith(CryptoService.ENCRYPTED_TEXT_PREFIX);
-            if (watcherParser.cryptoService != null) {
+        if (isEncryptedAlready) {
-                chars = watcherParser.cryptoService.encrypt(text.toCharArray());
-            }
             return new Secret(chars);
         }
+
+        if (parser instanceof WatcherXContentParser) {
+            WatcherXContentParser watcherParser = (WatcherXContentParser) parser;
+            if (REDACTED_PASSWORD.equals(text)) {
+                if (watcherParser.allowRedactedPasswords) {
+                    return null;
+                } else {
+                    throw new ElasticsearchParseException("found redacted password in field [{}]", parser.currentName());
+                }
+            } else if (watcherParser.cryptoService != null) {
+                return new Secret(watcherParser.cryptoService.encrypt(chars));
+            }
+        }
+
         return new Secret(chars);
     }
 
     private final DateTime parseTime;
     private final XContentParser parser;
     @Nullable private final CryptoService cryptoService;
+    private final boolean allowRedactedPasswords;
 
-    public WatcherXContentParser(XContentParser parser, DateTime parseTime, @Nullable CryptoService cryptoService) {
+    public WatcherXContentParser(XContentParser parser, DateTime parseTime, @Nullable CryptoService cryptoService,
+                                 boolean allowRedactedPasswords) {
         this.parseTime = parseTime;
         this.parser = parser;
         this.cryptoService = cryptoService;
+        this.allowRedactedPasswords = allowRedactedPasswords;
     }
 
     public DateTime getParseDateTime() { return parseTime; }

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/transport/actions/get/GetWatchResponse.java

@@ -5,10 +5,12 @@
  */
 package org.elasticsearch.xpack.core.watcher.transport.actions.get;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.xpack.core.watcher.support.xcontent.XContentSource;
 import org.elasticsearch.xpack.core.watcher.watch.WatchStatus;
@@ -21,6 +23,7 @@ public class GetWatchResponse extends ActionResponse {
     private WatchStatus status;
     private boolean found = false;
     private XContentSource source;
+    private long version;
 
     public GetWatchResponse() {
     }
@@ -32,16 +35,18 @@ public class GetWatchResponse extends ActionResponse {
         this.id = id;
         this.found = false;
         this.source = null;
+        version = Versions.NOT_FOUND;
     }
 
     /**
      * ctor for found watch
      */
-    public GetWatchResponse(String id, WatchStatus status, BytesReference source, XContentType contentType) {
+    public GetWatchResponse(String id, long version, WatchStatus status, BytesReference source, XContentType contentType) {
         this.id = id;
         this.status = status;
         this.found = true;
         this.source = new XContentSource(source, contentType);
+        this.version = version;
     }
 
     public String getId() {
@@ -60,6 +65,10 @@ public class GetWatchResponse extends ActionResponse {
         return source;
     }
 
+    public long getVersion() {
+        return version;
+    }
+
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
@@ -68,6 +77,7 @@ public class GetWatchResponse extends ActionResponse {
         if (found) {
             status = WatchStatus.read(in);
             source = XContentSource.readFrom(in);
+            version = in.readZLong();
         }
     }
 
@@ -79,6 +89,7 @@ public class GetWatchResponse extends ActionResponse {
         if (found) {
             status.writeTo(out);
             XContentSource.writeTo(source, out);
+            out.writeZLong(version);
         }
     }
 }

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/transport/actions/put/PutWatchRequest.java

@@ -12,6 +12,7 @@ import org.elasticsearch.action.ValidateActions;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.xpack.core.watcher.client.WatchSourceBuilder;
 import org.elasticsearch.xpack.core.watcher.support.WatcherUtils;
@@ -28,14 +29,11 @@ public class PutWatchRequest extends ActionRequest {
     private BytesReference source;
     private boolean active = true;
     private XContentType xContentType = XContentType.JSON;
+    private long version = Versions.MATCH_ANY;
 
     public PutWatchRequest() {
     }
 
-    public PutWatchRequest(String id, WatchSourceBuilder source) {
-        this(id, source.buildAsBytes(XContentType.JSON), XContentType.JSON);
-    }
-
     public PutWatchRequest(String id, BytesReference source, XContentType xContentType) {
         this.id = id;
         this.source = source;
@@ -48,6 +46,7 @@ public class PutWatchRequest extends ActionRequest {
         source = in.readBytesReference();
         active = in.readBoolean();
         xContentType = XContentType.readFrom(in);
+        version = in.readZLong();
     }
 
     @Override
@@ -57,6 +56,7 @@ public class PutWatchRequest extends ActionRequest {
         out.writeBytesReference(source);
         out.writeBoolean(active);
         xContentType.writeTo(out);
+        out.writeZLong(version);
     }
 
     /**
@@ -116,6 +116,14 @@ public class PutWatchRequest extends ActionRequest {
         return xContentType;
     }
 
+    public long getVersion() {
+        return version;
+    }
+
+    public void setVersion(long version) {
+        this.version = version;
+    }
+
     @Override
     public ActionRequestValidationException validate() {
         ActionRequestValidationException validationException = null;

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/transport/actions/put/PutWatchRequestBuilder.java

@@ -54,4 +54,13 @@ public class PutWatchRequestBuilder extends ActionRequestBuilder<PutWatchRequest
         request.setActive(active);
         return this;
     }
+
+    /**
+     * @param version Sets the version to be set when running the update
+     */
+    public PutWatchRequestBuilder setVersion(long version) {
+        request.setVersion(version);
+        return this;
+    }
+
 }

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/watch/Watch.java

@@ -6,7 +6,6 @@
 package org.elasticsearch.xpack.core.watcher.watch;
 
 import org.elasticsearch.common.Nullable;
-import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -38,11 +37,11 @@ public class Watch implements ToXContentObject {
     @Nullable private final Map<String, Object> metadata;
     private final WatchStatus status;
 
-    private transient long version = Versions.MATCH_ANY;
+    private transient long version;
 
     public Watch(String id, Trigger trigger, ExecutableInput input, ExecutableCondition condition, @Nullable ExecutableTransform transform,
                  @Nullable TimeValue throttlePeriod, List<ActionWrapper> actions, @Nullable Map<String, Object> metadata,
-                 WatchStatus status) {
+                 WatchStatus status, long version) {
         this.id = id;
         this.trigger = trigger;
         this.input = input;
@@ -52,6 +51,7 @@ public class Watch implements ToXContentObject {
         this.throttlePeriod = throttlePeriod;
         this.metadata = metadata;
         this.status = status;
+        this.version = version;
     }
 
     public String id() {

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/watch/WatchField.java

@@ -17,6 +17,7 @@ public final class WatchField {
     public static final ParseField THROTTLE_PERIOD_HUMAN = new ParseField("throttle_period");
     public static final ParseField METADATA = new ParseField("metadata");
     public static final ParseField STATUS = new ParseField("status");
+    public static final ParseField VERSION = new ParseField("_version");
     public static final String ALL_ACTIONS_ID = "_all";
 
     private WatchField() {}

plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/watch/WatchStatus.java

@@ -265,7 +265,7 @@ public class WatchStatus implements ToXContentObject, Streamable {
         if (executionState != null) {
             builder.field(Field.EXECUTION_STATE.getPreferredName(), executionState.id());
         }
-        if (headers != null && headers.isEmpty() == false && params.paramAsBoolean(WatcherParams.HIDE_HEADERS, true) == false) {
+        if (headers != null && headers.isEmpty() == false && WatcherParams.hideHeaders(params) == false) {
             builder.field(Field.HEADERS.getPreferredName(), headers);
         }
         builder.field(Field.VERSION.getPreferredName(), version);

plugin/core/src/test/java/org/elasticsearch/xpack/watcher/support/xcontent/WatcherXContentParserTests.java

@@ -0,0 +1,46 @@
+/*
+ * 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.support.xcontent;
+
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherXContentParser;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+
+import java.time.Clock;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.hamcrest.Matchers.is;
+import static org.joda.time.DateTimeZone.UTC;
+
+public class WatcherXContentParserTests extends ESTestCase {
+
+    public void testThatRedactedSecretsThrowException() throws Exception {
+        String fieldName = randomAlphaOfLength(10);
+        try (XContentBuilder builder = jsonBuilder()) {
+            builder.startObject().field(fieldName, "::es_redacted::").endObject();
+
+            try (XContentParser xContentParser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
+                    LoggingDeprecationHandler.INSTANCE, builder.string())) {
+                xContentParser.nextToken();
+                xContentParser.nextToken();
+                assertThat(xContentParser.currentName(), is(fieldName));
+                xContentParser.nextToken();
+                assertThat(xContentParser.currentToken(), is(XContentParser.Token.VALUE_STRING));
+                WatcherXContentParser parser = new WatcherXContentParser(xContentParser, DateTime.now(UTC), null, false);
+                ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class,
+                        () -> WatcherXContentParser.secretOrNull(parser));
+                assertThat(e.getMessage(), is("found redacted password in field [" + fieldName + "]"));
+            }
+        }
+    }
+}

plugin/src/test/resources/rest-api-spec/api/xpack.watcher.put_watch.json

@@ -16,6 +16,10 @@
         "active": {
           "type": "boolean",
           "description": "Specify whether the watch is in/active by default"
+        },
+        "version" : {
+          "type" : "number",
+          "description" : "Explicit version number for concurrency control"
         }
       }
     },

plugin/src/test/resources/rest-api-spec/test/watcher/put_watch/80_put_get_watch_with_passwords.yml

@@ -0,0 +1,317 @@
+---
+setup:
+  - do:
+      cluster.health:
+          wait_for_status: yellow
+
+---
+"Test getting a watch does not contain the original password":
+
+  - do:
+      xpack.watcher.put_watch:
+        id: "watch_with_password"
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "http" : {
+                "request" : {
+                  "host" : "host.domain",
+                  "port" : 9200,
+                  "path" : "/myservice",
+                  "auth" : {
+                    "basic" : {
+                      "username" : "user",
+                      "password" : "pass"
+                    }
+                  }
+                }
+              }
+            },
+            "actions": {
+              "logging": {
+                "logging": {
+                  "text": "Log me Amadeus!"
+                }
+              }
+            }
+          }
+
+  - do:
+      xpack.watcher.get_watch:
+        id: "watch_with_password"
+  - match: { _id: "watch_with_password" }
+  - match: { watch.input.http.request.auth.basic.password: "::es_redacted::" }
+
+---
+"Test putting a watch with a redacted password without version returns an error":
+
+  # version 1
+  - do:
+      xpack.watcher.put_watch:
+        id: "watch_without_version_test"
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "http" : {
+                "request" : {
+                  "host" : "host.domain",
+                  "port" : 9200,
+                  "path" : "/myservice",
+                  "auth" : {
+                    "basic" : {
+                      "username" : "user",
+                      "password" : "pass"
+                    }
+                  }
+                }
+              }
+            },
+            "actions": {
+              "logging": {
+                "logging": {
+                  "text": "Log me Amadeus!"
+                }
+              }
+            }
+          }
+
+  - do:
+      catch: bad_request
+      xpack.watcher.put_watch:
+        id: "watch_without_version_test"
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "http" : {
+                "request" : {
+                  "host" : "host.domain",
+                  "port" : 9200,
+                  "path" : "/myservice",
+                  "auth" : {
+                    "basic" : {
+                      "username" : "user",
+                      "password" : "::es_redacted::"
+                    }
+                  }
+                }
+              }
+            },
+            "actions": {
+              "logging": {
+                "logging": {
+                  "text": "Log me Amadeus!"
+                }
+              }
+            }
+          }
+
+
+---
+"Test putting a watch with a redacted password with old version returns an error":
+
+  # version 1
+  - do:
+      xpack.watcher.put_watch:
+        id: "watch_old_version"
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "http" : {
+                "request" : {
+                  "host" : "host.domain",
+                  "port" : 9200,
+                  "path" : "/myservice",
+                  "auth" : {
+                    "basic" : {
+                      "username" : "user",
+                      "password" : "pass"
+                    }
+                  }
+                }
+              }
+            },
+            "actions": {
+              "logging": {
+                "logging": {
+                  "text": "Log me Amadeus!"
+                }
+              }
+            }
+          }
+
+  # version 2
+  - do:
+      xpack.watcher.put_watch:
+        id: "watch_old_version"
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "http" : {
+                "request" : {
+                  "host" : "host.domain",
+                  "port" : 9200,
+                  "path" : "/myservice",
+                  "auth" : {
+                    "basic" : {
+                      "username" : "user",
+                      "password" : "pass"
+                    }
+                  }
+                }
+              }
+            },
+            "actions": {
+              "logging": {
+                "logging": {
+                  "text": "Log me Amadeus!"
+                }
+              }
+            }
+          }
+
+
+  # using optimistic concurrency control, this one will loose
+  # as if two users in the watch UI tried to update the same watch
+  - do:
+      catch: conflict
+      xpack.watcher.put_watch:
+        id: "watch_old_version"
+        version: 1
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "http" : {
+                "request" : {
+                  "host" : "host.domain",
+                  "port" : 9200,
+                  "path" : "/myservice",
+                  "auth" : {
+                    "basic" : {
+                      "username" : "user",
+                      "password" : "::es_redacted::"
+                    }
+                  }
+                }
+              }
+            },
+            "actions": {
+              "logging": {
+                "logging": {
+                  "text": "Log me Amadeus!"
+                }
+              }
+            }
+          }
+
+---
+"Test putting a watch with a redacted password with current version works":
+
+  - do:
+      xpack.watcher.put_watch:
+        id: "my_watch_with_version"
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "http" : {
+                "request" : {
+                  "host" : "host.domain",
+                  "port" : 9200,
+                  "path" : "/myservice",
+                  "auth" : {
+                    "basic" : {
+                      "username" : "user",
+                      "password" : "pass"
+                    }
+                  }
+                }
+              }
+            },
+            "actions": {
+              "logging": {
+                "logging": {
+                  "text": "Log me Amadeus!"
+                }
+              }
+            }
+          }
+
+  - match: { _id: "my_watch_with_version" }
+  - match: { _version: 1 }
+
+  # this resembles the exact update from the UI and thus should work, no password change, any change in the watch
+  # but correct version provided
+  - do:
+      xpack.watcher.put_watch:
+        id: "my_watch_with_version"
+        version: 1
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "http" : {
+                "request" : {
+                  "host" : "host.domain",
+                  "port" : 9200,
+                  "path" : "/myservice",
+                  "auth" : {
+                    "basic" : {
+                      "username" : "user",
+                      "password" : "::es_redacted::"
+                    }
+                  }
+                }
+              }
+            },
+            "actions": {
+              "logging": {
+                "logging": {
+                  "text": "Log me Amadeus!"
+                }
+              }
+            }
+          }
+
+  - match: { _id: "my_watch_with_version" }
+  - match: { _version: 2 }
+
+  - do:
+      search:
+        index: .watches
+        body:  >
+          {
+            "query": {
+              "term": {
+                "_id": {
+                  "value": "my_watch_with_version"
+                }
+              }
+            }
+          }
+
+  - match: { hits.total: 1 }
+  - match: { hits.hits.0._id: "my_watch_with_version" }
+  - match: { hits.hits.0._source.input.http.request.auth.basic.password: "pass" }
+

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/actions/email/EmailAction.java

@@ -12,6 +12,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.watcher.actions.Action;
 import org.elasticsearch.xpack.core.watcher.common.secret.Secret;
+import org.elasticsearch.xpack.core.watcher.crypto.CryptoService;
 import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherXContentParser;
 import org.elasticsearch.xpack.watcher.notification.email.Authentication;
@@ -104,7 +105,9 @@ public class EmailAction implements Action {
         }
         if (auth != null) {
             builder.field(Field.USER.getPreferredName(), auth.user());
-            if (WatcherParams.hideSecrets(params) == false) {
+            if (WatcherParams.hideSecrets(params) && auth.password().value().startsWith(CryptoService.ENCRYPTED_TEXT_PREFIX) == false) {
+                builder.field(Field.PASSWORD.getPreferredName(), WatcherXContentParser.REDACTED_PASSWORD);
+            } else {
                 builder.field(Field.PASSWORD.getPreferredName(), auth.password().value());
             }
         }

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/common/http/auth/basic/BasicAuth.java

@@ -10,11 +10,13 @@ import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.watcher.common.secret.Secret;
+import org.elasticsearch.xpack.core.watcher.crypto.CryptoService;
 import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherXContentParser;
 import org.elasticsearch.xpack.watcher.common.http.auth.HttpAuth;
 
 import java.io.IOException;
+import java.util.Objects;
 
 public class BasicAuth implements HttpAuth {
 
@@ -46,26 +48,29 @@ public class BasicAuth implements HttpAuth {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
 
-        BasicAuth basicAuth = (BasicAuth) o;
+        BasicAuth other = (BasicAuth) o;
 
-        if (!username.equals(basicAuth.username)) return false;
+        return Objects.equals(username, other.username) && Objects.equals(password, other.password);
-        return password.equals(basicAuth.password);
     }
 
     @Override
     public int hashCode() {
-        int result = username.hashCode();
+        return Objects.hash(username, password);
-        result = 31 * result + password.hashCode();
-        return result;
     }
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         builder.field(Field.USERNAME.getPreferredName(), username);
-        if (!WatcherParams.hideSecrets(params)) {
+        // if the password is null, do not render it out, so we have the possibility to call toXContent when we want to update a watch
+        // if the password is not null, ensure we never return the original password value, unless it is encrypted with the CryptoService
+        if (password != null) {
+            if (WatcherParams.hideSecrets(params) && password.value().startsWith(CryptoService.ENCRYPTED_TEXT_PREFIX) == false) {
+                builder.field(Field.PASSWORD.getPreferredName(), WatcherXContentParser.REDACTED_PASSWORD);
+            } else {
                 builder.field(Field.PASSWORD.getPreferredName(), password.value());
             }
+        }
         return builder.endObject();
     }
 
@@ -82,7 +87,7 @@ public class BasicAuth implements HttpAuth {
                 if (Field.USERNAME.getPreferredName().equals(fieldName)) {
                     username = parser.text();
                 } else if (Field.PASSWORD.getPreferredName().equals(fieldName)) {
-                    password = WatcherXContentParser.secret(parser);
+                    password = WatcherXContentParser.secretOrNull(parser);
                 } else {
                     throw new ElasticsearchParseException("unsupported field [" + fieldName + "]");
                 }
@@ -94,9 +99,6 @@ public class BasicAuth implements HttpAuth {
         if (username == null) {
             throw new ElasticsearchParseException("username is a required option");
         }
-        if (password == null) {
-            throw new ElasticsearchParseException("password is a required option");
-        }
 
         return new BasicAuth(username, password);
     }

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/history/HistoryStore.java

@@ -79,7 +79,7 @@ public class HistoryStore extends AbstractComponent {
         putUpdateLock.lock();
         try (XContentBuilder builder = XContentFactory.jsonBuilder();
              ThreadContext.StoredContext ignore = stashWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN)) {
-            watchRecord.toXContent(builder, WatcherParams.builder().hideSecrets(true).build());
+            watchRecord.toXContent(builder, WatcherParams.HIDE_SECRETS);
 
             IndexRequest request = new IndexRequest(index, DOC_TYPE, watchRecord.id().value())
                     .source(builder)
@@ -106,7 +106,7 @@ public class HistoryStore extends AbstractComponent {
         try {
             try (XContentBuilder builder = XContentFactory.jsonBuilder();
                  ThreadContext.StoredContext ignore = stashWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN)) {
-                watchRecord.toXContent(builder, WatcherParams.builder().hideSecrets(true).build());
+                watchRecord.toXContent(builder, WatcherParams.HIDE_SECRETS);
 
                 IndexRequest request = new IndexRequest(index, DOC_TYPE, watchRecord.id().value())
                         .source(builder)

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/rest/action/RestGetWatchAction.java

@@ -37,7 +37,7 @@ public class RestGetWatchAction extends WatcherRestHandler {
     }
 
     @Override
-    protected RestChannelConsumer doPrepareRequest(final RestRequest request, WatcherClient client) throws IOException {
+    protected RestChannelConsumer doPrepareRequest(final RestRequest request, WatcherClient client) {
         final GetWatchRequest getWatchRequest = new GetWatchRequest(request.param("id"));
         return channel -> client.getWatch(getWatchRequest, new RestBuilderListener<GetWatchResponse>(channel) {
             @Override
@@ -46,6 +46,7 @@ public class RestGetWatchAction extends WatcherRestHandler {
                         .field("found", response.isFound())
                         .field("_id", response.getId());
                         if (response.isFound()) {
+                            builder.field("_version", response.getVersion());
                             ToXContent.MapParams xContentParams = new ToXContent.MapParams(request.params());
                             builder.field("status", response.getStatus(), xContentParams);
                             builder.field("watch", response.getSource(), xContentParams);

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/rest/action/RestPutWatchAction.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.watcher.rest.action;
 
+import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -46,6 +47,7 @@ public class RestPutWatchAction extends WatcherRestHandler implements RestReques
     protected RestChannelConsumer doPrepareRequest(final RestRequest request, WatcherClient client) throws IOException {
         PutWatchRequest putWatchRequest =
                 new PutWatchRequest(request.param("id"), request.content(), request.getXContentType());
+        putWatchRequest.setVersion(request.paramAsLong("version", Versions.MATCH_ANY));
         putWatchRequest.setActive(request.paramAsBoolean("active", putWatchRequest.isActive()));
         return channel -> client.putWatch(putWatchRequest, new RestBuilderListener<PutWatchResponse>(channel) {
             @Override

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/ack/TransportAckWatchAction.java

@@ -71,8 +71,8 @@ public class TransportAckWatchAction extends WatcherTransportAction<AckWatchRequ
                         listener.onFailure(new ResourceNotFoundException("Watch with id [{}] does not exist", request.getWatchId()));
                     } else {
                         DateTime now = new DateTime(clock.millis(), UTC);
-                        Watch watch =
+                        Watch watch = parser.parseWithSecrets(request.getWatchId(), true, response.getSourceAsBytesRef(),
-                                parser.parseWithSecrets(request.getWatchId(), true, response.getSourceAsBytesRef(), now, XContentType.JSON);
+                                now, XContentType.JSON);
                         watch.version(response.getVersion());
                         watch.status().version(response.getVersion());
                         String[] actionIds = request.getActionIds();

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/get/TransportGetWatchAction.java

@@ -70,11 +70,12 @@ public class TransportGetWatchAction extends WatcherTransportAction<GetWatchRequ
                                     XContentType.JSON);
                             watch.toXContent(builder, WatcherParams.builder()
                                     .hideSecrets(true)
-                                    .put(Watch.INCLUDE_STATUS_KEY, false)
+                                    .includeStatus(false)
                                     .build());
                             watch.version(getResponse.getVersion());
                             watch.status().version(getResponse.getVersion());
-                            listener.onResponse(new GetWatchResponse(watch.id(), watch.status(), builder.bytes(), XContentType.JSON));
+                            listener.onResponse(new GetWatchResponse(watch.id(), getResponse.getVersion(), watch.status(), builder.bytes(),
+                                    XContentType.JSON));
                         }
                     } else {
                         listener.onResponse(new GetWatchResponse(request.getId()));

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/put/TransportPutWatchAction.java

@@ -7,17 +7,16 @@ package org.elasticsearch.xpack.watcher.transport.actions.put;
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DocWriteResponse;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
-import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -25,7 +24,6 @@ import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.core.watcher.transport.actions.put.PutWatchAction;
 import org.elasticsearch.xpack.core.watcher.transport.actions.put.PutWatchRequest;
 import org.elasticsearch.xpack.core.watcher.transport.actions.put.PutWatchResponse;
-import org.elasticsearch.xpack.core.watcher.watch.Payload;
 import org.elasticsearch.xpack.core.watcher.watch.Watch;
 import org.elasticsearch.xpack.watcher.Watcher;
 import org.elasticsearch.xpack.watcher.transport.actions.WatcherTransportAction;
@@ -41,11 +39,28 @@ import static org.elasticsearch.xpack.core.ClientHelper.WATCHER_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
 import static org.joda.time.DateTimeZone.UTC;
 
+/**
+ * This action internally has two modes of operation - an insert and an update mode
+ *
+ * The insert mode will simply put a watch and that is it.
+ * The update mode is a bit more complex and uses versioning. First this prevents the
+ * last-write-wins issue, when two users store the same watch. This could happen due
+ * to UI users. To prevent this a version is required to trigger the update mode.
+ * This mode has been mainly introduced to deal with updates, where the user does not
+ * need to provide secrets like passwords for basic auth or sending emails. If this
+ * is an update, the watch will not parse the secrets coming in, and the resulting JSON
+ * to store the new watch will not contain a password allowing for updates.
+ *
+ * Internally both requests result in an update call, albeit with different parameters and
+ * use of versioning as well as setting the docAsUpsert boolean.
+ */
 public class TransportPutWatchAction extends WatcherTransportAction<PutWatchRequest, PutWatchResponse> {
 
     private final Clock clock;
     private final WatchParser parser;
     private final Client client;
+    private static final ToXContent.Params DEFAULT_PARAMS =
+            WatcherParams.builder().hideSecrets(false).hideHeaders(false).includeStatus(true).build();
 
     @Inject
     public TransportPutWatchAction(Settings settings, TransportService transportService, ThreadPool threadPool, ActionFilters actionFilters,
@@ -62,7 +77,8 @@ public class TransportPutWatchAction extends WatcherTransportAction<PutWatchRequ
     protected void doExecute(PutWatchRequest request, ActionListener<PutWatchResponse> listener) {
         try {
             DateTime now = new DateTime(clock.millis(), UTC);
-            Watch watch = parser.parseWithSecrets(request.getId(), false, request.getSource(), now, request.xContentType());
+            boolean isUpdate = request.getVersion() > 0;
+            Watch watch = parser.parseWithSecrets(request.getId(), false, request.getSource(), now, request.xContentType(), isUpdate);
             watch.setState(request.isActive(), now);
 
             // ensure we only filter for the allowed headers
@@ -72,24 +88,20 @@ public class TransportPutWatchAction extends WatcherTransportAction<PutWatchRequ
             watch.status().setHeaders(filteredHeaders);
 
             try (XContentBuilder builder = jsonBuilder()) {
-                Payload.XContent.Params params = WatcherParams.builder()
+                watch.toXContent(builder, DEFAULT_PARAMS);
-                        .hideSecrets(false)
-                        .hideHeaders(false)
-                        .put(Watch.INCLUDE_STATUS_KEY, "true")
-                        .build();
-                watch.toXContent(builder, params);
-                final BytesReference bytesReference = builder.bytes();
 
-                IndexRequest indexRequest = new IndexRequest(Watch.INDEX).type(Watch.DOC_TYPE).id(request.getId());
+                UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, Watch.DOC_TYPE, request.getId());
-                indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+                updateRequest.docAsUpsert(isUpdate == false);
-                indexRequest.source(bytesReference, XContentType.JSON);
+                updateRequest.version(request.getVersion());
+                updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+                updateRequest.doc(builder);
 
-                executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, indexRequest,
+                executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, updateRequest,
-                        ActionListener.<IndexResponse>wrap(indexResponse -> {
+                        ActionListener.<UpdateResponse>wrap(response -> {
-                            boolean created = indexResponse.getResult() == DocWriteResponse.Result.CREATED;
+                            boolean created = response.getResult() == DocWriteResponse.Result.CREATED;
-                            listener.onResponse(new PutWatchResponse(indexResponse.getId(), indexResponse.getVersion(), created));
+                            listener.onResponse(new PutWatchResponse(response.getId(), response.getVersion(), created));
                         }, listener::onFailure),
-                        client::index);
+                        client::update);
             }
         } catch (Exception e) {
             listener.onFailure(e);

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/watch/WatchParser.java

@@ -9,6 +9,7 @@ import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
@@ -72,12 +73,12 @@ public class WatchParser extends AbstractComponent {
     }
 
     public Watch parse(String name, boolean includeStatus, BytesReference source, XContentType xContentType) throws IOException {
-        return parse(name, includeStatus, false, source, new DateTime(clock.millis(), UTC), xContentType);
+        return parse(name, includeStatus, false, source, new DateTime(clock.millis(), UTC), xContentType, false);
     }
 
     public Watch parse(String name, boolean includeStatus, BytesReference source, DateTime now,
                        XContentType xContentType) throws IOException {
-        return parse(name, includeStatus, false, source, now, xContentType);
+        return parse(name, includeStatus, false, source, now, xContentType, false);
     }
 
     /**
@@ -91,20 +92,24 @@ public class WatchParser extends AbstractComponent {
      * of the watch in the system will be use secrets for sensitive data.
      *
      */
-    public Watch parseWithSecrets(String id, boolean includeStatus, BytesReference source, DateTime now, XContentType xContentType)
+    public Watch parseWithSecrets(String id, boolean includeStatus, BytesReference source, DateTime now,
-            throws IOException {
+                                                  XContentType xContentType, boolean allowRedactedPasswords) throws IOException {
-        return parse(id, includeStatus, true, source, now, xContentType);
+        return parse(id, includeStatus, true, source, now, xContentType, allowRedactedPasswords);
+    }
+
+    public Watch parseWithSecrets(String id, boolean includeStatus, BytesReference source, DateTime now,
+                                  XContentType xContentType) throws IOException {
+        return parse(id, includeStatus, true, source, now, xContentType, false);
     }
 
     private Watch parse(String id, boolean includeStatus, boolean withSecrets, BytesReference source, DateTime now,
-                        XContentType xContentType) throws IOException {
+                        XContentType xContentType, boolean allowRedactedPasswords) throws IOException {
         if (logger.isTraceEnabled()) {
             logger.trace("parsing watch [{}] ", source.utf8ToString());
         }
         // EMPTY is safe here because we never use namedObject
-        try (WatcherXContentParser parser = new WatcherXContentParser(xContentType.xContent()
+        try (WatcherXContentParser parser = new WatcherXContentParser(xContentType.xContent().createParser(NamedXContentRegistry.EMPTY,
-                .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, source),
+                    LoggingDeprecationHandler.INSTANCE, source), now, withSecrets ? cryptoService : null, allowRedactedPasswords)) {
-                now, withSecrets ? cryptoService : null)) {
             parser.nextToken();
             return parse(id, includeStatus, parser);
         } catch (IOException ioe) {
@@ -121,6 +126,7 @@ public class WatchParser extends AbstractComponent {
         TimeValue throttlePeriod = null;
         Map<String, Object> metatdata = null;
         WatchStatus status = null;
+        long version = Versions.MATCH_ANY;
 
         String currentFieldName = null;
         XContentParser.Token token;
@@ -153,6 +159,8 @@ public class WatchParser extends AbstractComponent {
                 actions = actionRegistry.parseActions(id, parser);
             } else if (WatchField.METADATA.match(currentFieldName, parser.getDeprecationHandler())) {
                 metatdata = parser.map();
+            } else if (WatchField.VERSION.match(currentFieldName, parser.getDeprecationHandler())) {
+                version = parser.longValue();
             } else if (WatchField.STATUS.match(currentFieldName, parser.getDeprecationHandler())) {
                 if (includeStatus) {
                     status = WatchStatus.parse(id, parser);
@@ -185,6 +193,7 @@ public class WatchParser extends AbstractComponent {
             status = new WatchStatus(parser.getParseDateTime(), unmodifiableMap(actionsStatuses));
         }
 
-        return new Watch(id, trigger, input, condition, transform, throttlePeriod, actions, metatdata, status);
+
+        return new Watch(id, trigger, input, condition, transform, throttlePeriod, actions,  metatdata, status, version);
     }
 }

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/actions/email/EmailActionTests.java

@@ -71,6 +71,7 @@ import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.startsWith;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -392,7 +393,8 @@ public class EmailActionTests extends ESTestCase {
             }
             if (auth != null) {
                 assertThat(parsed.action().getAuth().user(), is(executable.action().getAuth().user()));
-                assertThat(parsed.action().getAuth().password(), nullValue());
+                assertThat(parsed.action().getAuth().password(), notNullValue());
+                assertThat(parsed.action().getAuth().password().value(), startsWith("::es_redacted::"));
                 assertThat(executable.action().getAuth().password(), notNullValue());
             }
         }

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/actions/throttler/ActionThrottleTests.java

@@ -59,7 +59,8 @@ public class ActionThrottleTests extends AbstractWatcherIntegrationTestCase {
         Action.Builder action = availableAction.action();
         watchSourceBuilder.addAction("test_id", action);
 
-        watcherClient().putWatch(new PutWatchRequest("_id", watchSourceBuilder)).actionGet();
+        watcherClient().putWatch(new PutWatchRequest("_id", watchSourceBuilder.buildAsBytes(XContentType.JSON),
+                XContentType.JSON)).actionGet();
         refresh(Watch.INDEX);
 
         ExecuteWatchRequestBuilder executeWatchRequestBuilder = watcherClient().prepareExecuteWatch("_id")
@@ -103,7 +104,8 @@ public class ActionThrottleTests extends AbstractWatcherIntegrationTestCase {
             }
         }
 
-        watcherClient().putWatch(new PutWatchRequest("_id", watchSourceBuilder)).actionGet();
+        watcherClient().putWatch(new PutWatchRequest("_id",
+                watchSourceBuilder.buildAsBytes(XContentType.JSON), XContentType.JSON)).actionGet();
         refresh(Watch.INDEX);
         executeWatch("_id");
 
@@ -140,7 +142,8 @@ public class ActionThrottleTests extends AbstractWatcherIntegrationTestCase {
         watchSourceBuilder.addAction("fifteen_sec_throttle", new TimeValue(15, TimeUnit.SECONDS),
                 randomFrom(AvailableAction.values()).action());
 
-        watcherClient().putWatch(new PutWatchRequest("_id", watchSourceBuilder)).actionGet();
+        watcherClient().putWatch(new PutWatchRequest("_id",
+                watchSourceBuilder.buildAsBytes(XContentType.JSON), XContentType.JSON)).actionGet();
         refresh(Watch.INDEX);
 
         timeWarp().clock().fastForwardSeconds(1);
@@ -177,7 +180,8 @@ public class ActionThrottleTests extends AbstractWatcherIntegrationTestCase {
         AvailableAction availableAction = randomFrom(AvailableAction.values());
         watchSourceBuilder.addAction("default_global_throttle", availableAction.action());
 
-        watcherClient().putWatch(new PutWatchRequest("_id", watchSourceBuilder)).actionGet();
+        watcherClient().putWatch(new PutWatchRequest("_id",
+                watchSourceBuilder.buildAsBytes(XContentType.JSON), XContentType.JSON)).actionGet();
         refresh(Watch.INDEX);
 
         timeWarp().clock().setTime(new DateTime(DateTimeZone.UTC));
@@ -229,7 +233,8 @@ public class ActionThrottleTests extends AbstractWatcherIntegrationTestCase {
         AvailableAction availableAction = randomFrom(AvailableAction.values());
         watchSourceBuilder.addAction("default_global_throttle", availableAction.action());
 
-        watcherClient().putWatch(new PutWatchRequest("_id", watchSourceBuilder)).actionGet();
+        watcherClient().putWatch(new PutWatchRequest("_id",
+                watchSourceBuilder.buildAsBytes(XContentType.JSON), XContentType.JSON)).actionGet();
         refresh(Watch.INDEX);
 
         timeWarp().clock().setTime(new DateTime(DateTimeZone.UTC));

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

@@ -13,6 +13,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.watcher.common.http.auth.HttpAuthRegistry;
 import org.elasticsearch.xpack.watcher.common.http.auth.basic.BasicAuth;
 import org.elasticsearch.xpack.watcher.common.http.auth.basic.BasicAuthFactory;
@@ -137,7 +138,7 @@ public class HttpRequestTemplateTests extends ESTestCase {
                 new BasicAuthFactory(null)));
         HttpRequestTemplate.Parser parser = new HttpRequestTemplate.Parser(registry);
 
-        XContentBuilder xContentBuilder = template.toXContent(jsonBuilder(), ToXContent.EMPTY_PARAMS);
+        XContentBuilder xContentBuilder = template.toXContent(jsonBuilder(), WatcherParams.builder().hideSecrets(false).build());
         XContentParser xContentParser = createParser(xContentBuilder);
         xContentParser.nextToken();
         HttpRequestTemplate parsed = parser.parse(xContentParser);

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

@@ -11,6 +11,7 @@ import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.watcher.common.http.auth.HttpAuthRegistry;
 import org.elasticsearch.xpack.watcher.common.http.auth.basic.BasicAuth;
 import org.elasticsearch.xpack.watcher.common.http.auth.basic.BasicAuthFactory;
@@ -124,7 +125,7 @@ public class HttpRequestTests extends ESTestCase {
         assertNotNull(httpRequest);
 
         try (XContentBuilder xContentBuilder = randomFrom(jsonBuilder(), smileBuilder(), yamlBuilder(), cborBuilder())) {
-            httpRequest.toXContent(xContentBuilder, ToXContent.EMPTY_PARAMS);
+            httpRequest.toXContent(xContentBuilder, WatcherParams.builder().hideSecrets(false).build());
 
             HttpAuthRegistry registry = new HttpAuthRegistry(singletonMap(BasicAuth.TYPE, new BasicAuthFactory(null)));
             HttpRequest.Parser httpRequestParser = new HttpRequest.Parser(registry);

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/execution/ExecutionServiceTests.java

@@ -1014,7 +1014,7 @@ public class ExecutionServiceTests extends ESTestCase {
     public void testUpdateWatchStatusDoesNotUpdateState() throws Exception {
         WatchStatus status = new WatchStatus(DateTime.now(UTC), Collections.emptyMap());
         Watch watch = new Watch("_id", new ManualTrigger(), new ExecutableNoneInput(logger), InternalAlwaysCondition.INSTANCE, null, null,
-                Collections.emptyList(), null, status);
+                Collections.emptyList(), null, status, 1L);
 
         final AtomicBoolean assertionsTriggered = new AtomicBoolean(false);
         doAnswer(invocation -> {

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/history/HistoryStoreTests.java

@@ -32,23 +32,21 @@ import org.elasticsearch.xpack.watcher.common.http.HttpResponse;
 import org.elasticsearch.xpack.watcher.notification.jira.JiraAccount;
 import org.elasticsearch.xpack.watcher.notification.jira.JiraIssue;
 import org.elasticsearch.xpack.watcher.trigger.schedule.ScheduleTriggerEvent;
-import org.hamcrest.Description;
-import org.hamcrest.Matcher;
-import org.hamcrest.TypeSafeMatcher;
 import org.joda.time.DateTime;
 import org.junit.Before;
+import org.mockito.ArgumentCaptor;
 
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static org.elasticsearch.xpack.core.watcher.history.HistoryStoreField.getHistoryIndexNameForTime;
 import static org.elasticsearch.xpack.core.watcher.support.WatcherIndexTemplateRegistryField.INDEX_TEMPLATE_VERSION;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.core.IsEqual.equalTo;
 import static org.joda.time.DateTimeZone.UTC;
 import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.argThat;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -127,7 +125,7 @@ public class HistoryStoreTests extends ESTestCase {
         when(httpClient.execute(any(HttpRequest.class))).thenReturn(new HttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR));
 
         final String username = randomFrom("admin", "elastic", "test");
-        final String password = randomFrom("secret", "password", "123456");
+        final String password = randomFrom("secret", "supersecret", "123456");
         final String url = "https://" + randomFrom("localhost", "internal-jira.elastic.co") + ":" + randomFrom(80, 8080, 449, 9443);
 
         Settings settings = Settings.builder().put("url", url).put("user", username).put("password", password).build();
@@ -161,49 +159,17 @@ public class HistoryStoreTests extends ESTestCase {
 
         PlainActionFuture<IndexResponse> indexResponseFuture = PlainActionFuture.newFuture();
         indexResponseFuture.onResponse(mock(IndexResponse.class));
-        when(client.index(any())).thenReturn(indexResponseFuture);
+        ArgumentCaptor<IndexRequest> requestCaptor = ArgumentCaptor.forClass(IndexRequest.class);
+        when(client.index(requestCaptor.capture())).thenReturn(indexResponseFuture);
         if (randomBoolean()) {
             historyStore.put(watchRecord);
         } else {
             historyStore.forcePut(watchRecord);
         }
-        verify(client).index(argThat(indexRequestDoesNotContainPassword(username, password)));
-    }
 
-    private static Matcher<IndexRequest> indexRequestDoesNotContainPassword(String username, String password) {
+        assertThat(requestCaptor.getAllValues(), hasSize(1));
-        return new IndexRequestNoPasswordMatcher(username, password);
+        String indexedJson = requestCaptor.getValue().source().utf8ToString();
-    }
+        assertThat(indexedJson, containsString(username));
-
+        assertThat(indexedJson, not(containsString(password)));
-    private static class IndexRequestNoPasswordMatcher extends TypeSafeMatcher<IndexRequest> {
-
-        private String username;
-        private String password;
-
-        IndexRequestNoPasswordMatcher(String username, String password) {
-            this.username = username;
-            this.password = password;
-        }
-
-        @Override
-        protected boolean matchesSafely(IndexRequest indexRequest) {
-            String source  = indexRequest.source().utf8ToString();
-            assertThat(source, containsString(username));
-            assertThat(source, not(containsString(password)));
-            return true;
-        }
-
-        @Override
-        public void describeMismatchSafely(final IndexRequest indexRequest, final Description mismatchDescription) {
-            mismatchDescription.appendText(" was ").appendValue(indexRequest.sourceAsMap());
-        }
-
-        @Override
-        public void describeTo(final Description description) {
-            description.appendText("IndexRequest id should contain username [")
-                    .appendValue(username)
-                    .appendText("] and should not contain [")
-                    .appendValue(password)
-                    .appendText("]");
-        }
     }
 }

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/input/http/HttpInputTests.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.watcher.input.http;
 
+import com.carrotsearch.randomizedtesting.annotations.Repeat;
 import io.netty.handler.codec.http.HttpHeaders;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.MapBuilder;
@@ -19,6 +20,7 @@ import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.watcher.execution.WatchExecutionContext;
 import org.elasticsearch.xpack.core.watcher.support.xcontent.ObjectPath;
+import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.core.watcher.watch.Payload;
 import org.elasticsearch.xpack.watcher.common.http.HttpClient;
 import org.elasticsearch.xpack.watcher.common.http.HttpContentType;
@@ -63,6 +65,7 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class HttpInputTests extends ESTestCase {
+
     private HttpClient httpClient;
     private HttpInputFactory httpParser;
     private TextTemplateEngine templateEngine;
@@ -181,7 +184,8 @@ public class HttpInputTests extends ESTestCase {
         }
 
         inputBuilder.expectedResponseXContentType(expectedResponseXContentType);
-        XContentBuilder source = jsonBuilder().value(inputBuilder.build());
+        XContentBuilder source = jsonBuilder();
+        inputBuilder.build().toXContent(source, WatcherParams.builder().hideSecrets(false).build());
         XContentParser parser = createParser(source);
         parser.nextToken();
         HttpInput result = httpParser.parseInput("_id", parser);

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/notification/email/EmailSecretsIntegrationTests.java

@@ -41,6 +41,7 @@ import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.startsWith;
 
 public class EmailSecretsIntegrationTests extends AbstractWatcherIntegrationTestCase {
     private EmailServer server;
@@ -119,7 +120,11 @@ public class EmailSecretsIntegrationTests extends AbstractWatcherIntegrationTest
         assertThat(watchResponse.getId(), is("_id"));
         XContentSource contentSource = watchResponse.getSource();
         value = contentSource.getValue("actions._email.email.password");
-        assertThat(value, nullValue());
+        if (encryptSensitiveData) {
+            assertThat(value.toString(), startsWith("::es_encrypted::"));
+        } else {
+            assertThat(value, is("::es_redacted::"));
+        }
 
         // now we restart, to make sure the watches and their secrets are reloaded from the index properly
         stopWatcher();

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/notification/email/attachment/ReportingAttachmentParserTests.java

@@ -11,12 +11,12 @@ import org.elasticsearch.common.ParsingException;
 import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.watcher.execution.WatchExecutionContext;
 import org.elasticsearch.xpack.core.watcher.execution.Wid;
+import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.core.watcher.watch.Payload;
 import org.elasticsearch.xpack.watcher.common.http.HttpClient;
 import org.elasticsearch.xpack.watcher.common.http.HttpMethod;
@@ -114,10 +114,11 @@ public class ReportingAttachmentParserTests extends ESTestCase {
 
         HttpAuth auth = null;
         boolean withAuth = randomBoolean();
+        boolean isPasswordEncrypted = randomBoolean();
         if (withAuth) {
             builder.startObject("auth").startObject("basic")
                     .field("username", "foo")
-                    .field("password", "secret")
+                    .field("password", isPasswordEncrypted ? "::es_redacted::" :"secret")
                     .endObject().endObject();
             auth = new BasicAuth("foo", "secret".toCharArray());
         }
@@ -140,13 +141,14 @@ public class ReportingAttachmentParserTests extends ESTestCase {
 
         XContentBuilder toXcontentBuilder = jsonBuilder().startObject();
         List<EmailAttachmentParser.EmailAttachment> attachments = new ArrayList<>(emailAttachments.getAttachments());
-        attachments.get(0).toXContent(toXcontentBuilder, ToXContent.EMPTY_PARAMS);
+        WatcherParams watcherParams = WatcherParams.builder().hideSecrets(isPasswordEncrypted).build();
+        attachments.get(0).toXContent(toXcontentBuilder, watcherParams);
         toXcontentBuilder.endObject();
         assertThat(toXcontentBuilder.string(), is(builder.string()));
 
         XContentBuilder attachmentXContentBuilder = jsonBuilder().startObject();
         ReportingAttachment attachment = new ReportingAttachment(id, dashboardUrl, isInline, interval, retries, auth, proxy);
-        attachment.toXContent(attachmentXContentBuilder, ToXContent.EMPTY_PARAMS);
+        attachment.toXContent(attachmentXContentBuilder, watcherParams);
         attachmentXContentBuilder.endObject();
         assertThat(attachmentXContentBuilder.string(), is(builder.string()));
 

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/test/WatcherTestUtils.java

@@ -133,7 +133,7 @@ public final class WatcherTestUtils {
                 null,
                 new ArrayList<>(),
                 null,
-                new WatchStatus(new DateTime(0, UTC), emptyMap()));
+                new WatchStatus(new DateTime(0, UTC), emptyMap()), 1L);
         TriggeredExecutionContext context = new TriggeredExecutionContext(watch.id(),
                 new DateTime(0, UTC),
                 new ScheduleTriggerEvent(watch.id(), new DateTime(0, UTC), new DateTime(0, UTC)),
@@ -180,7 +180,7 @@ public final class WatcherTestUtils {
                 new TimeValue(0),
                 actions,
                 Collections.singletonMap("foo", "bar"),
-                new WatchStatus(now, statuses));
+                new WatchStatus(now, statuses), 1L);
     }
 
     public static SearchType getRandomSupportedSearchType() {

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/test/bench/ScheduleEngineTriggerBenchmark.java

@@ -62,7 +62,7 @@ public class ScheduleEngineTriggerBenchmark {
         List<Watch> watches = new ArrayList<>(numWatches);
         for (int i = 0; i < numWatches; i++) {
             watches.add(new Watch("job_" + i, new ScheduleTrigger(interval(interval + "s")), new ExecutableNoneInput(logger),
-                    InternalAlwaysCondition.INSTANCE, null, null, Collections.emptyList(), null, null));
+                    InternalAlwaysCondition.INSTANCE, null, null, Collections.emptyList(), null, null, 1L));
         }
         ScheduleRegistry scheduleRegistry = new ScheduleRegistry(emptySet());
 

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/test/bench/WatcherExecutorServiceBenchmark.java

@@ -85,7 +85,7 @@ public class WatcherExecutorServiceBenchmark {
                                 ScriptType.INLINE,
                                 Script.DEFAULT_SCRIPT_LANG,
                                 "ctx.payload.hits.total > 0",
-                                emptyMap()))));
+                                emptyMap()))).buildAsBytes(XContentType.JSON), XContentType.JSON);
                 putAlertRequest.setId(name);
                 watcherClient.putWatch(putAlertRequest).actionGet();
             }
@@ -128,7 +128,7 @@ public class WatcherExecutorServiceBenchmark {
                         .input(searchInput(templateRequest(new SearchSourceBuilder(), "test"))
                                 .extractKeys("hits.total"))
                         .condition(new ScriptCondition(new Script(ScriptType.INLINE, Script.DEFAULT_SCRIPT_LANG, "1 == 1", emptyMap())))
-                        .addAction("_id", indexAction("index", "type")));
+                        .addAction("_id", indexAction("index", "type")).buildAsBytes(XContentType.JSON), XContentType.JSON);
                 putAlertRequest.setId(name);
                 watcherClient.putWatch(putAlertRequest).actionGet();
             }
@@ -175,7 +175,7 @@ public class WatcherExecutorServiceBenchmark {
                                 ScriptType.INLINE,
                                 Script.DEFAULT_SCRIPT_LANG,
                                 "ctx.payload.tagline == \"You Know, for Search\"",
-                                emptyMap()))));
+                                emptyMap()))).buildAsBytes(XContentType.JSON), XContentType.JSON);
                 putAlertRequest.setId(name);
                 watcherClient.putWatch(putAlertRequest).actionGet();
             }
@@ -186,15 +186,12 @@ public class WatcherExecutorServiceBenchmark {
             for (int i = 0; i < numThreads; i++) {
                 final int begin = i * watchersPerThread;
                 final int end = (i + 1) * watchersPerThread;
-                Runnable r = new Runnable() {
+                Runnable r = () -> {
-                    @Override
-                    public void run() {
                     while (true) {
                         for (int j = begin; j < end; j++) {
                             scheduler.trigger("_name" + j);
                         }
                     }
-                    }
                 };
                 threads[i] = new Thread(r);
                 threads[i].start();

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/test/integration/HttpSecretsIntegrationTests.java

@@ -46,17 +46,17 @@ import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
-import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.startsWith;
 import static org.joda.time.DateTimeZone.UTC;
 
 public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTestCase {
 
-    static final String USERNAME = "_user";
+    private static final String USERNAME = "_user";
-    static final String PASSWORD = "_passwd";
+    private static final String PASSWORD = "_passwd";
 
     private MockWebServer webServer = new MockWebServer();
-    private static Boolean encryptSensitiveData;
+    private static Boolean encryptSensitiveData = null;
-    private static byte[] encryptionKey;
+    private static byte[] encryptionKey = CryptoServiceTests.generateKey();
 
     @Before
     public void init() throws Exception {
@@ -64,7 +64,7 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTestC
     }
 
     @After
-    public void cleanup() throws Exception {
+    public void cleanup() {
         webServer.close();
     }
 
@@ -72,9 +72,6 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTestC
     protected Settings nodeSettings(int nodeOrdinal) {
         if (encryptSensitiveData == null) {
             encryptSensitiveData = randomBoolean();
-            if (encryptSensitiveData) {
-                encryptionKey = CryptoServiceTests.generateKey();
-            }
         }
         if (encryptSensitiveData) {
             MockSecureSettings secureSettings = new MockSecureSettings();
@@ -109,14 +106,14 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTestC
         Object value = XContentMapValues.extractValue("input.http.request.auth.basic.password", source);
         assertThat(value, notNullValue());
         if (encryptSensitiveData) {
-            assertThat(value, not(is((Object) PASSWORD)));
+            assertThat(value.toString(), startsWith("::es_encrypted::"));
             MockSecureSettings mockSecureSettings = new MockSecureSettings();
             mockSecureSettings.setFile(WatcherField.ENCRYPTION_KEY_SETTING.getKey(), encryptionKey);
             Settings settings = Settings.builder().setSecureSettings(mockSecureSettings).build();
             CryptoService cryptoService = new CryptoService(settings);
             assertThat(new String(cryptoService.decrypt(((String) value).toCharArray())), is(PASSWORD));
         } else {
-            assertThat(value, is((Object) PASSWORD));
+            assertThat(value, is(PASSWORD));
         }
 
         // verifying the password is not returned by the GET watch API
@@ -127,7 +124,11 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTestC
         value = contentSource.getValue("input.http.request.auth.basic");
         assertThat(value, notNullValue()); // making sure we have the basic auth
         value = contentSource.getValue("input.http.request.auth.basic.password");
-        assertThat(value, nullValue()); // and yet we don't have the password
+        if (encryptSensitiveData) {
+            assertThat(value.toString(), startsWith("::es_encrypted::"));
+        } else {
+            assertThat(value, is("::es_redacted::"));
+        }
 
         // now we restart, to make sure the watches and their secrets are reloaded from the index properly
         stopWatcher();
@@ -195,7 +196,11 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTestC
         value = contentSource.getValue("actions._webhook.webhook.auth.basic");
         assertThat(value, notNullValue()); // making sure we have the basic auth
         value = contentSource.getValue("actions._webhook.webhook.auth.basic.password");
-        assertThat(value, nullValue()); // and yet we don't have the password
+        if (encryptSensitiveData) {
+            assertThat(value.toString(), startsWith("::es_encrypted::"));
+        } else {
+            assertThat(value, is("::es_redacted::"));
+        }
 
         // now we restart, to make sure the watches and their secrets are reloaded from the index properly
         stopWatcher();
@@ -229,7 +234,11 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTestC
         assertThat(value, is(USERNAME)); // the auth username exists
 
         value = contentSource.getValue("result.actions.0.webhook.request.auth.basic.password");
-        assertThat(value, nullValue()); // but the auth password was filtered out
+        if (encryptSensitiveData) {
+            assertThat(value.toString(), startsWith("::es_encrypted::"));
+        } else {
+            assertThat(value.toString(), is("::es_redacted::"));
+        }
 
         assertThat(webServer.requests(), hasSize(1));
         assertThat(webServer.requests().get(0).getHeader("Authorization"),

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/transport/action/put/PutWatchSerializationTests.java

@@ -7,6 +7,7 @@ package org.elasticsearch.xpack.watcher.transport.action.put;
 
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.test.ESTestCase;
@@ -33,6 +34,7 @@ public class PutWatchSerializationTests extends ESTestCase {
         assertThat(readRequest.getId(), is(request.getId()));
         assertThat(readRequest.getSource(), is(request.getSource()));
         assertThat(readRequest.xContentType(), is(request.xContentType()));
+        assertThat(readRequest.getVersion(), is(request.getVersion()));
     }
 
     public void testPutWatchSerializationXContent() throws Exception {
@@ -52,5 +54,6 @@ public class PutWatchSerializationTests extends ESTestCase {
         assertThat(readRequest.getId(), is(request.getId()));
         assertThat(readRequest.getSource(), is(request.getSource()));
         assertThat(readRequest.xContentType(), is(XContentType.JSON));
+        assertThat(readRequest.getVersion(), is(Versions.MATCH_ANY));
     }
 }

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/transport/actions/put/TransportPutWatchActionTests.java

@@ -11,6 +11,7 @@ import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.index.Index;
@@ -36,6 +37,7 @@ import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyBoolean;
 import static org.mockito.Matchers.anyObject;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.doAnswer;
@@ -57,7 +59,7 @@ public class TransportPutWatchActionTests extends ESTestCase {
         TransportService transportService = mock(TransportService.class);
 
         WatchParser parser = mock(WatchParser.class);
-        when(parser.parseWithSecrets(eq("_id"), eq(false), anyObject(), anyObject(), anyObject())).thenReturn(watch);
+        when(parser.parseWithSecrets(eq("_id"), eq(false), anyObject(), anyObject(), anyObject(), anyBoolean())).thenReturn(watch);
 
         Client client = mock(Client.class);
         when(client.threadPool()).thenReturn(threadPool);

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/trigger/schedule/engine/TickerScheduleEngineTests.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.watcher.trigger.schedule.engine;
 
+import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.watcher.trigger.TriggerEvent;
@@ -252,6 +253,6 @@ public class TickerScheduleEngineTests extends ESTestCase {
     private Watch createWatch(String name, Schedule schedule) {
         return new Watch(name, new ScheduleTrigger(schedule), new ExecutableNoneInput(logger),
                 InternalAlwaysCondition.INSTANCE, null, null,
-                Collections.emptyList(), null, null);
+                Collections.emptyList(), null, null, Versions.MATCH_ANY);
     }
 }

plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/watch/WatchTests.java

@@ -212,7 +212,7 @@ public class WatchTests extends ESTestCase {
 
         TimeValue throttlePeriod = randomBoolean() ? null : TimeValue.timeValueSeconds(randomIntBetween(5, 10000));
 
-        Watch watch = new Watch("_name", trigger, input, condition, transform, throttlePeriod, actions, metadata, watchStatus);
+        Watch watch = new Watch("_name", trigger, input, condition, transform, throttlePeriod, actions, metadata, watchStatus, 1L);
 
         BytesReference bytes = jsonBuilder().value(watch).bytes();
         logger.info("{}", bytes.utf8ToString());

qa/full-cluster-restart/src/test/java/org/elasticsearch/xpack/restart/FullClusterRestartIT.java

@@ -45,6 +45,7 @@ import java.util.stream.Collectors;
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static org.elasticsearch.common.unit.TimeValue.timeValueSeconds;
+import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.everyItem;
@@ -340,10 +341,10 @@ public class FullClusterRestartIT extends ESRestTestCase {
         assertEquals("example.com", request.get("host"));
         assertEquals("{{ctx.metadata.report_url}}", request.get("path"));
         assertEquals(8443, request.get("port"));
-        Map<?, ?> basic = ObjectPath.eval("auth.basic", request);
+        Map<String, String> basic = ObjectPath.eval("auth.basic", request);
         assertThat(basic, hasEntry("username", "Aladdin"));
         // password doesn't come back because it is hidden
-        assertThat(basic, not(hasKey("password")));
+        assertThat(basic, hasEntry(is("password"), anyOf(startsWith("::es_encrypted::"), is("::es_redacted::"))));
 
         Map<String, Object> history = toMap(client().performRequest("GET", ".watcher-history*/_search"));
         Map<String, Object> hits = (Map<String, Object>) history.get("hits");

qa/smoke-test-watcher/src/test/resources/rest-api-spec/test/actions/20_jira.yml

@@ -211,7 +211,7 @@
   - match: { hits.hits.0._source.result.actions.0.jira.request.method: "post" }
   - match: { hits.hits.0._source.result.actions.0.jira.request.path: "/rest/api/2/issue" }
   - match: { hits.hits.0._source.result.actions.0.jira.request.auth.basic.username: "xpack-user@elastic.co" }
-  - is_false: hits.hits.0._source.result.actions.0.jira.request.auth.basic.password
+  - match: { hits.hits.0._source.result.actions.0.jira.request.auth.basic.password: "::es_redacted::" }
   - match: { hits.hits.0._source.result.actions.0.jira.response.body: "{\"errorMessages\":[],\"errors\":{\"issuetype\":\"issue type is required\"}}" }
 
 ---