Introduce Enroll API endpoint. (#108835)

src/plugins/interactive_setup/common/elasticsearch_connection_status.ts

@@ -10,11 +10,6 @@
  * Describes current status of the Elasticsearch connection.
  */
 export enum ElasticsearchConnectionStatus {
-  /**
-   * Indicates that Kibana hasn't figured out yet if existing Elasticsearch connection configuration is valid.
-   */
-  Unknown = 'unknown',
-
   /**
    * Indicates that current Elasticsearch connection configuration valid and sufficient.
    */

src/plugins/interactive_setup/server/config.test.ts

@@ -0,0 +1,43 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { ConfigSchema } from './config';
+
+describe('config schema', () => {
+  it('generates proper defaults', () => {
+    expect(ConfigSchema.validate({})).toMatchInlineSnapshot(`
+      Object {
+        "connectionCheck": Object {
+          "interval": "PT5S",
+        },
+        "enabled": false,
+      }
+    `);
+  });
+
+  describe('#connectionCheck', () => {
+    it('should properly set required connection check interval', () => {
+      expect(ConfigSchema.validate({ connectionCheck: { interval: '1s' } })).toMatchInlineSnapshot(`
+        Object {
+          "connectionCheck": Object {
+            "interval": "PT1S",
+          },
+          "enabled": false,
+        }
+      `);
+    });
+
+    it('should throw error if interactiveSetup.connectionCheck.interval is less than 1 second', () => {
+      expect(() =>
+        ConfigSchema.validate({ connectionCheck: { interval: 100 } })
+      ).toThrowErrorMatchingInlineSnapshot(
+        `"[connectionCheck.interval]: the value must be greater or equal to 1 second."`
+      );
+    });
+  });
+});

src/plugins/interactive_setup/server/config.ts

@@ -13,4 +13,14 @@ export type ConfigType = TypeOf<typeof ConfigSchema>;
 
 export const ConfigSchema = schema.object({
   enabled: schema.boolean({ defaultValue: false }),
+  connectionCheck: schema.object({
+    interval: schema.duration({
+      defaultValue: '5s',
+      validate(value) {
+        if (value.asSeconds() < 1) {
+          return 'the value must be greater or equal to 1 second.';
+        }
+      },
+    }),
+  }),
 });

src/plugins/interactive_setup/server/elasticsearch_service.mock.ts

@@ -0,0 +1,20 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { BehaviorSubject } from 'rxjs';
+
+import { ElasticsearchConnectionStatus } from '../common';
+
+export const elasticsearchServiceMock = {
+  createSetup: () => ({
+    connectionStatus$: new BehaviorSubject<ElasticsearchConnectionStatus>(
+      ElasticsearchConnectionStatus.Configured
+    ),
+    enroll: jest.fn(),
+  }),
+};

src/plugins/interactive_setup/server/elasticsearch_service.test.ts

@@ -0,0 +1,497 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { errors } from '@elastic/elasticsearch';
+
+import { nextTick } from '@kbn/test/jest';
+import { elasticsearchServiceMock, loggingSystemMock } from 'src/core/server/mocks';
+
+import { ElasticsearchConnectionStatus } from '../common';
+import { ConfigSchema } from './config';
+import type { ElasticsearchServiceSetup } from './elasticsearch_service';
+import { ElasticsearchService } from './elasticsearch_service';
+import { interactiveSetupMock } from './mocks';
+
+describe('ElasticsearchService', () => {
+  let service: ElasticsearchService;
+  let mockElasticsearchPreboot: ReturnType<typeof elasticsearchServiceMock.createPreboot>;
+  beforeEach(() => {
+    service = new ElasticsearchService(loggingSystemMock.createLogger());
+    mockElasticsearchPreboot = elasticsearchServiceMock.createPreboot();
+  });
+
+  describe('#setup()', () => {
+    let mockConnectionStatusClient: ReturnType<
+      typeof elasticsearchServiceMock.createCustomClusterClient
+    >;
+    let mockEnrollClient: ReturnType<typeof elasticsearchServiceMock.createCustomClusterClient>;
+    let mockAuthenticateClient: ReturnType<
+      typeof elasticsearchServiceMock.createCustomClusterClient
+    >;
+    let setupContract: ElasticsearchServiceSetup;
+    beforeEach(() => {
+      mockConnectionStatusClient = elasticsearchServiceMock.createCustomClusterClient();
+      mockEnrollClient = elasticsearchServiceMock.createCustomClusterClient();
+      mockAuthenticateClient = elasticsearchServiceMock.createCustomClusterClient();
+      mockElasticsearchPreboot.createClient.mockImplementation((type) => {
+        switch (type) {
+          case 'enroll':
+            return mockEnrollClient;
+          case 'authenticate':
+            return mockAuthenticateClient;
+          default:
+            return mockConnectionStatusClient;
+        }
+      });
+
+      setupContract = service.setup({
+        elasticsearch: mockElasticsearchPreboot,
+        connectionCheckInterval: ConfigSchema.validate({}).connectionCheck.interval,
+      });
+    });
+
+    describe('#connectionStatus$', () => {
+      beforeEach(() => jest.useFakeTimers());
+      afterEach(() => jest.useRealTimers());
+
+      it('does not repeat ping request if have multiple subscriptions', async () => {
+        mockConnectionStatusClient.asInternalUser.ping.mockRejectedValue(
+          new errors.ConnectionError(
+            'some-message',
+            interactiveSetupMock.createApiResponse({ body: {} })
+          )
+        );
+
+        const mockHandler1 = jest.fn();
+        const mockHandler2 = jest.fn();
+        setupContract.connectionStatus$.subscribe(mockHandler1);
+        setupContract.connectionStatus$.subscribe(mockHandler2);
+
+        jest.advanceTimersByTime(0);
+        await nextTick();
+
+        // Late subscription.
+        const mockHandler3 = jest.fn();
+        setupContract.connectionStatus$.subscribe(mockHandler3);
+
+        jest.advanceTimersByTime(100);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(1);
+        expect(mockHandler1).toHaveBeenCalledTimes(1);
+        expect(mockHandler1).toHaveBeenCalledWith(ElasticsearchConnectionStatus.NotConfigured);
+        expect(mockHandler2).toHaveBeenCalledTimes(1);
+        expect(mockHandler2).toHaveBeenCalledWith(ElasticsearchConnectionStatus.NotConfigured);
+        expect(mockHandler3).toHaveBeenCalledTimes(1);
+        expect(mockHandler3).toHaveBeenCalledWith(ElasticsearchConnectionStatus.NotConfigured);
+      });
+
+      it('does not report the same status twice', async () => {
+        mockConnectionStatusClient.asInternalUser.ping.mockRejectedValue(
+          new errors.ConnectionError(
+            'some-message',
+            interactiveSetupMock.createApiResponse({ body: {} })
+          )
+        );
+
+        const mockHandler = jest.fn();
+        setupContract.connectionStatus$.subscribe(mockHandler);
+
+        jest.advanceTimersByTime(0);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledWith(ElasticsearchConnectionStatus.NotConfigured);
+
+        mockHandler.mockClear();
+
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(2);
+        expect(mockHandler).not.toHaveBeenCalled();
+
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(3);
+        expect(mockHandler).not.toHaveBeenCalled();
+      });
+
+      it('stops status checks as soon as connection is known to be configured', async () => {
+        mockConnectionStatusClient.asInternalUser.ping.mockRejectedValue(
+          new errors.ConnectionError(
+            'some-message',
+            interactiveSetupMock.createApiResponse({ body: {} })
+          )
+        );
+
+        const mockHandler = jest.fn();
+        setupContract.connectionStatus$.subscribe(mockHandler);
+
+        jest.advanceTimersByTime(0);
+        await nextTick();
+
+        // Initial ping (connection error).
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledWith(ElasticsearchConnectionStatus.NotConfigured);
+
+        // Repeated ping (Unauthorized error).
+        mockConnectionStatusClient.asInternalUser.ping.mockRejectedValue(
+          new errors.ResponseError(
+            interactiveSetupMock.createApiResponse({ statusCode: 401, body: {} })
+          )
+        );
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(2);
+        expect(mockHandler).toHaveBeenCalledTimes(2);
+        expect(mockHandler).toHaveBeenCalledWith(ElasticsearchConnectionStatus.Configured);
+
+        mockHandler.mockClear();
+        mockConnectionStatusClient.asInternalUser.ping.mockClear();
+
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).not.toHaveBeenCalled();
+        expect(mockHandler).not.toHaveBeenCalled();
+      });
+
+      it('checks connection status only once if connection is known to be configured right from start', async () => {
+        mockConnectionStatusClient.asInternalUser.ping.mockResolvedValue(
+          interactiveSetupMock.createApiResponse({ body: true })
+        );
+
+        const mockHandler = jest.fn();
+        setupContract.connectionStatus$.subscribe(mockHandler);
+
+        jest.advanceTimersByTime(0);
+        await nextTick();
+
+        // Initial ping (connection error).
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledWith(ElasticsearchConnectionStatus.Configured);
+
+        mockHandler.mockClear();
+        mockConnectionStatusClient.asInternalUser.ping.mockClear();
+
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).not.toHaveBeenCalled();
+        expect(mockHandler).not.toHaveBeenCalled();
+
+        const mockHandler2 = jest.fn();
+        setupContract.connectionStatus$.subscribe(mockHandler2);
+
+        // Source observable is complete, and handler should be called immediately.
+        expect(mockHandler2).toHaveBeenCalledTimes(1);
+        expect(mockHandler2).toHaveBeenCalledWith(ElasticsearchConnectionStatus.Configured);
+
+        mockHandler2.mockClear();
+
+        // No status check should be made after the first attempt.
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).not.toHaveBeenCalled();
+        expect(mockHandler).not.toHaveBeenCalled();
+        expect(mockHandler2).not.toHaveBeenCalled();
+      });
+
+      it('does not check connection status if there are no subscribers', async () => {
+        mockConnectionStatusClient.asInternalUser.ping.mockRejectedValue(
+          new errors.ConnectionError(
+            'some-message',
+            interactiveSetupMock.createApiResponse({ body: {} })
+          )
+        );
+
+        const mockHandler = jest.fn();
+        const mockSubscription = setupContract.connectionStatus$.subscribe(mockHandler);
+
+        jest.advanceTimersByTime(0);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledWith(ElasticsearchConnectionStatus.NotConfigured);
+
+        mockSubscription.unsubscribe();
+        mockHandler.mockClear();
+        mockConnectionStatusClient.asInternalUser.ping.mockClear();
+
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).not.toHaveBeenCalled();
+        expect(mockHandler).not.toHaveBeenCalled();
+
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).not.toHaveBeenCalled();
+        expect(mockHandler).not.toHaveBeenCalled();
+      });
+
+      it('treats non-connection errors the same as successful response', async () => {
+        mockConnectionStatusClient.asInternalUser.ping.mockRejectedValue(
+          new errors.ResponseError(
+            interactiveSetupMock.createApiResponse({ statusCode: 401, body: {} })
+          )
+        );
+
+        const mockHandler = jest.fn();
+        setupContract.connectionStatus$.subscribe(mockHandler);
+
+        jest.advanceTimersByTime(0);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledWith(ElasticsearchConnectionStatus.Configured);
+
+        mockHandler.mockClear();
+        mockConnectionStatusClient.asInternalUser.ping.mockClear();
+
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).not.toHaveBeenCalled();
+        expect(mockHandler).not.toHaveBeenCalled();
+      });
+
+      it('treats product check error the same as successful response', async () => {
+        mockConnectionStatusClient.asInternalUser.ping.mockRejectedValue(
+          new errors.ProductNotSupportedError(interactiveSetupMock.createApiResponse({ body: {} }))
+        );
+
+        const mockHandler = jest.fn();
+        setupContract.connectionStatus$.subscribe(mockHandler);
+
+        jest.advanceTimersByTime(0);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledTimes(1);
+        expect(mockHandler).toHaveBeenCalledWith(ElasticsearchConnectionStatus.Configured);
+
+        mockHandler.mockClear();
+        mockConnectionStatusClient.asInternalUser.ping.mockClear();
+
+        jest.advanceTimersByTime(5000);
+        await nextTick();
+
+        expect(mockConnectionStatusClient.asInternalUser.ping).not.toHaveBeenCalled();
+        expect(mockHandler).not.toHaveBeenCalled();
+      });
+    });
+
+    describe('#enroll()', () => {
+      it('fails if enroll call fails', async () => {
+        const mockScopedClusterClient = elasticsearchServiceMock.createScopedClusterClient();
+        mockScopedClusterClient.asCurrentUser.transport.request.mockRejectedValue(
+          new errors.ResponseError(
+            interactiveSetupMock.createApiResponse({ statusCode: 401, body: { message: 'oh no' } })
+          )
+        );
+        mockEnrollClient.asScoped.mockReturnValue(mockScopedClusterClient);
+
+        await expect(
+          setupContract.enroll({ apiKey: 'apiKey', hosts: ['host1'] })
+        ).rejects.toMatchInlineSnapshot(`[ResponseError: {"message":"oh no"}]`);
+
+        expect(mockEnrollClient.asScoped).toHaveBeenCalledTimes(1);
+        expect(mockEnrollClient.close).toHaveBeenCalledTimes(1);
+        expect(mockAuthenticateClient.asInternalUser.security.authenticate).not.toHaveBeenCalled();
+      });
+
+      it('fails if none of the hosts are accessible', async () => {
+        const mockScopedClusterClient = elasticsearchServiceMock.createScopedClusterClient();
+        mockScopedClusterClient.asCurrentUser.transport.request.mockRejectedValue(
+          new errors.ConnectionError(
+            'some-message',
+            interactiveSetupMock.createApiResponse({ body: {} })
+          )
+        );
+        mockEnrollClient.asScoped.mockReturnValue(mockScopedClusterClient);
+
+        await expect(
+          setupContract.enroll({ apiKey: 'apiKey', hosts: ['host1', 'host2'] })
+        ).rejects.toMatchInlineSnapshot(`[Error: Unable to connect to any of the provided hosts.]`);
+
+        expect(mockEnrollClient.close).toHaveBeenCalledTimes(2);
+        expect(mockAuthenticateClient.asInternalUser.security.authenticate).not.toHaveBeenCalled();
+      });
+
+      it('fails if authenticate call fails', async () => {
+        const mockEnrollScopedClusterClient = elasticsearchServiceMock.createScopedClusterClient();
+        mockEnrollScopedClusterClient.asCurrentUser.transport.request.mockResolvedValue(
+          interactiveSetupMock.createApiResponse({
+            statusCode: 200,
+            body: { token: { name: 'some-name', value: 'some-value' }, http_ca: 'some-ca' },
+          })
+        );
+        mockEnrollClient.asScoped.mockReturnValue(mockEnrollScopedClusterClient);
+
+        mockAuthenticateClient.asInternalUser.security.authenticate.mockRejectedValue(
+          new errors.ResponseError(
+            interactiveSetupMock.createApiResponse({ statusCode: 401, body: { message: 'oh no' } })
+          )
+        );
+
+        await expect(
+          setupContract.enroll({ apiKey: 'apiKey', hosts: ['host1'] })
+        ).rejects.toMatchInlineSnapshot(`[ResponseError: {"message":"oh no"}]`);
+
+        expect(mockEnrollClient.asScoped).toHaveBeenCalledTimes(1);
+        expect(mockEnrollClient.close).toHaveBeenCalledTimes(1);
+        expect(mockAuthenticateClient.asInternalUser.security.authenticate).toHaveBeenCalledTimes(
+          1
+        );
+        expect(mockAuthenticateClient.close).toHaveBeenCalledTimes(1);
+      });
+
+      it('iterates through all provided hosts until find an accessible one', async () => {
+        mockElasticsearchPreboot.createClient.mockClear();
+
+        const mockHostOneEnrollScopedClusterClient = elasticsearchServiceMock.createScopedClusterClient();
+        mockHostOneEnrollScopedClusterClient.asCurrentUser.transport.request.mockRejectedValue(
+          new errors.ConnectionError(
+            'some-message',
+            interactiveSetupMock.createApiResponse({ body: {} })
+          )
+        );
+
+        const mockHostTwoEnrollScopedClusterClient = elasticsearchServiceMock.createScopedClusterClient();
+        mockHostTwoEnrollScopedClusterClient.asCurrentUser.transport.request.mockResolvedValue(
+          interactiveSetupMock.createApiResponse({
+            statusCode: 200,
+            body: {
+              token: { name: 'some-name', value: 'some-value' },
+              http_ca: '\n\nsome weird-ca_with\n content\n\n',
+            },
+          })
+        );
+
+        mockEnrollClient.asScoped
+          .mockReturnValueOnce(mockHostOneEnrollScopedClusterClient)
+          .mockReturnValueOnce(mockHostTwoEnrollScopedClusterClient);
+
+        mockAuthenticateClient.asInternalUser.security.authenticate.mockResolvedValue(
+          interactiveSetupMock.createApiResponse({ statusCode: 200, body: {} as any })
+        );
+
+        const expectedCa = `-----BEGIN CERTIFICATE-----
+
+
+some weird+ca/with
+
+ content
+
+
+-----END CERTIFICATE-----
+`;
+
+        await expect(
+          setupContract.enroll({ apiKey: 'apiKey', hosts: ['host1', 'host2'] })
+        ).resolves.toEqual({
+          ca: expectedCa,
+          host: 'host2',
+          serviceAccountToken: {
+            name: 'some-name',
+            value: 'some-value',
+          },
+        });
+
+        // Check that we created clients with the right parameters
+        expect(mockElasticsearchPreboot.createClient).toHaveBeenCalledTimes(3);
+        expect(mockElasticsearchPreboot.createClient).toHaveBeenCalledWith('enroll', {
+          hosts: ['host1'],
+          ssl: { verificationMode: 'none' },
+        });
+        expect(mockElasticsearchPreboot.createClient).toHaveBeenCalledWith('enroll', {
+          hosts: ['host2'],
+          ssl: { verificationMode: 'none' },
+        });
+        expect(mockElasticsearchPreboot.createClient).toHaveBeenCalledWith('authenticate', {
+          hosts: ['host2'],
+          serviceAccountToken: 'some-value',
+          ssl: { certificateAuthorities: [expectedCa] },
+        });
+
+        // Check that we properly provided apiKeys to scoped clients.
+        expect(mockEnrollClient.asScoped).toHaveBeenCalledTimes(2);
+        expect(mockEnrollClient.asScoped).toHaveBeenNthCalledWith(1, {
+          headers: { authorization: 'ApiKey apiKey' },
+        });
+        expect(mockEnrollClient.asScoped).toHaveBeenNthCalledWith(2, {
+          headers: { authorization: 'ApiKey apiKey' },
+        });
+
+        // Check that we properly called all required ES APIs.
+        expect(
+          mockHostOneEnrollScopedClusterClient.asCurrentUser.transport.request
+        ).toHaveBeenCalledTimes(1);
+        expect(
+          mockHostOneEnrollScopedClusterClient.asCurrentUser.transport.request
+        ).toHaveBeenCalledWith({
+          method: 'GET',
+          path: '/_security/enroll/kibana',
+        });
+        expect(
+          mockHostTwoEnrollScopedClusterClient.asCurrentUser.transport.request
+        ).toHaveBeenCalledTimes(1);
+        expect(
+          mockHostTwoEnrollScopedClusterClient.asCurrentUser.transport.request
+        ).toHaveBeenCalledWith({
+          method: 'GET',
+          path: '/_security/enroll/kibana',
+        });
+        expect(mockAuthenticateClient.asInternalUser.security.authenticate).toHaveBeenCalledTimes(
+          1
+        );
+
+        // Check that we properly closed all clients.
+        expect(mockEnrollClient.close).toHaveBeenCalledTimes(2);
+        expect(mockAuthenticateClient.close).toHaveBeenCalledTimes(1);
+      });
+    });
+  });
+
+  describe('#stop()', () => {
+    it('does not fail if called before `setup`', () => {
+      expect(() => service.stop()).not.toThrow();
+    });
+
+    it('closes connection status check client', async () => {
+      const mockConnectionStatusClient = elasticsearchServiceMock.createCustomClusterClient();
+      mockElasticsearchPreboot.createClient.mockImplementation((type) => {
+        switch (type) {
+          case 'ping':
+            return mockConnectionStatusClient;
+          default:
+            throw new Error(`Unexpected client type: ${type}`);
+        }
+      });
+
+      service.setup({
+        elasticsearch: mockElasticsearchPreboot,
+        connectionCheckInterval: ConfigSchema.validate({}).connectionCheck.interval,
+      });
+      service.stop();
+
+      expect(mockConnectionStatusClient.close).toHaveBeenCalled();
+    });
+  });
+});

src/plugins/interactive_setup/server/elasticsearch_service.ts

@@ -0,0 +1,239 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import type { ApiResponse } from '@elastic/elasticsearch';
+import { errors } from '@elastic/elasticsearch';
+import type { Duration } from 'moment';
+import type { Observable } from 'rxjs';
+import { from, of, timer } from 'rxjs';
+import {
+  catchError,
+  distinctUntilChanged,
+  exhaustMap,
+  map,
+  shareReplay,
+  takeWhile,
+} from 'rxjs/operators';
+
+import type {
+  ElasticsearchClientConfig,
+  ElasticsearchServicePreboot,
+  ICustomClusterClient,
+  Logger,
+  ScopeableRequest,
+} from 'src/core/server';
+
+import { ElasticsearchConnectionStatus } from '../common';
+import { getDetailedErrorMessage } from './errors';
+
+interface EnrollParameters {
+  apiKey: string;
+  hosts: string[];
+  // TODO: Integrate fingerprint check as soon core supports this new option:
+  // https://github.com/elastic/kibana/pull/108514
+  caFingerprint?: string;
+}
+
+export interface ElasticsearchServiceSetupDeps {
+  /**
+   * Core Elasticsearch service preboot contract;
+   */
+  elasticsearch: ElasticsearchServicePreboot;
+
+  /**
+   * Interval for the Elasticsearch connection check (whether it's configured or not).
+   */
+  connectionCheckInterval: Duration;
+}
+
+export interface ElasticsearchServiceSetup {
+  /**
+   * Observable that yields the last result of the Elasticsearch connection status check.
+   */
+  connectionStatus$: Observable<ElasticsearchConnectionStatus>;
+
+  /**
+   * Iterates through provided {@param hosts} one by one trying to call Kibana enrollment API using
+   * the specified {@param apiKey}.
+   * @param apiKey The ApiKey to use to authenticate Kibana enrollment request.
+   * @param hosts The list of Elasticsearch node addresses to enroll with. The addresses are supposed
+   * to point to exactly same Elasticsearch node, potentially available via different network interfaces.
+   */
+  enroll: (params: EnrollParameters) => Promise<EnrollResult>;
+}
+
+/**
+ * Result of the enrollment request.
+ */
+export interface EnrollResult {
+  /**
+   * Host address of the Elasticsearch node that successfully processed enrollment request.
+   */
+  host: string;
+  /**
+   * PEM CA certificate for the Elasticsearch HTTP certificates.
+   */
+  ca: string;
+  /**
+   * Service account token for the "elastic/kibana" service account.
+   */
+  serviceAccountToken: { name: string; value: string };
+}
+
+export class ElasticsearchService {
+  /**
+   * Elasticsearch client used to check Elasticsearch connection status.
+   */
+  private connectionStatusClient?: ICustomClusterClient;
+  constructor(private readonly logger: Logger) {}
+
+  public setup({
+    elasticsearch,
+    connectionCheckInterval,
+  }: ElasticsearchServiceSetupDeps): ElasticsearchServiceSetup {
+    const connectionStatusClient = (this.connectionStatusClient = elasticsearch.createClient(
+      'ping'
+    ));
+
+    return {
+      connectionStatus$: timer(0, connectionCheckInterval.asMilliseconds()).pipe(
+        exhaustMap(() => {
+          return from(connectionStatusClient.asInternalUser.ping()).pipe(
+            map(() => ElasticsearchConnectionStatus.Configured),
+            catchError((pingError) =>
+              of(
+                pingError instanceof errors.ConnectionError
+                  ? ElasticsearchConnectionStatus.NotConfigured
+                  : ElasticsearchConnectionStatus.Configured
+              )
+            )
+          );
+        }),
+        takeWhile(
+          (status) => status !== ElasticsearchConnectionStatus.Configured,
+          /* inclusive */ true
+        ),
+        distinctUntilChanged(),
+        shareReplay({ refCount: true, bufferSize: 1 })
+      ),
+      enroll: this.enroll.bind(this, elasticsearch),
+    };
+  }
+
+  public stop() {
+    if (this.connectionStatusClient) {
+      this.connectionStatusClient.close().catch((err) => {
+        this.logger.debug(`Failed to stop Elasticsearch service: ${getDetailedErrorMessage(err)}`);
+      });
+      this.connectionStatusClient = undefined;
+    }
+  }
+
+  /**
+   * Iterates through provided {@param hosts} one by one trying to call Kibana enrollment API using
+   * the specified {@param apiKey}.
+   * @param elasticsearch Core Elasticsearch service preboot contract.
+   * @param apiKey The ApiKey to use to authenticate Kibana enrollment request.
+   * @param hosts The list of Elasticsearch node addresses to enroll with. The addresses are supposed
+   * to point to exactly same Elasticsearch node, potentially available via different network interfaces.
+   */
+  private async enroll(
+    elasticsearch: ElasticsearchServicePreboot,
+    { apiKey, hosts }: EnrollParameters
+  ): Promise<EnrollResult> {
+    const scopeableRequest: ScopeableRequest = { headers: { authorization: `ApiKey ${apiKey}` } };
+    const elasticsearchConfig: Partial<ElasticsearchClientConfig> = {
+      ssl: { verificationMode: 'none' },
+    };
+
+    // We should iterate through all provided hosts until we find an accessible one.
+    for (const host of hosts) {
+      this.logger.debug(`Trying to enroll with "${host}" host`);
+      const enrollClient = elasticsearch.createClient('enroll', {
+        ...elasticsearchConfig,
+        hosts: [host],
+      });
+
+      let enrollmentResponse;
+      try {
+        enrollmentResponse = (await enrollClient
+          .asScoped(scopeableRequest)
+          .asCurrentUser.transport.request({
+            method: 'GET',
+            path: '/_security/enroll/kibana',
+          })) as ApiResponse<{ token: { name: string; value: string }; http_ca: string }>;
+      } catch (err) {
+        // We expect that all hosts belong to exactly same node and any non-connection error for one host would mean
+        // that enrollment will fail for any other host and we should bail out.
+        if (err instanceof errors.ConnectionError || err instanceof errors.TimeoutError) {
+          this.logger.error(
+            `Unable to connect to "${host}" host, will proceed to the next host if available: ${getDetailedErrorMessage(
+              err
+            )}`
+          );
+          continue;
+        }
+
+        this.logger.error(`Failed to enroll with "${host}" host: ${getDetailedErrorMessage(err)}`);
+        throw err;
+      } finally {
+        await enrollClient.close();
+      }
+
+      this.logger.debug(
+        `Successfully enrolled with "${host}" host, token name: ${enrollmentResponse.body.token.name}, CA certificate: ${enrollmentResponse.body.http_ca}`
+      );
+
+      const enrollResult = {
+        host,
+        ca: ElasticsearchService.createPemCertificate(enrollmentResponse.body.http_ca),
+        serviceAccountToken: enrollmentResponse.body.token,
+      };
+
+      // Now try to use retrieved password and CA certificate to authenticate to this host.
+      const authenticateClient = elasticsearch.createClient('authenticate', {
+        hosts: [host],
+        serviceAccountToken: enrollResult.serviceAccountToken.value,
+        ssl: { certificateAuthorities: [enrollResult.ca] },
+      });
+
+      this.logger.debug(
+        `Verifying if "${enrollmentResponse.body.token.name}" token can authenticate to "${host}" host.`
+      );
+
+      try {
+        await authenticateClient.asInternalUser.security.authenticate();
+        this.logger.debug(
+          `Successfully authenticated "${enrollmentResponse.body.token.name}" token to "${host}" host.`
+        );
+      } catch (err) {
+        this.logger.error(
+          `Failed to authenticate "${
+            enrollmentResponse.body.token.name
+          }" token to "${host}" host: ${getDetailedErrorMessage(err)}.`
+        );
+        throw err;
+      } finally {
+        await authenticateClient.close();
+      }
+
+      return enrollResult;
+    }
+
+    throw new Error('Unable to connect to any of the provided hosts.');
+  }
+
+  private static createPemCertificate(derCaString: string) {
+    // Use `X509Certificate` class once we upgrade to Node v16.
+    return `-----BEGIN CERTIFICATE-----\n${derCaString
+      .replace(/_/g, '/')
+      .replace(/-/g, '+')
+      .replace(/([^\n]{1,65})/g, '$1\n')
+      .replace(/\n$/g, '')}\n-----END CERTIFICATE-----\n`;
+  }
+}

src/plugins/interactive_setup/server/errors.test.ts

@@ -0,0 +1,57 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { errors as esErrors } from '@elastic/elasticsearch';
+
+import * as errors from './errors';
+import { interactiveSetupMock } from './mocks';
+
+describe('errors', () => {
+  describe('#getErrorStatusCode', () => {
+    it('extracts status code from Elasticsearch client response error', () => {
+      expect(
+        errors.getErrorStatusCode(
+          new esErrors.ResponseError(
+            interactiveSetupMock.createApiResponse({ statusCode: 400, body: {} })
+          )
+        )
+      ).toBe(400);
+      expect(
+        errors.getErrorStatusCode(
+          new esErrors.ResponseError(
+            interactiveSetupMock.createApiResponse({ statusCode: 401, body: {} })
+          )
+        )
+      ).toBe(401);
+    });
+
+    it('extracts status code from `status` property', () => {
+      expect(errors.getErrorStatusCode({ statusText: 'Bad Request', status: 400 })).toBe(400);
+      expect(errors.getErrorStatusCode({ statusText: 'Unauthorized', status: 401 })).toBe(401);
+    });
+  });
+
+  describe('#getDetailedErrorMessage', () => {
+    it('extracts body from Elasticsearch client response error', () => {
+      expect(
+        errors.getDetailedErrorMessage(
+          new esErrors.ResponseError(
+            interactiveSetupMock.createApiResponse({
+              statusCode: 401,
+              body: { field1: 'value-1', field2: 'value-2' },
+            })
+          )
+        )
+      ).toBe(JSON.stringify({ field1: 'value-1', field2: 'value-2' }));
+    });
+
+    it('extracts `message` property', () => {
+      expect(errors.getDetailedErrorMessage(new Error('some-message'))).toBe('some-message');
+    });
+  });
+});

src/plugins/interactive_setup/server/errors.ts

@@ -0,0 +1,35 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { errors } from '@elastic/elasticsearch';
+
+/**
+ * Extracts error code from Boom and Elasticsearch "native" errors.
+ * @param error Error instance to extract status code from.
+ */
+export function getErrorStatusCode(error: any): number {
+  if (error instanceof errors.ResponseError) {
+    return error.statusCode;
+  }
+
+  return error.statusCode || error.status;
+}
+
+/**
+ * Extracts detailed error message from Boom and Elasticsearch "native" errors. It's supposed to be
+ * only logged on the server side and never returned to the client as it may contain sensitive
+ * information.
+ * @param error Error instance to extract message from.
+ */
+export function getDetailedErrorMessage(error: any): string {
+  if (error instanceof errors.ResponseError) {
+    return JSON.stringify(error.body);
+  }
+
+  return error.message;
+}

src/plugins/interactive_setup/server/kibana_config_writer.mock.ts

@@ -0,0 +1,18 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import type { PublicMethodsOf } from '@kbn/utility-types';
+
+import type { KibanaConfigWriter } from './kibana_config_writer';
+
+export const kibanaConfigWriterMock = {
+  create: (): jest.Mocked<PublicMethodsOf<KibanaConfigWriter>> => ({
+    isConfigWritable: jest.fn().mockResolvedValue(true),
+    writeConfig: jest.fn().mockResolvedValue(undefined),
+  }),
+};

src/plugins/interactive_setup/server/kibana_config_writer.test.ts

@@ -0,0 +1,140 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+jest.mock('fs/promises');
+import { constants } from 'fs';
+
+import { loggingSystemMock } from 'src/core/server/mocks';
+
+import { KibanaConfigWriter } from './kibana_config_writer';
+
+describe('KibanaConfigWriter', () => {
+  let mockFsAccess: jest.Mock;
+  let mockWriteFile: jest.Mock;
+  let mockAppendFile: jest.Mock;
+  let kibanaConfigWriter: KibanaConfigWriter;
+  beforeEach(() => {
+    jest.spyOn(Date, 'now').mockReturnValue(1234);
+
+    const fsMocks = jest.requireMock('fs/promises');
+    mockFsAccess = fsMocks.access;
+    mockWriteFile = fsMocks.writeFile;
+    mockAppendFile = fsMocks.appendFile;
+
+    kibanaConfigWriter = new KibanaConfigWriter(
+      '/some/path/kibana.yml',
+      loggingSystemMock.createLogger()
+    );
+  });
+
+  afterEach(() => jest.resetAllMocks());
+
+  describe('#isConfigWritable()', () => {
+    it('returns `false` if config directory is not writable even if kibana yml is writable', async () => {
+      mockFsAccess.mockImplementation((path, modifier) =>
+        path === '/some/path' && modifier === constants.W_OK ? Promise.reject() : Promise.resolve()
+      );
+
+      await expect(kibanaConfigWriter.isConfigWritable()).resolves.toBe(false);
+    });
+
+    it('returns `false` if kibana yml is NOT writable if even config directory is writable', async () => {
+      mockFsAccess.mockImplementation((path, modifier) =>
+        path === '/some/path/kibana.yml' && modifier === constants.W_OK
+          ? Promise.reject()
+          : Promise.resolve()
+      );
+
+      await expect(kibanaConfigWriter.isConfigWritable()).resolves.toBe(false);
+    });
+
+    it('returns `true` if both kibana yml and config directory are writable', async () => {
+      mockFsAccess.mockResolvedValue(undefined);
+
+      await expect(kibanaConfigWriter.isConfigWritable()).resolves.toBe(true);
+    });
+
+    it('returns `true` even if kibana yml does not exist when config directory is writable', async () => {
+      mockFsAccess.mockImplementation((path) =>
+        path === '/some/path/kibana.yml' ? Promise.reject() : Promise.resolve()
+      );
+
+      await expect(kibanaConfigWriter.isConfigWritable()).resolves.toBe(true);
+    });
+  });
+
+  describe('#writeConfig()', () => {
+    it('throws if cannot write CA file', async () => {
+      mockWriteFile.mockRejectedValue(new Error('Oh no!'));
+
+      await expect(
+        kibanaConfigWriter.writeConfig({
+          ca: 'ca-content',
+          host: '',
+          serviceAccountToken: { name: '', value: '' },
+        })
+      ).rejects.toMatchInlineSnapshot(`[Error: Oh no!]`);
+
+      expect(mockWriteFile).toHaveBeenCalledTimes(1);
+      expect(mockWriteFile).toHaveBeenCalledWith('/some/path/ca_1234.crt', 'ca-content');
+      expect(mockAppendFile).not.toHaveBeenCalled();
+    });
+
+    it('throws if cannot append config to yaml file', async () => {
+      mockAppendFile.mockRejectedValue(new Error('Oh no!'));
+
+      await expect(
+        kibanaConfigWriter.writeConfig({
+          ca: 'ca-content',
+          host: 'some-host',
+          serviceAccountToken: { name: 'some-token', value: 'some-value' },
+        })
+      ).rejects.toMatchInlineSnapshot(`[Error: Oh no!]`);
+
+      expect(mockWriteFile).toHaveBeenCalledTimes(1);
+      expect(mockWriteFile).toHaveBeenCalledWith('/some/path/ca_1234.crt', 'ca-content');
+      expect(mockAppendFile).toHaveBeenCalledTimes(1);
+      expect(mockAppendFile).toHaveBeenCalledWith(
+        '/some/path/kibana.yml',
+        `
+
+# This section was automatically generated during setup (service account token name is "some-token").
+elasticsearch.hosts: [some-host]
+elasticsearch.serviceAccountToken: some-value
+elasticsearch.ssl.certificateAuthorities: [/some/path/ca_1234.crt]
+
+`
+      );
+    });
+
+    it('can successfully write CA certificate and elasticsearch config to the disk', async () => {
+      await expect(
+        kibanaConfigWriter.writeConfig({
+          ca: 'ca-content',
+          host: 'some-host',
+          serviceAccountToken: { name: 'some-token', value: 'some-value' },
+        })
+      ).resolves.toBeUndefined();
+
+      expect(mockWriteFile).toHaveBeenCalledTimes(1);
+      expect(mockWriteFile).toHaveBeenCalledWith('/some/path/ca_1234.crt', 'ca-content');
+      expect(mockAppendFile).toHaveBeenCalledTimes(1);
+      expect(mockAppendFile).toHaveBeenCalledWith(
+        '/some/path/kibana.yml',
+        `
+
+# This section was automatically generated during setup (service account token name is "some-token").
+elasticsearch.hosts: [some-host]
+elasticsearch.serviceAccountToken: some-value
+elasticsearch.ssl.certificateAuthorities: [/some/path/ca_1234.crt]
+
+`
+      );
+    });
+  });
+});

src/plugins/interactive_setup/server/kibana_config_writer.ts

@@ -0,0 +1,93 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { constants } from 'fs';
+import fs from 'fs/promises';
+import yaml from 'js-yaml';
+import path from 'path';
+
+import type { Logger } from 'src/core/server';
+
+import { getDetailedErrorMessage } from './errors';
+
+export interface WriteConfigParameters {
+  host: string;
+  ca: string;
+  serviceAccountToken: { name: string; value: string };
+}
+
+export class KibanaConfigWriter {
+  constructor(private readonly configPath: string, private readonly logger: Logger) {}
+
+  /**
+   * Checks if we can write to the Kibana configuration file and configuration directory.
+   */
+  public async isConfigWritable() {
+    try {
+      // We perform two separate checks here:
+      // 1. If we can write to config directory to add a new CA certificate file and potentially Kibana configuration
+      // file if it doesn't exist for some reason.
+      // 2. If we can write to the Kibana configuration file if it exists.
+      const canWriteToConfigDirectory = fs.access(path.dirname(this.configPath), constants.W_OK);
+      await Promise.all([
+        canWriteToConfigDirectory,
+        fs.access(this.configPath, constants.F_OK).then(
+          () => fs.access(this.configPath, constants.W_OK),
+          () => canWriteToConfigDirectory
+        ),
+      ]);
+      return true;
+    } catch {
+      return false;
+    }
+  }
+
+  /**
+   * Writes Elasticsearch configuration to the disk.
+   * @param params
+   */
+  public async writeConfig(params: WriteConfigParameters) {
+    const caPath = path.join(path.dirname(this.configPath), `ca_${Date.now()}.crt`);
+
+    this.logger.debug(`Writing CA certificate to ${caPath}.`);
+    try {
+      await fs.writeFile(caPath, params.ca);
+      this.logger.debug(`Successfully wrote CA certificate to ${caPath}.`);
+    } catch (err) {
+      this.logger.error(
+        `Failed to write CA certificate to ${caPath}: ${getDetailedErrorMessage(err)}.`
+      );
+      throw err;
+    }
+
+    this.logger.debug(`Writing Elasticsearch configuration to ${this.configPath}.`);
+    try {
+      await fs.appendFile(
+        this.configPath,
+        `\n\n# This section was automatically generated during setup (service account token name is "${
+          params.serviceAccountToken.name
+        }").\n${yaml.safeDump(
+          {
+            'elasticsearch.hosts': [params.host],
+            'elasticsearch.serviceAccountToken': params.serviceAccountToken.value,
+            'elasticsearch.ssl.certificateAuthorities': [caPath],
+          },
+          { flowLevel: 1 }
+        )}\n`
+      );
+      this.logger.debug(`Successfully wrote Elasticsearch configuration to ${this.configPath}.`);
+    } catch (err) {
+      this.logger.error(
+        `Failed to write  Elasticsearch configuration to ${
+          this.configPath
+        }: ${getDetailedErrorMessage(err)}.`
+      );
+      throw err;
+    }
+  }
+}

src/plugins/interactive_setup/server/mocks.ts

@@ -0,0 +1,26 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import type { ApiResponse } from '@elastic/elasticsearch';
+
+function createApiResponseMock<TResponse, TContext>(
+  apiResponse: Pick<ApiResponse<TResponse, TContext>, 'body'> &
+    Partial<Omit<ApiResponse<TResponse, TContext>, 'body'>>
+): ApiResponse<TResponse, TContext> {
+  return {
+    statusCode: null,
+    headers: null,
+    warnings: null,
+    meta: {} as any,
+    ...apiResponse,
+  };
+}
+
+export const interactiveSetupMock = {
+  createApiResponse: createApiResponseMock,
+};

src/plugins/interactive_setup/server/plugin.ts

@@ -13,11 +13,18 @@ import type { CorePreboot, Logger, PluginInitializerContext, PrebootPlugin } fro
 
 import { ElasticsearchConnectionStatus } from '../common';
 import type { ConfigSchema, ConfigType } from './config';
+import { ElasticsearchService } from './elasticsearch_service';
+import { KibanaConfigWriter } from './kibana_config_writer';
 import { defineRoutes } from './routes';
 
 export class UserSetupPlugin implements PrebootPlugin {
   readonly #logger: Logger;
 
+  #elasticsearchConnectionStatusSubscription?: Subscription;
+  readonly #elasticsearch = new ElasticsearchService(
+    this.initializerContext.logger.get('elasticsearch')
+  );
+
   #configSubscription?: Subscription;
   #config?: ConfigType;
   readonly #getConfig = () => {
@@ -27,11 +34,6 @@ export class UserSetupPlugin implements PrebootPlugin {
     return this.#config;
   };
 
-  #elasticsearchConnectionStatus = ElasticsearchConnectionStatus.Unknown;
-  readonly #getElasticsearchConnectionStatus = () => {
-    return this.#elasticsearchConnectionStatus;
-  };
-
   constructor(private readonly initializerContext: PluginInitializerContext) {
     this.#logger = this.initializerContext.logger.get();
   }
@@ -65,45 +67,48 @@ export class UserSetupPlugin implements PrebootPlugin {
       })
     );
 
-    // If preliminary check above indicates that user didn't alter default Elasticsearch connection
-    // details, it doesn't mean Elasticsearch connection isn't configured. There is a chance that they
-    // already disabled security features in Elasticsearch and everything should work by default.
-    // We should check if we can connect to Elasticsearch with default configuration to know if we
-    // need to activate interactive setup. This check can take some time, so we should register our
-    // routes to let interactive setup UI to handle user requests until the check is complete.
-    core.elasticsearch
-      .createClient('ping')
-      .asInternalUser.ping()
-      .then(
-        (pingResponse) => {
-          if (pingResponse.body) {
-            this.#logger.debug(
-              'Kibana is already properly configured to connect to Elasticsearch. Interactive setup mode will not be activated.'
-            );
-            this.#elasticsearchConnectionStatus = ElasticsearchConnectionStatus.Configured;
-            completeSetup({ shouldReloadConfig: false });
-          } else {
-            this.#logger.debug(
-              'Kibana is not properly configured to connect to Elasticsearch. Interactive setup mode will be activated.'
-            );
-            this.#elasticsearchConnectionStatus = ElasticsearchConnectionStatus.NotConfigured;
-          }
-        },
-        () => {
-          // TODO: we should probably react differently to different errors. 401 - credentials aren't correct, etc.
-          // Do we want to constantly ping ES if interactive mode UI isn't active? Just in case user runs Kibana and then
-          // configure Elasticsearch so that it can eventually connect to it without any configuration changes?
-          this.#elasticsearchConnectionStatus = ElasticsearchConnectionStatus.NotConfigured;
+    // If preliminary checks above indicate that user didn't alter default Elasticsearch connection
+    // details, it doesn't mean Elasticsearch connection isn't configured. There is a chance that
+    // user has already disabled security features in Elasticsearch and everything should work by
+    // default. We should check if we can connect to Elasticsearch with default configuration to
+    // know if we need to activate interactive setup. This check can take some time, so we should
+    // register our routes to let interactive setup UI to handle user requests until the check is
+    // complete. Moreover Elasticsearch may be just temporarily unavailable and we should poll its
+    // status until we can connect or use configures connection via interactive setup mode.
+    const elasticsearch = this.#elasticsearch.setup({
+      elasticsearch: core.elasticsearch,
+      connectionCheckInterval: this.#getConfig().connectionCheck.interval,
+    });
+    this.#elasticsearchConnectionStatusSubscription = elasticsearch.connectionStatus$.subscribe(
+      (status) => {
+        if (status === ElasticsearchConnectionStatus.Configured) {
+          this.#logger.debug(
+            'Skipping interactive setup mode since Kibana is already properly configured to connect to Elasticsearch at http://localhost:9200.'
+          );
+          completeSetup({ shouldReloadConfig: false });
+        } else {
+          this.#logger.debug(
+            'Starting interactive setup mode since Kibana cannot to connect to Elasticsearch at http://localhost:9200.'
+          );
         }
-      );
+      }
+    );
+
+    // If possible, try to use `*.dev.yml` config when Kibana is run in development mode.
+    const configPath = this.initializerContext.env.mode.dev
+      ? this.initializerContext.env.configs.find((config) => config.endsWith('.dev.yml')) ??
+        this.initializerContext.env.configs[0]
+      : this.initializerContext.env.configs[0];
 
     core.http.registerRoutes('', (router) => {
       defineRoutes({
         router,
         basePath: core.http.basePath,
         logger: this.#logger.get('routes'),
+        preboot: { ...core.preboot, completeSetup },
+        kibanaConfigWriter: new KibanaConfigWriter(configPath, this.#logger.get('kibana-config')),
+        elasticsearch,
         getConfig: this.#getConfig.bind(this),
-        getElasticsearchConnectionStatus: this.#getElasticsearchConnectionStatus.bind(this),
       });
     });
   }
@@ -115,5 +120,12 @@ export class UserSetupPlugin implements PrebootPlugin {
       this.#configSubscription.unsubscribe();
       this.#configSubscription = undefined;
     }
+
+    if (this.#elasticsearchConnectionStatusSubscription) {
+      this.#elasticsearchConnectionStatusSubscription.unsubscribe();
+      this.#elasticsearchConnectionStatusSubscription = undefined;
+    }
+
+    this.#elasticsearch.stop();
   }
 }

src/plugins/interactive_setup/server/routes/enroll.test.ts

@@ -0,0 +1,305 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { errors } from '@elastic/elasticsearch';
+
+import type { ObjectType } from '@kbn/config-schema';
+import type { IRouter, RequestHandler, RequestHandlerContext, RouteConfig } from 'src/core/server';
+import { kibanaResponseFactory } from 'src/core/server';
+import { httpServerMock } from 'src/core/server/mocks';
+
+import { ElasticsearchConnectionStatus } from '../../common';
+import { interactiveSetupMock } from '../mocks';
+import { defineEnrollRoutes } from './enroll';
+import { routeDefinitionParamsMock } from './index.mock';
+
+describe('Enroll routes', () => {
+  let router: jest.Mocked<IRouter>;
+  let mockRouteParams: ReturnType<typeof routeDefinitionParamsMock.create>;
+  let mockContext: RequestHandlerContext;
+  beforeEach(() => {
+    mockRouteParams = routeDefinitionParamsMock.create();
+    router = mockRouteParams.router;
+
+    mockContext = ({} as unknown) as RequestHandlerContext;
+
+    defineEnrollRoutes(mockRouteParams);
+  });
+
+  describe('#enroll', () => {
+    let routeHandler: RequestHandler<any, any, any>;
+    let routeConfig: RouteConfig<any, any, any, any>;
+
+    beforeEach(() => {
+      const [enrollRouteConfig, enrollRouteHandler] = router.post.mock.calls.find(
+        ([{ path }]) => path === '/internal/interactive_setup/enroll'
+      )!;
+
+      routeConfig = enrollRouteConfig;
+      routeHandler = enrollRouteHandler;
+    });
+
+    it('correctly defines route.', () => {
+      expect(routeConfig.options).toEqual({ authRequired: false });
+
+      const bodySchema = (routeConfig.validate as any).body as ObjectType;
+      expect(() => bodySchema.validate({})).toThrowErrorMatchingInlineSnapshot(
+        `"[hosts]: expected value of type [array] but got [undefined]"`
+      );
+
+      expect(() => bodySchema.validate({ hosts: [] })).toThrowErrorMatchingInlineSnapshot(
+        `"[hosts]: array size is [0], but cannot be smaller than [1]"`
+      );
+      expect(() =>
+        bodySchema.validate({ hosts: ['localhost:9200'] })
+      ).toThrowErrorMatchingInlineSnapshot(`"[hosts.0]: expected URI with scheme [https]."`);
+      expect(() =>
+        bodySchema.validate({ hosts: ['http://localhost:9200'] })
+      ).toThrowErrorMatchingInlineSnapshot(`"[hosts.0]: expected URI with scheme [https]."`);
+      expect(() =>
+        bodySchema.validate({
+          apiKey: 'some-key',
+          hosts: ['https://localhost:9200', 'http://localhost:9243'],
+        })
+      ).toThrowErrorMatchingInlineSnapshot(`"[hosts.1]: expected URI with scheme [https]."`);
+
+      expect(() =>
+        bodySchema.validate({ hosts: ['https://localhost:9200'] })
+      ).toThrowErrorMatchingInlineSnapshot(
+        `"[apiKey]: expected value of type [string] but got [undefined]"`
+      );
+      expect(() =>
+        bodySchema.validate({ apiKey: '', hosts: ['https://localhost:9200'] })
+      ).toThrowErrorMatchingInlineSnapshot(
+        `"[apiKey]: value has length [0] but it must have a minimum length of [1]."`
+      );
+
+      expect(() =>
+        bodySchema.validate({ apiKey: 'some-key', hosts: ['https://localhost:9200'] })
+      ).toThrowErrorMatchingInlineSnapshot(
+        `"[caFingerprint]: expected value of type [string] but got [undefined]"`
+      );
+      expect(() =>
+        bodySchema.validate({
+          apiKey: 'some-key',
+          hosts: ['https://localhost:9200'],
+          caFingerprint: '12345',
+        })
+      ).toThrowErrorMatchingInlineSnapshot(
+        `"[caFingerprint]: value has length [5] but it must have a minimum length of [64]."`
+      );
+
+      expect(
+        bodySchema.validate(
+          bodySchema.validate({
+            apiKey: 'some-key',
+            hosts: ['https://localhost:9200'],
+            caFingerprint: 'a'.repeat(64),
+          })
+        )
+      ).toEqual({
+        apiKey: 'some-key',
+        hosts: ['https://localhost:9200'],
+        caFingerprint: 'a'.repeat(64),
+      });
+    });
+
+    it('fails if setup is not on hold.', async () => {
+      mockRouteParams.preboot.isSetupOnHold.mockReturnValue(false);
+
+      const mockRequest = httpServerMock.createKibanaRequest({
+        body: { apiKey: 'some-key', hosts: ['host1', 'host2'], caFingerprint: 'ab:cd:ef' },
+      });
+
+      await expect(routeHandler(mockContext, mockRequest, kibanaResponseFactory)).resolves.toEqual({
+        status: 400,
+        options: { body: 'Cannot process request outside of preboot stage.' },
+        payload: 'Cannot process request outside of preboot stage.',
+      });
+
+      expect(mockRouteParams.elasticsearch.enroll).not.toHaveBeenCalled();
+      expect(mockRouteParams.kibanaConfigWriter.writeConfig).not.toHaveBeenCalled();
+      expect(mockRouteParams.preboot.completeSetup).not.toHaveBeenCalled();
+    });
+
+    it('fails if Elasticsearch connection is already configured.', async () => {
+      mockRouteParams.preboot.isSetupOnHold.mockReturnValue(true);
+      mockRouteParams.elasticsearch.connectionStatus$.next(
+        ElasticsearchConnectionStatus.Configured
+      );
+
+      const mockRequest = httpServerMock.createKibanaRequest({
+        body: { apiKey: 'some-key', hosts: ['host1', 'host2'], caFingerprint: 'ab:cd:ef' },
+      });
+
+      await expect(routeHandler(mockContext, mockRequest, kibanaResponseFactory)).resolves.toEqual({
+        status: 400,
+        options: {
+          body: {
+            message: 'Elasticsearch connection is already configured.',
+            attributes: { type: 'elasticsearch_connection_configured' },
+          },
+        },
+        payload: {
+          message: 'Elasticsearch connection is already configured.',
+          attributes: { type: 'elasticsearch_connection_configured' },
+        },
+      });
+
+      expect(mockRouteParams.elasticsearch.enroll).not.toHaveBeenCalled();
+      expect(mockRouteParams.kibanaConfigWriter.writeConfig).not.toHaveBeenCalled();
+      expect(mockRouteParams.preboot.completeSetup).not.toHaveBeenCalled();
+    });
+
+    it('fails if Kibana config is not writable.', async () => {
+      mockRouteParams.preboot.isSetupOnHold.mockReturnValue(true);
+      mockRouteParams.elasticsearch.connectionStatus$.next(
+        ElasticsearchConnectionStatus.NotConfigured
+      );
+      mockRouteParams.kibanaConfigWriter.isConfigWritable.mockResolvedValue(false);
+
+      const mockRequest = httpServerMock.createKibanaRequest({
+        body: { apiKey: 'some-key', hosts: ['host1', 'host2'], caFingerprint: 'ab:cd:ef' },
+      });
+
+      await expect(routeHandler(mockContext, mockRequest, kibanaResponseFactory)).resolves.toEqual({
+        status: 500,
+        options: {
+          body: {
+            message: 'Kibana process does not have enough permissions to write to config file.',
+            attributes: { type: 'kibana_config_not_writable' },
+          },
+          statusCode: 500,
+        },
+        payload: {
+          message: 'Kibana process does not have enough permissions to write to config file.',
+          attributes: { type: 'kibana_config_not_writable' },
+        },
+      });
+
+      expect(mockRouteParams.elasticsearch.enroll).not.toHaveBeenCalled();
+      expect(mockRouteParams.kibanaConfigWriter.writeConfig).not.toHaveBeenCalled();
+      expect(mockRouteParams.preboot.completeSetup).not.toHaveBeenCalled();
+    });
+
+    it('fails if enroll call fails.', async () => {
+      mockRouteParams.preboot.isSetupOnHold.mockReturnValue(true);
+      mockRouteParams.elasticsearch.connectionStatus$.next(
+        ElasticsearchConnectionStatus.NotConfigured
+      );
+      mockRouteParams.kibanaConfigWriter.isConfigWritable.mockResolvedValue(true);
+      mockRouteParams.elasticsearch.enroll.mockRejectedValue(
+        new errors.ResponseError(
+          interactiveSetupMock.createApiResponse({
+            statusCode: 401,
+            body: { message: 'some-secret-message' },
+          })
+        )
+      );
+
+      const mockRequest = httpServerMock.createKibanaRequest({
+        body: { apiKey: 'some-key', hosts: ['host1', 'host2'], caFingerprint: 'ab:cd:ef' },
+      });
+
+      await expect(routeHandler(mockContext, mockRequest, kibanaResponseFactory)).resolves.toEqual({
+        status: 500,
+        options: {
+          body: { message: 'Failed to enroll.', attributes: { type: 'enroll_failure' } },
+          statusCode: 500,
+        },
+        payload: { message: 'Failed to enroll.', attributes: { type: 'enroll_failure' } },
+      });
+
+      expect(mockRouteParams.elasticsearch.enroll).toHaveBeenCalledTimes(1);
+      expect(mockRouteParams.kibanaConfigWriter.writeConfig).not.toHaveBeenCalled();
+      expect(mockRouteParams.preboot.completeSetup).not.toHaveBeenCalled();
+    });
+
+    it('fails if cannot write configuration to the disk.', async () => {
+      mockRouteParams.preboot.isSetupOnHold.mockReturnValue(true);
+      mockRouteParams.elasticsearch.connectionStatus$.next(
+        ElasticsearchConnectionStatus.NotConfigured
+      );
+      mockRouteParams.kibanaConfigWriter.isConfigWritable.mockResolvedValue(true);
+      mockRouteParams.elasticsearch.enroll.mockResolvedValue({
+        ca: 'some-ca',
+        host: 'host',
+        serviceAccountToken: { name: 'some-name', value: 'some-value' },
+      });
+      mockRouteParams.kibanaConfigWriter.writeConfig.mockRejectedValue(
+        new Error('Some error with sensitive path')
+      );
+
+      const mockRequest = httpServerMock.createKibanaRequest({
+        body: { apiKey: 'some-key', hosts: ['host1', 'host2'], caFingerprint: 'ab:cd:ef' },
+      });
+
+      await expect(routeHandler(mockContext, mockRequest, kibanaResponseFactory)).resolves.toEqual({
+        status: 500,
+        options: {
+          body: {
+            message: 'Failed to save configuration.',
+            attributes: { type: 'kibana_config_failure' },
+          },
+          statusCode: 500,
+        },
+        payload: {
+          message: 'Failed to save configuration.',
+          attributes: { type: 'kibana_config_failure' },
+        },
+      });
+
+      expect(mockRouteParams.elasticsearch.enroll).toHaveBeenCalledTimes(1);
+      expect(mockRouteParams.kibanaConfigWriter.writeConfig).toHaveBeenCalledTimes(1);
+      expect(mockRouteParams.preboot.completeSetup).not.toHaveBeenCalled();
+    });
+
+    it('can successfully enrol and save configuration to the disk.', async () => {
+      mockRouteParams.preboot.isSetupOnHold.mockReturnValue(true);
+      mockRouteParams.elasticsearch.connectionStatus$.next(
+        ElasticsearchConnectionStatus.NotConfigured
+      );
+      mockRouteParams.kibanaConfigWriter.isConfigWritable.mockResolvedValue(true);
+      mockRouteParams.elasticsearch.enroll.mockResolvedValue({
+        ca: 'some-ca',
+        host: 'host',
+        serviceAccountToken: { name: 'some-name', value: 'some-value' },
+      });
+      mockRouteParams.kibanaConfigWriter.writeConfig.mockResolvedValue();
+
+      const mockRequest = httpServerMock.createKibanaRequest({
+        body: { apiKey: 'some-key', hosts: ['host1', 'host2'], caFingerprint: 'ab:cd:ef' },
+      });
+
+      await expect(routeHandler(mockContext, mockRequest, kibanaResponseFactory)).resolves.toEqual({
+        status: 204,
+        options: {},
+        payload: undefined,
+      });
+
+      expect(mockRouteParams.elasticsearch.enroll).toHaveBeenCalledTimes(1);
+      expect(mockRouteParams.elasticsearch.enroll).toHaveBeenCalledWith({
+        apiKey: 'some-key',
+        hosts: ['host1', 'host2'],
+        caFingerprint: 'ab:cd:ef',
+      });
+
+      expect(mockRouteParams.kibanaConfigWriter.writeConfig).toHaveBeenCalledTimes(1);
+      expect(mockRouteParams.kibanaConfigWriter.writeConfig).toHaveBeenCalledWith({
+        ca: 'some-ca',
+        host: 'host',
+        serviceAccountToken: { name: 'some-name', value: 'some-value' },
+      });
+
+      expect(mockRouteParams.preboot.completeSetup).toHaveBeenCalledTimes(1);
+      expect(mockRouteParams.preboot.completeSetup).toHaveBeenCalledWith({
+        shouldReloadConfig: true,
+      });
+    });
+  });
+});

src/plugins/interactive_setup/server/routes/enroll.ts

@@ -6,26 +6,105 @@
  * Side Public License, v 1.
  */
 
+import { first } from 'rxjs/operators';
+
 import { schema } from '@kbn/config-schema';
 
+import { ElasticsearchConnectionStatus } from '../../common';
+import type { EnrollResult } from '../elasticsearch_service';
 import type { RouteDefinitionParams } from './';
 
 /**
  * Defines routes to deal with Elasticsearch `enroll_kibana` APIs.
  */
-export function defineEnrollRoutes({ router }: RouteDefinitionParams) {
+export function defineEnrollRoutes({
+  router,
+  logger,
+  kibanaConfigWriter,
+  elasticsearch,
+  preboot,
+}: RouteDefinitionParams) {
   router.post(
     {
       path: '/internal/interactive_setup/enroll',
       validate: {
-        body: schema.object({ token: schema.string() }),
+        body: schema.object({
+          hosts: schema.arrayOf(schema.uri({ scheme: 'https' }), {
+            minSize: 1,
+          }),
+          apiKey: schema.string({ minLength: 1 }),
+          caFingerprint: schema.string({ maxLength: 64, minLength: 64 }),
+        }),
       },
       options: { authRequired: false },
     },
     async (context, request, response) => {
-      return response.forbidden({
-        body: { message: `API is not implemented yet.` },
-      });
+      if (!preboot.isSetupOnHold()) {
+        logger.error(`Invalid request to [path=${request.url.pathname}] outside of preboot stage`);
+        return response.badRequest({ body: 'Cannot process request outside of preboot stage.' });
+      }
+
+      const connectionStatus = await elasticsearch.connectionStatus$.pipe(first()).toPromise();
+      if (connectionStatus === ElasticsearchConnectionStatus.Configured) {
+        logger.error(
+          `Invalid request to [path=${request.url.pathname}], Elasticsearch connection is already configured.`
+        );
+        return response.badRequest({
+          body: {
+            message: 'Elasticsearch connection is already configured.',
+            attributes: { type: 'elasticsearch_connection_configured' },
+          },
+        });
+      }
+
+      // The most probable misconfiguration case is when Kibana process isn't allowed to write to the
+      // Kibana configuration file. We'll still have to handle possible filesystem access errors
+      // when we actually write to the disk, but this preliminary check helps us to avoid unnecessary
+      // enrollment call and communicate that to the user early.
+      const isConfigWritable = await kibanaConfigWriter.isConfigWritable();
+      if (!isConfigWritable) {
+        logger.error('Kibana process does not have enough permissions to write to config file');
+        return response.customError({
+          statusCode: 500,
+          body: {
+            message: 'Kibana process does not have enough permissions to write to config file.',
+            attributes: { type: 'kibana_config_not_writable' },
+          },
+        });
+      }
+
+      let enrollResult: EnrollResult;
+      try {
+        enrollResult = await elasticsearch.enroll({
+          apiKey: request.body.apiKey,
+          hosts: request.body.hosts,
+          caFingerprint: request.body.caFingerprint,
+        });
+      } catch {
+        // For security reasons, we shouldn't leak to the user whether Elasticsearch node couldn't process enrollment
+        // request or we just couldn't connect to any of the provided hosts.
+        return response.customError({
+          statusCode: 500,
+          body: { message: 'Failed to enroll.', attributes: { type: 'enroll_failure' } },
+        });
+      }
+
+      try {
+        await kibanaConfigWriter.writeConfig(enrollResult);
+      } catch {
+        // For security reasons, we shouldn't leak any filesystem related errors.
+        return response.customError({
+          statusCode: 500,
+          body: {
+            message: 'Failed to save configuration.',
+            attributes: { type: 'kibana_config_failure' },
+          },
+        });
+      }
+
+      preboot.completeSetup({ shouldReloadConfig: true });
+
+      return response.noContent();
     }
   );
 }

src/plugins/interactive_setup/server/routes/index.mock.ts

@@ -0,0 +1,26 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { coreMock, httpServiceMock, loggingSystemMock } from 'src/core/server/mocks';
+
+import { ConfigSchema } from '../config';
+import { elasticsearchServiceMock } from '../elasticsearch_service.mock';
+import { kibanaConfigWriterMock } from '../kibana_config_writer.mock';
+
+export const routeDefinitionParamsMock = {
+  create: (config: Record<string, unknown> = {}) => ({
+    router: httpServiceMock.createRouter(),
+    basePath: httpServiceMock.createBasePath(),
+    csp: httpServiceMock.createSetupContract().csp,
+    logger: loggingSystemMock.create().get(),
+    preboot: { ...coreMock.createPreboot().preboot, completeSetup: jest.fn() },
+    getConfig: jest.fn().mockReturnValue(ConfigSchema.validate(config)),
+    elasticsearch: elasticsearchServiceMock.createSetup(),
+    kibanaConfigWriter: kibanaConfigWriterMock.create(),
+  }),
+};

src/plugins/interactive_setup/server/routes/index.ts

@@ -6,10 +6,12 @@
  * Side Public License, v 1.
  */
 
-import type { IBasePath, IRouter, Logger } from 'src/core/server';
+import type { PublicMethodsOf } from '@kbn/utility-types';
+import type { IBasePath, IRouter, Logger, PrebootServicePreboot } from 'src/core/server';
 
-import type { ElasticsearchConnectionStatus } from '../../common';
 import type { ConfigType } from '../config';
+import type { ElasticsearchServiceSetup } from '../elasticsearch_service';
+import type { KibanaConfigWriter } from '../kibana_config_writer';
 import { defineEnrollRoutes } from './enroll';
 
 /**
@@ -19,8 +21,12 @@ export interface RouteDefinitionParams {
   readonly router: IRouter;
   readonly basePath: IBasePath;
   readonly logger: Logger;
+  readonly preboot: PrebootServicePreboot & {
+    completeSetup: (result: { shouldReloadConfig: boolean }) => void;
+  };
+  readonly kibanaConfigWriter: PublicMethodsOf<KibanaConfigWriter>;
+  readonly elasticsearch: ElasticsearchServiceSetup;
   readonly getConfig: () => ConfigType;
-  readonly getElasticsearchConnectionStatus: () => ElasticsearchConnectionStatus;
 }
 
 export function defineRoutes(params: RouteDefinitionParams) {