Backport PR #7996

---------

**Commit 1:**
Configurable headers for all elasticsearch requests

A new server-side configuration, elasticsearch.customHeaders, allows
people to configure any number of custom headers that will get sent
along to all requests to Elasticsearch that are made via the proxy or
exposed client.

This allows for advanced architectures that do things such as dynamic
routing based on install-specific headers.

* Original sha: d00d177
* Authored by Court Ewing <court@epixa.com> on 2016-08-13T16:46:54Z

config/kibana.yml

@@ -56,6 +56,10 @@
 # This must be > 0
 # elasticsearch.requestTimeout: 30000
 
+# Header names and values that are sent to Elasticsearch. Any custom headers cannot be overwritten
+# by client-side headers.
+# elasticsearch.customHeaders: {}
+
 # Time in milliseconds for Elasticsearch to wait for responses from shards.
 # Set to 0 to disable.
 # elasticsearch.shardTimeout: 0

src/plugins/elasticsearch/index.js

@@ -20,6 +20,7 @@ module.exports = function ({ Plugin }) {
         password: string(),
         shardTimeout: number().default(0),
         requestTimeout: number().default(30000),
+        customHeaders: object().default({}),
         pingTimeout: number().default(30000),
         startupTimeout: number().default(5000),
         ssl: object({

src/plugins/elasticsearch/lib/__tests__/map_uri.js

@@ -0,0 +1,47 @@
+import expect from 'expect.js';
+import mapUri from '../map_uri';
+import sinon from 'sinon';
+
+describe('plugins/elasticsearch', function () {
+  describe('lib/map_uri', function () {
+
+    let request;
+
+    beforeEach(function () {
+      request = {
+        path: '/elasticsearch/some/path',
+        headers: {
+          cookie: 'some_cookie_string',
+          'accept-encoding': 'gzip, deflate',
+          origin: 'https://localhost:5601',
+          'content-type': 'application/json',
+          'x-my-custom-header': '42',
+          accept: 'application/json, text/plain, */*',
+          authorization: '2343d322eda344390fdw42'
+        }
+      };
+    });
+
+    it('sends custom headers if set', function () {
+      const get = sinon.stub();
+      get.withArgs('elasticsearch.customHeaders').returns({ foo: 'bar' });
+      const server = { config: () => ({ get }) };
+
+      mapUri(server)(request, function (err, upstreamUri, upstreamHeaders) {
+        expect(err).to.be(null);
+        expect(upstreamHeaders).to.have.property('foo', 'bar');
+      });
+    });
+
+    it('sends configured custom headers even if the same named header exists in request', function () {
+      const get = sinon.stub();
+      get.withArgs('elasticsearch.customHeaders').returns({'x-my-custom-header': 'asconfigured'});
+      const server = { config: () => ({ get }) };
+
+      mapUri(server)(request, function (err, upstreamUri, upstreamHeaders) {
+        expect(err).to.be(null);
+        expect(upstreamHeaders).to.have.property('x-my-custom-header', 'asconfigured');
+      });
+    });
+  });
+});

src/plugins/elasticsearch/lib/__tests__/set_headers.js

@@ -0,0 +1,39 @@
+import expect from 'expect.js';
+import sinon from 'sinon';
+import setHeaders from '../set_headers';
+
+describe('plugins/elasticsearch', function () {
+  describe('lib/set_headers', function () {
+    it('throws if not given an object as the first argument', function () {
+      const fn = () => setHeaders(null, {});
+      expect(fn).to.throwError();
+    });
+
+    it('throws if not given an object as the second argument', function () {
+      const fn = () => setHeaders({}, null);
+      expect(fn).to.throwError();
+    });
+
+    it('returns a new object', function () {
+      const originalHeaders = {};
+      const newHeaders = {};
+      const returnedHeaders = setHeaders(originalHeaders, newHeaders);
+      expect(returnedHeaders).not.to.be(originalHeaders);
+      expect(returnedHeaders).not.to.be(newHeaders);
+    });
+
+    it('returns object with newHeaders merged with originalHeaders', function () {
+      const originalHeaders = { foo: 'bar' };
+      const newHeaders = { one: 'two' };
+      const returnedHeaders = setHeaders(originalHeaders, newHeaders);
+      expect(returnedHeaders).to.eql({ foo: 'bar', one: 'two' });
+    });
+
+    it('returns object where newHeaders takes precedence for any matching keys', function () {
+      const originalHeaders = { foo: 'bar' };
+      const newHeaders = { one: 'two', foo: 'notbar' };
+      const returnedHeaders = setHeaders(originalHeaders, newHeaders);
+      expect(returnedHeaders).to.eql({ foo: 'notbar', one: 'two' });
+    });
+  });
+});

src/plugins/elasticsearch/lib/create_proxy.js

@@ -16,10 +16,12 @@ function createProxy(server, method, route, config) {
     handler: {
       proxy: {
         mapUri: mapUri(server),
-        passThrough: true,
         agent: createAgent(server),
         xforward: true,
-        timeout: server.config().get('elasticsearch.requestTimeout')
+        timeout: server.config().get('elasticsearch.requestTimeout'),
+        onResponse: function (err, responseFromUpstream, request, reply) {
+          reply(err, responseFromUpstream);
+        }
       }
     },
   };

src/plugins/elasticsearch/lib/expose_client.js

@@ -40,9 +40,19 @@ module.exports = function (server) {
       ssl.ca = options.ca.map(readFile);
     }
 
+    const host = {
+      host: uri.hostname,
+      port: uri.port,
+      protocol: uri.protocol,
+      path: uri.pathname,
+      auth: uri.auth,
+      query: uri.query,
+      headers: config.get('elasticsearch.customHeaders')
+    };
+
     return new elasticsearch.Client({
-      host: url.format(uri),
-      ssl: ssl,
+      host,
+      ssl,
       plugins: options.plugins,
       apiVersion: options.apiVersion,
       keepAlive: options.keepAlive,

src/plugins/elasticsearch/lib/map_uri.js

@@ -1,6 +1,9 @@
-const querystring = require('querystring');
-const resolve = require('url').resolve;
-module.exports = function mapUri(server, prefix) {
+import querystring from 'querystring';
+import { resolve } from 'url';
+import setHeaders from './set_headers';
+
+export default function mapUri(server, prefix) {
+
   const config = server.config();
   return function (request, done) {
     const path = request.path.replace('/elasticsearch', '');
@@ -11,6 +14,7 @@ module.exports = function mapUri(server, prefix) {
     }
     const query = querystring.stringify(request.query);
     if (query) url += '?' + query;
-    done(null, url);
+    const customHeaders = setHeaders(request.headers, config.get('elasticsearch.customHeaders'));
+    done(null, url, customHeaders);
   };
 };

src/plugins/elasticsearch/lib/set_headers.js

@@ -0,0 +1,15 @@
+import { isPlainObject } from 'lodash';
+
+export default function setHeaders(originalHeaders, newHeaders) {
+  if (!isPlainObject(originalHeaders)) {
+    throw new Error(`Expected originalHeaders to be an object, but ${typeof originalHeaders} given`);
+  }
+  if (!isPlainObject(newHeaders)) {
+    throw new Error(`Expected newHeaders to be an object, but ${typeof newHeaders} given`);
+  }
+
+  return {
+    ...originalHeaders,
+    ...newHeaders
+  };
+}

src/server/config/config.js

@@ -1,28 +1,27 @@
-let Promise = require('bluebird');
-let Joi = require('joi');
-let _ = require('lodash');
-let { zipObject } = require('lodash');
-let override = require('./override');
+import Joi from 'joi';
+import _ from 'lodash';
+import override from './override';
+import unset from './unset';
+
 let pkg = require('requirefrom')('src/utils')('packageJson');
 const clone = require('./deepCloneWithBuffers');
 
 const schema = Symbol('Joi Schema');
-const schemaKeys = Symbol('Schema Extensions');
+const schemaExts = Symbol('Schema Extensions');
 const vals = Symbol('config values');
 const pendingSets = Symbol('Pending Settings');
 
 module.exports = class Config {
   constructor(initialSchema, initialSettings) {
-    this[schemaKeys] = new Map();
-
+    this[schemaExts] = Object.create(null);
     this[vals] = Object.create(null);
-    this[pendingSets] = new Map(_.pairs(clone(initialSettings || {})));
+    this[pendingSets] = _.merge(Object.create(null), initialSettings || {});
 
     if (initialSchema) this.extendSchema(initialSchema);
   }
 
   getPendingSets() {
-    return this[pendingSets];
+    return new Map(_.pairs(this[pendingSets]));
   }
 
   extendSchema(key, extension) {
@@ -36,27 +35,27 @@ module.exports = class Config {
       throw new Error(`Config schema already has key: ${key}`);
     }
 
-    this[schemaKeys].set(key, extension);
+    _.set(this[schemaExts], key, extension);
     this[schema] = null;
 
-    let initialVals = this[pendingSets].get(key);
+    let initialVals = _.get(this[pendingSets], key);
     if (initialVals) {
       this.set(key, initialVals);
-      this[pendingSets].delete(key);
+      unset(this[pendingSets], key);
     } else {
       this._commit(this[vals]);
     }
   }
 
   removeSchema(key) {
-    if (!this[schemaKeys].has(key)) {
+    if (!_.has(this[schemaExts], key)) {
       throw new TypeError(`Unknown schema key: ${key}`);
     }
 
     this[schema] = null;
-    this[schemaKeys].delete(key);
-    this[pendingSets].delete(key);
-    delete this[vals][key];
+    unset(this[schemaExts], key);
+    unset(this[pendingSets], key);
+    unset(this[vals], key);
   }
 
   resetTo(obj) {
@@ -133,7 +132,7 @@ module.exports = class Config {
       // Catch the partial paths
       if (path.join('.') === key) return true;
       // Only go deep on inner objects with children
-      if (schema._inner.children.length) {
+      if (_.size(schema._inner.children)) {
         for (let i = 0; i < schema._inner.children.length; i++) {
           let child = schema._inner.children[i];
           // If the child is an object recurse through it's children and return
@@ -158,8 +157,22 @@ module.exports = class Config {
 
   getSchema() {
     if (!this[schema]) {
-      let objKeys = zipObject([...this[schemaKeys]]);
-      this[schema] = Joi.object().keys(objKeys).default();
+      this[schema] = (function convertToSchema(children) {
+        let schema = Joi.object().keys({}).default();
+
+        for (const key of Object.keys(children)) {
+          const child = children[key];
+          const childSchema = _.isPlainObject(child) ? convertToSchema(child) : child;
+
+          if (!childSchema || !childSchema.isJoi) {
+            throw new TypeError('Unable to convert configuration definition value to Joi schema: ' + childSchema);
+          }
+
+          schema = schema.keys({ [key]: childSchema });
+        }
+
+        return schema;
+      }(this[schemaExts]));
     }
 
     return this[schema];

src/server/config/unset.js

@@ -0,0 +1,26 @@
+import _ from 'lodash';
+import toPath from 'lodash/internal/toPath';
+
+module.exports = function unset(object, rawPath) {
+  if (!object) return;
+  const path = toPath(rawPath);
+
+  switch (path.length) {
+    case 0:
+      return;
+
+    case 1:
+      delete object[rawPath];
+      break;
+
+    default:
+      const leaf = path.pop();
+      const parentPath = path.slice();
+      const parent = _.get(object, parentPath);
+      unset(parent, leaf);
+      if (!_.size(parent)) {
+        unset(object, parentPath);
+      }
+      break;
+  }
+};