Hijack document APIs for .watches index

This change hijacks the document APIs for the .watches index and prevents direct access to these APIs via REST.
This hijacking is controlled by the "watcher.index.rest.direct_access" setting, this setting defaults to false which restricts access to the index. To allow direct access to the .watches index from the REST API set this setting to `true`
Copied json files from core es to test this feature and added comments to the json files indicating why they have been copied from es core.

Fixes elastic/elasticsearch#336

Original commit: elastic/x-pack-elasticsearch@22335750bd

rest-api-spec/api/bulk.json

@@ -0,0 +1,54 @@
+{
+  "bulk": {
+    "documentation": "This file is copied from es core just to verify that the .watches api hijacking works",
+    "methods": ["POST", "PUT"],
+    "url": {
+      "path": "/_bulk",
+      "paths": ["/_bulk", "/{index}/_bulk", "/{index}/{type}/_bulk"],
+      "parts": {
+        "index": {
+          "type" : "string",
+          "description" : "Default index for items which don't provide one"
+        },
+        "type": {
+          "type" : "string",
+          "description" : "Default document type for items which don't provide one"
+        }
+      },
+      "params": {
+        "consistency": {
+          "type" : "enum",
+          "options" : ["one", "quorum", "all"],
+          "description" : "Explicit write consistency setting for the operation"
+        },
+        "refresh": {
+          "type" : "boolean",
+          "description" : "Refresh the index after performing the operation"
+        },
+        "replication": {
+          "type" : "enum",
+          "options" : ["sync","async"],
+          "default" : "sync",
+          "description" : "Explicitely set the replication type"
+        },
+        "routing": {
+          "type" : "string",
+          "description" : "Specific routing value"
+        },
+        "timeout": {
+          "type" : "time",
+          "description" : "Explicit operation timeout"
+        },
+        "type": {
+          "type" : "string",
+          "description" : "Default document type for items which don't provide one"
+        }
+      }
+    },
+    "body": {
+      "description" : "The operation definition and data (action-data pairs), separated by newlines",
+      "required" : true,
+      "serialize" : "bulk"
+    }
+  }
+}

rest-api-spec/api/cluster.health.json

@@ -1,6 +1,6 @@
 {
   "cluster.health": {
-    "documentation": "http://www.elasticsearch.org/guide/en/elasticsearch/reference/1.x/cluster-health.html",
+    "documentation": "This file is copied from es core because the REST test framework requires it",
     "methods": ["GET"],
     "url": {
       "path": "/_cluster/health",

rest-api-spec/api/delete.json

@@ -0,0 +1,66 @@
+{
+  "delete": {
+    "documentation": "This file is copied from es core just to verify that the .watches api hijacking works",
+    "methods": ["DELETE"],
+    "url": {
+      "path": "/{index}/{type}/{id}",
+      "paths": ["/{index}/{type}/{id}"],
+      "parts": {
+        "id": {
+          "type" : "string",
+          "required" : true,
+          "description" : "The document ID"
+        },
+        "index": {
+          "type" : "string",
+          "required" : true,
+          "description" : "The name of the index"
+        },
+        "type": {
+          "type" : "string",
+          "required" : true,
+          "description" : "The type of the document"
+        }
+      },
+      "params": {
+        "consistency": {
+          "type" : "enum",
+          "options" : ["one", "quorum", "all"],
+          "description" : "Specific write consistency setting for the operation"
+        },
+        "parent": {
+          "type" : "string",
+          "description" : "ID of parent document"
+        },
+        "refresh": {
+          "type" : "boolean",
+          "description" : "Refresh the index after performing the operation"
+        },
+        "replication": {
+          "type" : "enum",
+          "options" : ["sync","async"],
+          "default" : "sync",
+          "description" : "Specific replication type"
+        },
+        "routing": {
+          "type" : "string",
+          "description" : "Specific routing value"
+        },
+        "timeout": {
+          "type" : "time",
+          "description" : "Explicit operation timeout"
+        },
+        "version" : {
+          "type" : "number",
+          "description" : "Explicit version number for concurrency control"
+        },
+        "version_type": {
+          "type" : "enum",
+          "options" : ["internal", "external", "external_gte", "force"],
+          "description" : "Specific version type"
+        }
+      }
+    },
+    "body": null
+  }
+}

rest-api-spec/api/delete_by_query.json

@@ -0,0 +1,81 @@
+{
+  "delete_by_query": {
+    "documentation": "This file is copied from es core just to verify that the .watches api hijacking works",
+    "methods": ["DELETE"],
+    "url": {
+      "path": "/{index}/_query",
+      "paths": ["/{index}/_query", "/{index}/{type}/_query"],
+      "parts": {
+        "index": {
+          "type" : "list",
+          "required": true,
+          "description" : "A comma-separated list of indices to restrict the operation; use `_all` to perform the operation on all indices"
+        },
+        "type": {
+          "type" : "list",
+          "description" : "A comma-separated list of types to restrict the operation"
+        }
+      },
+      "params": {
+        "analyzer": {
+          "type" : "string",
+          "description" : "The analyzer to use for the query string"
+        },
+        "consistency": {
+          "type" : "enum",
+          "options" : ["one", "quorum", "all"],
+          "description" : "Specific write consistency setting for the operation"
+        },
+        "default_operator": {
+          "type" : "enum",
+          "options" : ["AND","OR"],
+          "default" : "OR",
+          "description" : "The default operator for query string query (AND or OR)"
+        },
+        "df": {
+          "type" : "string",
+          "description" : "The field to use as default where no field prefix is given in the query string"
+        },
+        "ignore_unavailable": {
+            "type" : "boolean",
+            "description" : "Whether specified concrete indices should be ignored when unavailable (missing or closed)"
+        },
+        "allow_no_indices": {
+            "type" : "boolean",
+            "description" : "Whether to ignore if a wildcard indices expression resolves into no concrete indices. (This includes `_all` string or when no indices have been specified)"
+        },
+        "expand_wildcards": {
+            "type" : "enum",
+            "options" : ["open","closed","none","all"],
+            "default" : "open",
+            "description" : "Whether to expand wildcard expression to concrete indices that are open, closed or both."
+        },
+        "replication": {
+          "type" : "enum",
+          "options" : ["sync","async"],
+          "default" : "sync",
+          "description" : "Specific replication type"
+        },
+        "q": {
+          "type" : "string",
+          "description" : "Query in the Lucene query string syntax"
+        },
+        "routing": {
+          "type" : "string",
+          "description" : "Specific routing value"
+        },
+        "source": {
+          "type" : "string",
+          "description" : "The URL-encoded query definition (instead of using the request body)"
+        },
+        "timeout": {
+          "type" : "time",
+          "description" : "Explicit operation timeout"
+        }
+      }
+    },
+    "body": {
+      "description" : "A query to restrict the operation specified with the Query DSL"
+    }
+  }
+}

rest-api-spec/api/get.json

@@ -0,0 +1,75 @@
+{
+  "get": {
+    "documentation": "This file is copied from es core just to verify that the .watches api hijacking works",
+    "methods": ["GET"],
+    "url": {
+      "path": "/{index}/{type}/{id}",
+      "paths": ["/{index}/{type}/{id}"],
+      "parts": {
+        "id": {
+          "type" : "string",
+          "required" : true,
+          "description" : "The document ID"
+        },
+        "index": {
+          "type" : "string",
+          "required" : true,
+          "description" : "The name of the index"
+        },
+        "type": {
+          "type" : "string",
+          "required" : true,
+          "description" : "The type of the document (use `_all` to fetch the first document matching the ID across all types)"
+        }
+      },
+      "params": {
+        "fields": {
+          "type": "list",
+          "description" : "A comma-separated list of fields to return in the response"
+        },
+        "parent": {
+          "type" : "string",
+          "description" : "The ID of the parent document"
+        },
+        "preference": {
+          "type" : "string",
+          "description" : "Specify the node or shard the operation should be performed on (default: random)"
+        },
+        "realtime": {
+          "type" : "boolean",
+          "description" : "Specify whether to perform the operation in realtime or search mode"
+        },
+        "refresh": {
+          "type" : "boolean",
+          "description" : "Refresh the shard containing the document before performing the operation"
+        },
+        "routing": {
+          "type" : "string",
+          "description" : "Specific routing value"
+        },
+        "_source": {
+          "type" : "list",
+          "description" : "True or false to return the _source field or not, or a list of fields to return"
+        },
+        "_source_exclude": {
+          "type" : "list",
+          "description" : "A list of fields to exclude from the returned _source field"
+        },
+        "_source_include": {
+          "type" : "list",
+          "description" : "A list of fields to extract and return from the _source field"
+        },
+        "version" : {
+          "type" : "number",
+          "description" : "Explicit version number for concurrency control"
+        },
+        "version_type": {
+          "type" : "enum",
+          "options" : ["internal", "external", "external_gte", "force"],
+          "description" : "Specific version type"
+        }
+      }
+    },
+    "body": null
+  }
+}

rest-api-spec/api/index.json

@@ -0,0 +1,82 @@
+{
+  "index": {
+    "documentation": "This file is copied from es core just to verify that the .watches api hijacking works",
+    "methods": ["POST", "PUT"],
+    "url": {
+      "path": "/{index}/{type}",
+      "paths": ["/{index}/{type}", "/{index}/{type}/{id}"],
+      "parts": {
+        "id": {
+          "type" : "string",
+          "description" : "Document ID"
+        },
+        "index": {
+          "type" : "string",
+          "required" : true,
+          "description" : "The name of the index"
+        },
+        "type": {
+          "type" : "string",
+          "required" : true,
+          "description" : "The type of the document"
+        }
+      },
+      "params": {
+        "consistency": {
+          "type" : "enum",
+          "options" : ["one", "quorum", "all"],
+          "description" : "Explicit write consistency setting for the operation"
+        },
+        "op_type": {
+          "type" : "enum",
+          "options" : ["index", "create"],
+          "default" : "index",
+          "description" : "Explicit operation type"
+        },
+        "parent": {
+          "type" : "string",
+          "description" : "ID of the parent document"
+        },
+        "refresh": {
+          "type" : "boolean",
+          "description" : "Refresh the index after performing the operation"
+        },
+        "replication": {
+          "type" : "enum",
+          "options" : ["sync","async"],
+          "default" : "sync",
+          "description" : "Specific replication type"
+        },
+        "routing": {
+          "type" : "string",
+          "description" : "Specific routing value"
+        },
+        "timeout": {
+          "type" : "time",
+          "description" : "Explicit operation timeout"
+        },
+        "timestamp": {
+          "type" : "time",
+          "description" : "Explicit timestamp for the document"
+        },
+        "ttl": {
+          "type" : "duration",
+          "description" : "Expiration time for the document"
+        },
+        "version" : {
+          "type" : "number",
+          "description" : "Explicit version number for concurrency control"
+        },
+        "version_type": {
+          "type" : "enum",
+          "options" : ["internal", "external", "external_gte", "force"],
+          "description" : "Specific version type"
+        }
+      }
+    },
+    "body": {
+      "description" : "The document",
+      "required" : true
+    }
+  }
+}

rest-api-spec/api/info.json

@@ -1,6 +1,6 @@
 {
   "info": {
-    "documentation": "http://www.elasticsearch.org/guide/",
+    "documentation": "This file is copied from es core because the REST test framework requires it",
     "methods": ["GET"],
     "url": {
       "path": "/",

rest-api-spec/api/update.json

@@ -0,0 +1,98 @@
+{
+  "update": {
+    "documentation": "This file is copied from es core just to verify that the .watches api hijacking works",
+    "methods": ["POST"],
+    "url": {
+      "path": "/{index}/{type}/{id}/_update",
+      "paths": ["/{index}/{type}/{id}/_update"],
+      "parts": {
+        "id": {
+          "type": "string",
+          "required": true,
+          "description": "Document ID"
+        },
+        "index": {
+          "type": "string",
+          "required": true,
+          "description": "The name of the index"
+        },
+        "type": {
+          "type": "string",
+          "required": true,
+          "description": "The type of the document"
+        }
+      },
+      "params": {
+        "consistency": {
+          "type": "enum",
+          "options": ["one", "quorum", "all"],
+          "description": "Explicit write consistency setting for the operation"
+        },
+        "fields": {
+          "type": "list",
+          "description": "A comma-separated list of fields to return in the response"
+        },
+        "lang": {
+          "type": "string",
+          "description": "The script language (default: groovy)"
+        },
+        "parent": {
+          "type": "string",
+          "description": "ID of the parent document"
+        },
+        "refresh": {
+          "type": "boolean",
+          "description": "Refresh the index after performing the operation"
+        },
+        "replication": {
+          "type": "enum",
+          "options": ["sync", "async"],
+          "default": "sync",
+          "description": "Specific replication type"
+        },
+        "retry_on_conflict": {
+          "type": "number",
+          "description": "Specify how many times should the operation be retried when a conflict occurs (default: 0)"
+        },
+        "routing": {
+          "type": "string",
+          "description": "Specific routing value"
+        },
+        "script": {
+          "description": "The URL-encoded script definition (instead of using request body)"
+        },
+        "script_id": {
+          "description": "The id of a stored script"
+        },
+        "scripted_upsert": {
+          "type": "boolean",        
+          "description": "True if the script referenced in script or script_id should be called to perform inserts - defaults to false"
+        },
+        "timeout": {
+          "type": "time",
+          "description": "Explicit operation timeout"
+        },
+        "timestamp": {
+          "type": "time",
+          "description": "Explicit timestamp for the document"
+        },
+        "ttl": {
+          "type": "duration",
+          "description": "Expiration time for the document"
+        },
+        "version": {
+          "type": "number",
+          "description": "Explicit version number for concurrency control"
+        },
+        "version_type": {
+          "type": "enum",
+          "options": ["internal", "force"],
+          "description": "Specific version type"
+        }
+      }
+    },
+    "body": {
+      "description": "The request definition using either `script` or partial `doc`"
+    }
+  }
+}

rest-api-spec/test/hijack/10_basic.yaml

@@ -0,0 +1,48 @@
+---
+"Test Hijack api":
+  - do:
+      cluster.health:
+          wait_for_status: green
+
+  - do:
+      catch: /not supported/
+      get:
+          index: ".watches"
+          type: "watch"
+          id: "foo"
+
+  - do:
+      catch: /not supported/
+      index:
+          index: ".watches"
+          type: "watch"
+          id: "foo"
+          body: {}
+
+  - do:
+      catch: /not supported/
+      delete:
+          index: ".watches"
+          type: "watch"
+          id: "foo"
+
+  - do:
+      catch: /not supported/
+      update:
+          index: ".watches"
+          type: "watch"
+          id: "foo"
+          body: {}
+
+  - do:
+      catch: /not supported/
+      bulk:
+          index: ".watches"
+          type: "watch"
+          body: {}
+
+  - do:
+      catch: /not supported/
+      bulk:
+          index: ".watches"
+          body: {}

src/main/java/org/elasticsearch/watcher/rest/WatcherRestModule.java

@@ -28,6 +28,7 @@ public class WatcherRestModule extends AbstractModule implements PreProcessModul
             restModule.addRestAction(RestWatchServiceAction.class);
             restModule.addRestAction(RestAckWatchAction.class);
             restModule.addRestAction(RestExecuteWatchAction.class);
+            restModule.addRestAction(RestHijackOperationAction.class);
         }
     }
 

src/main/java/org/elasticsearch/watcher/rest/action/RestHijackOperationAction.java

@@ -0,0 +1,71 @@
+/*
+ * 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.watcher.rest.action;
+
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.rest.*;
+import org.elasticsearch.watcher.client.WatcherClient;
+import org.elasticsearch.watcher.rest.WatcherRestHandler;
+import org.elasticsearch.watcher.watch.WatchStore;
+
+/**
+  */
+public class RestHijackOperationAction extends WatcherRestHandler {
+    private static String ALLOW_DIRECT_ACCESS_TO_WATCH_INDEX_SETTING = "watcher.index.rest.direct_access";
+
+
+    @Inject
+    public RestHijackOperationAction(Settings settings, RestController controller, Client client) {
+        super(settings, controller, client);
+        if (!settings.getAsBoolean(ALLOW_DIRECT_ACCESS_TO_WATCH_INDEX_SETTING, false)) {
+            WatcherRestHandler unsupportedHandler = new UnsupportedHandler(settings, controller, client);
+            controller.registerHandler(RestRequest.Method.POST, WatchStore.INDEX + "/watch", this);
+            controller.registerHandler(RestRequest.Method.POST, WatchStore.INDEX + "/watch/{id}", this);
+            controller.registerHandler(RestRequest.Method.PUT, WatchStore.INDEX + "/watch/{id}", this);
+            controller.registerHandler(RestRequest.Method.POST, WatchStore.INDEX + "/watch/{id}/_update", this);
+            controller.registerHandler(RestRequest.Method.DELETE, WatchStore.INDEX + "/watch/_query", this);
+            controller.registerHandler(RestRequest.Method.DELETE, WatchStore.INDEX + "/watch/{id}", this);
+            controller.registerHandler(RestRequest.Method.GET, WatchStore.INDEX + "/watch/{id}", this);
+            controller.registerHandler(RestRequest.Method.POST, WatchStore.INDEX + "/watch/_bulk", unsupportedHandler);
+            controller.registerHandler(RestRequest.Method.POST, WatchStore.INDEX + "/_bulk", unsupportedHandler);
+            controller.registerHandler(RestRequest.Method.PUT, WatchStore.INDEX + "/watch/_bulk", unsupportedHandler);
+            controller.registerHandler(RestRequest.Method.PUT, WatchStore.INDEX + "/_bulk", unsupportedHandler);
+        }
+    }
+
+    @Override
+    protected void handleRequest(RestRequest request, RestChannel channel, WatcherClient client) throws Exception {
+        XContentBuilder jsonBuilder = XContentFactory.jsonBuilder();
+        jsonBuilder.startObject().field("error","This endpoint is not supported for " +
+                request.method().name() + " on " + WatchStore.INDEX + " index. Please use " +
+                request.method().name() + " " + URI_BASE + "/watch/<watch_id> instead");
+        jsonBuilder.field("status", RestStatus.BAD_REQUEST.getStatus());
+        jsonBuilder.endObject();
+        channel.sendResponse(new BytesRestResponse(RestStatus.BAD_REQUEST, jsonBuilder));
+    }
+
+    public static class UnsupportedHandler extends WatcherRestHandler{
+
+        public UnsupportedHandler(Settings settings, RestController controller, Client client) {
+            super(settings, controller, client);
+        }
+
+        @Override
+        protected void handleRequest(RestRequest request, RestChannel channel, WatcherClient client) throws Exception {
+            request.path();
+            XContentBuilder jsonBuilder = XContentFactory.jsonBuilder();
+            jsonBuilder.startObject().field("error","This endpoint is not supported for " +
+                    request.method().name() + " on " + WatchStore.INDEX + " index.");
+            jsonBuilder.field("status", RestStatus.BAD_REQUEST.getStatus());
+            jsonBuilder.endObject();
+            channel.sendResponse(new BytesRestResponse(RestStatus.BAD_REQUEST, jsonBuilder));
+        }
+    }
+}

src/test/java/org/elasticsearch/watcher/test/rest/WatcherDisabledLicenseRestTests.java

@@ -54,6 +54,10 @@ public class WatcherDisabledLicenseRestTests extends WatcherRestTests {
             super.test();
             fail();
         } catch(AssertionError ae) {
+            if (ae.getMessage() == null || ae.getMessage().contains("not supported")){
+                //This was a test testing the "hijacked" methods
+                return;
+            }
             assertThat(ae.getMessage().contains("401 Unauthorized"), is(true));
             assertThat(ae.getMessage().contains(LicenseExpiredException.class.getSimpleName()), is(true));
         }