Merge pull request #8032 from elastic/jasper/backport/7996/4.x

[backport] PR #7996 to 4.x - Configurable headers for all elasticsearch requests

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;
+  }
+};