Watcher: Store username on watch execution

There is currently no way to see what user executed a watch. This commit
adds the decrypted username to each execution in the watch history, in a
new field executed_by.

Closes elastic#31772

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/Authentication.java

@@ -88,13 +88,17 @@ static Authentication deserializeHeaderAndPutInContext(String header, ThreadCont
             throws IOException, IllegalArgumentException {
         assert ctx.getTransient(AuthenticationField.AUTHENTICATION_KEY) == null;
 
+        Authentication authentication = decode(header);
+        ctx.putTransient(AuthenticationField.AUTHENTICATION_KEY, authentication);
+        return authentication;
+    }
+
+    public static Authentication decode(String header) throws IOException {
         byte[] bytes = Base64.getDecoder().decode(header);
         StreamInput input = StreamInput.wrap(bytes);
         Version version = Version.readVersion(input);
         input.setVersion(version);
-        Authentication authentication = new Authentication(input);
-        ctx.putTransient(AuthenticationField.AUTHENTICATION_KEY, authentication);
-        return authentication;
+        return new Authentication(input);
     }
 
     /**

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/execution/WatchExecutionContext.java

@@ -8,6 +8,8 @@
 import org.elasticsearch.common.CheckedSupplier;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationField;
 import org.elasticsearch.xpack.core.watcher.actions.ActionWrapperResult;
 import org.elasticsearch.xpack.core.watcher.condition.Condition;
 import org.elasticsearch.xpack.core.watcher.history.WatchRecord;
@@ -43,6 +45,7 @@ public abstract class WatchExecutionContext {
     private Transform.Result transformResult;
     private ConcurrentMap<String, ActionWrapperResult> actionsResults = ConcurrentCollections.newConcurrentMap();
     private String nodeId;
+    private String executedBy;
 
     public WatchExecutionContext(String watchId, DateTime executionTime, TriggerEvent triggerEvent, TimeValue defaultThrottlePeriod) {
         this.id = new Wid(watchId, executionTime);
@@ -85,6 +88,14 @@ public Watch watch() {
     public void ensureWatchExists(CheckedSupplier<Watch, Exception> supplier) throws Exception {
         if (watch == null) {
             watch = supplier.get();
+            // now that the watch exists, extract out the authentication
+            if (watch.status() != null && watch.status().getHeaders() != null) {
+                String header = watch.status().getHeaders().get(AuthenticationField.AUTHENTICATION_KEY);
+                if (header != null) {
+                    Authentication auth = Authentication.decode(header);
+                    executedBy = auth.getUser().principal();
+                }
+            }
         }
     }
 
@@ -137,6 +148,11 @@ public String getNodeId() {
         return nodeId;
     }
 
+    /**
+     * @return The user that executes the watch, which will be stored in the watch history
+     */
+    public String getExecutedBy() { return executedBy; }
+
     public void start() {
         assert phase == ExecutionPhase.AWAITS_EXECUTION;
         relativeStartTime = System.nanoTime();

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/history/WatchRecord.java

@@ -43,12 +43,14 @@ public abstract class WatchRecord implements ToXContentObject {
     private static final ParseField METADATA = new ParseField("metadata");
     private static final ParseField EXECUTION_RESULT = new ParseField("result");
     private static final ParseField EXCEPTION = new ParseField("exception");
+    private static final ParseField EXECUTED_BY = new ParseField("executed_by");
 
     protected final Wid id;
     protected final Watch watch;
     private final String nodeId;
     protected final TriggerEvent triggerEvent;
     protected final ExecutionState state;
+    private final String executedBy;
 
     // only emitted to xcontent in "debug" mode
     protected final Map<String, Object> vars;
@@ -60,7 +62,7 @@ public abstract class WatchRecord implements ToXContentObject {
 
     private WatchRecord(Wid id, TriggerEvent triggerEvent, ExecutionState state, Map<String, Object> vars, ExecutableInput input,
                         ExecutableCondition condition, Map<String, Object> metadata, Watch watch, WatchExecutionResult executionResult,
-                        String nodeId) {
+                        String nodeId, String executedBy) {
         this.id = id;
         this.triggerEvent = triggerEvent;
         this.state = state;
@@ -71,15 +73,16 @@ private WatchRecord(Wid id, TriggerEvent triggerEvent, ExecutionState state, Map
         this.executionResult = executionResult;
         this.watch = watch;
         this.nodeId = nodeId;
+        this.executedBy = executedBy;
     }
 
     private WatchRecord(Wid id, TriggerEvent triggerEvent, ExecutionState state, String nodeId) {
-        this(id, triggerEvent, state, Collections.emptyMap(), null, null, null, null, null, nodeId);
+        this(id, triggerEvent, state, Collections.emptyMap(), null, null, null, null, null, nodeId, null);
     }
 
     private WatchRecord(WatchRecord record, ExecutionState state) {
         this(record.id, record.triggerEvent, state, record.vars, record.input, record.condition, record.metadata, record.watch,
-                record.executionResult, record.nodeId);
+                record.executionResult, record.nodeId, record.executedBy);
     }
 
     private WatchRecord(WatchExecutionContext context, ExecutionState state) {
@@ -88,12 +91,13 @@ private WatchRecord(WatchExecutionContext context, ExecutionState state) {
                 context.watch() != null ? context.watch().condition() : null,
                 context.watch() != null ? context.watch().metadata() : null,
                 context.watch(),
-                null, context.getNodeId());
+                null, context.getNodeId(), context.getExecutedBy());
     }
 
     private WatchRecord(WatchExecutionContext context, WatchExecutionResult executionResult) {
         this(context.id(), context.triggerEvent(), getState(executionResult), context.vars(), context.watch().input(),
-                context.watch().condition(), context.watch().metadata(), context.watch(), executionResult, context.getNodeId());
+                context.watch().condition(), context.watch().metadata(), context.watch(), executionResult, context.getNodeId(),
+                context.getExecutedBy());
     }
 
     public static ExecutionState getState(WatchExecutionResult executionResult) {
@@ -179,6 +183,9 @@ public final XContentBuilder toXContent(XContentBuilder builder, Params params)
         if (executionResult != null) {
             builder.field(EXECUTION_RESULT.getPreferredName(), executionResult, params);
         }
+        if (executedBy != null) {
+            builder.field(EXECUTED_BY.getPreferredName(), executedBy);
+        }
         innerToXContent(builder, params);
         builder.endObject();
         return builder;

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/support/WatcherIndexTemplateRegistryField.java

@@ -13,8 +13,9 @@ public final class WatcherIndexTemplateRegistryField {
     // version 6: upgrade to ES 6, removal of _status field
     // version 7: add full exception stack traces for better debugging
     // version 8: fix slack attachment property not to be dynamic, causing field type issues
+    // version 9: add a executed_by field defining which user executed the watch
     // Note: if you change this, also inform the kibana team around the watcher-ui
-    public static final String INDEX_TEMPLATE_VERSION = "8";
+    public static final String INDEX_TEMPLATE_VERSION = "9";
     public static final String HISTORY_TEMPLATE_NAME = ".watch-history-" + INDEX_TEMPLATE_VERSION;
     public static final String TRIGGERED_TEMPLATE_NAME = ".triggered_watches";
     public static final String WATCHES_TEMPLATE_NAME = ".watches";

x-pack/plugin/core/src/main/resources/watch-history.json

@@ -120,6 +120,9 @@
         "messages": {
           "type": "text"
         },
+        "executed_by": {
+          "type": "text"
+        },
         "exception" : {
           "type" : "object",
           "enabled" : false

x-pack/qa/smoke-test-watcher-with-security/roles.yml

@@ -21,6 +21,7 @@ watcher_manager:
   run_as:
     - powerless_user
     - watcher_manager
+    - x_pack_rest_user
 
 watcher_monitor:
   cluster:

x-pack/qa/smoke-test-watcher-with-security/src/test/resources/rest-api-spec/test/watcher/watcher_and_security/20_test_run_as_execute_watch.yml

@@ -74,10 +74,63 @@ teardown:
         id: "my_watch"
   - match: { watch_record.watch_id: "my_watch" }
   - match: { watch_record.state: "executed" }
+  - match: { watch_record.executed_by: "watcher_manager" }
 
 
 
 
+---
+"Test watch is runas user properly recorded":
+  - do:
+      xpack.watcher.put_watch:
+        id: "my_watch"
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "search" : {
+                "request" : {
+                  "indices" : [ "my_test_index" ],
+                  "body" :{
+                    "query" : { "match_all": {} }
+                  }
+                }
+              }
+            },
+            "condition" : {
+              "compare" : {
+                "ctx.payload.hits.total" : {
+                  "gte" : 1
+                }
+              }
+            },
+            "actions": {
+              "logging": {
+                "logging": {
+                  "text": "Successfully ran my_watch to test for search input"
+                }
+              }
+            }
+          }
+  - match: { _id: "my_watch" }
+
+  - do:
+      xpack.watcher.get_watch:
+        id: "my_watch"
+  - match: { _id: "my_watch" }
+  - is_false: watch.status.headers
+
+  - do:
+      headers: { es-security-runas-user: x_pack_rest_user }
+      xpack.watcher.execute_watch:
+        id: "my_watch"
+  - match: { watch_record.watch_id: "my_watch" }
+  - match: { watch_record.state: "executed" }
+  - match: { watch_record.executed_by: "x_pack_rest_user" }
+
+
 ---
 "Test watch search input does not work against index user is not allowed to read":
 
@@ -130,6 +183,7 @@ teardown:
   - match: { watch_record.watch_id: "my_watch" }
   # because we are not allowed to read the index, there wont be any data
   - match: { watch_record.state: "execution_not_needed" }
+  - match: { watch_record.executed_by: "watcher_manager" }
 
 
 ---
@@ -272,6 +326,7 @@ teardown:
         id: "my_watch"
   - match: { watch_record.watch_id: "my_watch" }
   - match: { watch_record.state: "executed" }
+  - match: { watch_record.executed_by: "watcher_manager" }
 
   - do:
       get:
@@ -320,6 +375,7 @@ teardown:
         id: "my_watch"
   - match: { watch_record.watch_id: "my_watch" }
   - match: { watch_record.state: "executed" }
+  - match: { watch_record.executed_by: "watcher_manager" }
 
   - do:
       get: