diff --git a/x-pack/plugins/security/server/authentication/authenticator.ts b/x-pack/plugins/security/server/authentication/authenticator.ts
index 116574c7f359cab..80e76b0bc714935 100644
--- a/x-pack/plugins/security/server/authentication/authenticator.ts
+++ b/x-pack/plugins/security/server/authentication/authenticator.ts
@@ -401,7 +401,10 @@ export class Authenticator {
       throw new Error('Current license does not allow access agreement acknowledgement.');
     }
 
-    await this.session.set(request, { ...existingSessionValue, accessAgreementAcknowledged: true });
+    await this.session.update(request, {
+      ...existingSessionValue,
+      accessAgreementAcknowledged: true,
+    });
 
     this.options.auditLogger.accessAgreementAcknowledged(
       currentUser.username,
@@ -478,13 +481,6 @@ export class Authenticator {
       return null;
     }
 
-    // If authentication succeeds or requires redirect we should automatically extend existing user session,
-    // unless authentication has been triggered by a system API request. In case provider explicitly returns new
-    // state we should store it in the session regardless of whether it's a system API request or not.
-    const sessionCanBeUpdated =
-      (authenticationResult.succeeded() || authenticationResult.redirected()) &&
-      (authenticationResult.shouldUpdateState() || !request.isSystemRequest);
-
     // If provider owned the session, but failed to authenticate anyway, that likely means that
     // session is not valid and we should clear it. Also provider can specifically ask to clear
     // session by setting it to `null` even if authentication attempt didn't fail.
@@ -496,16 +492,31 @@ export class Authenticator {
       return null;
     }
 
-    if (sessionCanBeUpdated) {
-      return await this.session.set(request, {
-        ...(existingSessionValue || { provider, lifespanExpiration: null }),
-        state: authenticationResult.shouldUpdateState()
-          ? authenticationResult.state
-          : existingSessionValue?.state,
+    // If authentication succeeds or requires redirect we should automatically extend existing user session,
+    // unless authentication has been triggered by a system API request. In case provider explicitly returns new
+    // state we should store it in the session regardless of whether it's a system API request or not.
+    const sessionCanBeUpdated =
+      (authenticationResult.succeeded() || authenticationResult.redirected()) &&
+      (authenticationResult.shouldUpdateState() || !request.isSystemRequest);
+    if (!sessionCanBeUpdated) {
+      return existingSessionValue;
+    }
+
+    if (!existingSessionValue) {
+      return await this.session.create(request, {
+        provider,
+        state: authenticationResult.shouldUpdateState() ? authenticationResult.state : null,
+      });
+    }
+
+    if (authenticationResult.shouldUpdateState()) {
+      return await this.session.update(request, {
+        ...existingSessionValue,
+        state: authenticationResult.state,
       });
     }
 
-    return existingSessionValue;
+    return await this.session.extend(request, existingSessionValue);
   }
 
   /**
diff --git a/x-pack/plugins/security/server/authentication/index.ts b/x-pack/plugins/security/server/authentication/index.ts
index a7cb25abb68d1d7..1d10dfd90a8240c 100644
--- a/x-pack/plugins/security/server/authentication/index.ts
+++ b/x-pack/plugins/security/server/authentication/index.ts
@@ -12,7 +12,6 @@ import {
 } from '../../../../../src/core/server';
 import { SecurityLicense } from '../../common/licensing';
 import { AuthenticatedUser } from '../../common/model';
-import { SessionInfo } from '../../common/types';
 import { SecurityAuditLogger } from '../audit';
 import { ConfigType } from '../config';
 import { getErrorStatusCode } from '../errors';
@@ -46,6 +45,7 @@ interface SetupAuthenticationParams {
   config: ConfigType;
   license: SecurityLicense;
   loggers: LoggerFactory;
+  session: Session;
 }
 
 export type Authentication = UnwrapPromise<ReturnType<typeof setupAuthentication>>;
@@ -58,6 +58,7 @@ export async function setupAuthentication({
   config,
   license,
   loggers,
+  session,
 }: SetupAuthenticationParams) {
   const authLogger = loggers.get('authentication');
 
@@ -73,34 +74,6 @@ export async function setupAuthentication({
     return (http.auth.get(request).state ?? null) as AuthenticatedUser | null;
   };
 
-  /**
-   * Returns session information for the current request.
-   * @param request Request instance.
-   */
-  const getSessionInfo = async (request: KibanaRequest): Promise<SessionInfo | null> => {
-    const sessionValue = await session.get(request);
-    if (!sessionValue) {
-      return null;
-    }
-
-    // We can't rely on the client's system clock, so in addition to returning expiration timestamps, we also return
-    // the current server time -- that way the client can calculate the relative time to expiration.
-    return {
-      now: Date.now(),
-      idleTimeoutExpiration: sessionValue.idleTimeoutExpiration,
-      lifespanExpiration: sessionValue.lifespanExpiration,
-      provider: sessionValue.provider,
-    };
-  };
-
-  const session = new Session({
-    auditLogger,
-    logger: loggers.get('session'),
-    clusterClient,
-    config,
-    http,
-  });
-
   authLogger.debug('Successfully initialized session.');
 
   const authenticator = new Authenticator({
@@ -182,7 +155,6 @@ export async function setupAuthentication({
   return {
     login: authenticator.login.bind(authenticator),
     logout: authenticator.logout.bind(authenticator),
-    getSessionInfo,
     isProviderTypeEnabled: authenticator.isProviderTypeEnabled.bind(authenticator),
     acknowledgeAccessAgreement: authenticator.acknowledgeAccessAgreement.bind(authenticator),
     getCurrentUser,
diff --git a/x-pack/plugins/security/server/authorization/authorization_service.ts b/x-pack/plugins/security/server/authorization/authorization_service.ts
index 989784a1436d261..bb3d1d01be36158 100644
--- a/x-pack/plugins/security/server/authorization/authorization_service.ts
+++ b/x-pack/plugins/security/server/authorization/authorization_service.ts
@@ -4,16 +4,13 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { combineLatest, BehaviorSubject, Subscription } from 'rxjs';
-import { distinctUntilChanged, filter } from 'rxjs/operators';
+import { Subscription, Observable } from 'rxjs';
 import { UICapabilities } from 'ui/capabilities';
 import {
   LoggerFactory,
   KibanaRequest,
   IClusterClient,
-  ServiceStatusLevels,
   Logger,
-  StatusServiceSetup,
   HttpServiceSetup,
   CapabilitiesSetup,
 } from '../../../../../src/core/server';
@@ -44,6 +41,7 @@ import { validateReservedPrivileges } from './validate_reserved_privileges';
 import { registerPrivilegesWithCluster } from './register_privileges_with_cluster';
 import { APPLICATION_PREFIX } from '../../common/constants';
 import { SecurityLicense } from '../../common/licensing';
+import { OnlineStatusRetryScheduler } from '../elasticsearch';
 
 export { Actions } from './actions';
 export { CheckSavedObjectsPrivileges } from './check_saved_objects_privileges';
@@ -52,7 +50,6 @@ export { featurePrivilegeIterator } from './privileges';
 interface AuthorizationServiceSetupParams {
   packageVersion: string;
   http: HttpServiceSetup;
-  status: StatusServiceSetup;
   capabilities: CapabilitiesSetup;
   clusterClient: IClusterClient;
   license: SecurityLicense;
@@ -65,6 +62,7 @@ interface AuthorizationServiceSetupParams {
 interface AuthorizationServiceStartParams {
   features: FeaturesPluginStart;
   clusterClient: IClusterClient;
+  online$: Observable<OnlineStatusRetryScheduler>;
 }
 
 export interface AuthorizationServiceSetup {
@@ -79,8 +77,6 @@ export interface AuthorizationServiceSetup {
 
 export class AuthorizationService {
   private logger!: Logger;
-  private license!: SecurityLicense;
-  private status!: StatusServiceSetup;
   private applicationName!: string;
   private privileges!: PrivilegesService;
 
@@ -89,7 +85,6 @@ export class AuthorizationService {
   setup({
     http,
     capabilities,
-    status,
     packageVersion,
     clusterClient,
     license,
@@ -99,8 +94,6 @@ export class AuthorizationService {
     getSpacesService,
   }: AuthorizationServiceSetupParams): AuthorizationServiceSetup {
     this.logger = loggers.get('authorization');
-    this.license = license;
-    this.status = status;
     this.applicationName = `${APPLICATION_PREFIX}${kibanaIndexName}`;
 
     const mode = authorizationModeFactory(license);
@@ -158,12 +151,23 @@ export class AuthorizationService {
     return authz;
   }
 
-  start({ clusterClient, features }: AuthorizationServiceStartParams) {
+  start({ clusterClient, features, online$ }: AuthorizationServiceStartParams) {
     const allFeatures = features.getFeatures();
     validateFeaturePrivileges(allFeatures);
     validateReservedPrivileges(allFeatures);
 
-    this.registerPrivileges(clusterClient);
+    this.statusSubscription = online$.subscribe(async ({ scheduleRetry }) => {
+      try {
+        await registerPrivilegesWithCluster(
+          this.logger,
+          this.privileges,
+          this.applicationName,
+          clusterClient
+        );
+      } catch (err) {
+        scheduleRetry();
+      }
+    });
   }
 
   stop() {
@@ -172,50 +176,4 @@ export class AuthorizationService {
       this.statusSubscription = undefined;
     }
   }
-
-  private registerPrivileges(clusterClient: IClusterClient) {
-    const RETRY_SCALE_DURATION = 100;
-    const RETRY_TIMEOUT_MAX = 10000;
-    const retries$ = new BehaviorSubject(0);
-    let retryTimeout: NodeJS.Timeout;
-
-    // Register cluster privileges once Elasticsearch is available and Security plugin is enabled.
-    this.statusSubscription = combineLatest([
-      this.status.core$,
-      this.license.features$,
-      retries$.asObservable().pipe(
-        // We shouldn't emit new value if retry counter is reset. This comparator isn't called for
-        // the initial value.
-        distinctUntilChanged((prev, curr) => prev === curr || curr === 0)
-      ),
-    ])
-      .pipe(
-        filter(
-          ([status]) =>
-            this.license.isEnabled() && status.elasticsearch.level === ServiceStatusLevels.available
-        )
-      )
-      .subscribe(async () => {
-        // If status or license change occurred before retry timeout we should cancel it.
-        if (retryTimeout) {
-          clearTimeout(retryTimeout);
-        }
-
-        try {
-          await registerPrivilegesWithCluster(
-            this.logger,
-            this.privileges,
-            this.applicationName,
-            clusterClient
-          );
-          retries$.next(0);
-        } catch (err) {
-          const retriesElapsed = retries$.getValue() + 1;
-          retryTimeout = setTimeout(
-            () => retries$.next(retriesElapsed),
-            Math.min(retriesElapsed * RETRY_SCALE_DURATION, RETRY_TIMEOUT_MAX)
-          );
-        }
-      });
-  }
 }
diff --git a/x-pack/plugins/security/server/elasticsearch/elasticsearch_service.ts b/x-pack/plugins/security/server/elasticsearch/elasticsearch_service.ts
new file mode 100644
index 000000000000000..75f7bfde9ce84e1
--- /dev/null
+++ b/x-pack/plugins/security/server/elasticsearch/elasticsearch_service.ts
@@ -0,0 +1,121 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { BehaviorSubject, combineLatest, Observable } from 'rxjs';
+import { distinctUntilChanged, filter, map, shareReplay, tap } from 'rxjs/operators';
+import {
+  IClusterClient,
+  ICustomClusterClient,
+  Logger,
+  ServiceStatusLevels,
+  StatusServiceSetup,
+  ElasticsearchServiceSetup as CoreElasticsearchServiceSetup,
+} from '../../../../../src/core/server';
+import { SecurityLicense } from '../../common/licensing';
+import { elasticsearchClientPlugin } from './elasticsearch_client_plugin';
+
+export interface ElasticsearchServiceSetupParams {
+  readonly elasticsearch: CoreElasticsearchServiceSetup;
+  readonly status: StatusServiceSetup;
+  readonly license: SecurityLicense;
+}
+
+export interface ElasticsearchServiceSetup {
+  readonly clusterClient: IClusterClient;
+}
+
+export interface ElasticsearchServiceStart {
+  readonly clusterClient: IClusterClient;
+  readonly watchOnlineStatus$: () => Observable<OnlineStatusRetryScheduler>;
+}
+
+export interface OnlineStatusRetryScheduler {
+  scheduleRetry: () => void;
+}
+
+export class ElasticsearchService {
+  private clusterClient?: ICustomClusterClient;
+  private coreStatus$!: Observable<boolean>;
+
+  constructor(private readonly logger: Logger) {}
+
+  setup({
+    elasticsearch,
+    status,
+    license,
+  }: ElasticsearchServiceSetupParams): ElasticsearchServiceSetup {
+    this.clusterClient = elasticsearch.legacy.createClient('security', {
+      plugins: [elasticsearchClientPlugin],
+    });
+
+    this.coreStatus$ = combineLatest([status.core$, license.features$]).pipe(
+      map(
+        ([coreStatus]) =>
+          license.isEnabled() && coreStatus.elasticsearch.level === ServiceStatusLevels.available
+      ),
+      shareReplay(1)
+    );
+
+    return { clusterClient: this.clusterClient };
+  }
+
+  start(): ElasticsearchServiceStart {
+    return {
+      clusterClient: this.clusterClient!,
+      watchOnlineStatus$: () => {
+        const RETRY_SCALE_DURATION = 100;
+        const RETRY_TIMEOUT_MAX = 10000;
+        const retries$ = new BehaviorSubject(0);
+
+        const retryScheduler = {
+          scheduleRetry: () => {
+            const retriesElapsed = retries$.getValue() + 1;
+            const nextRetryTimeout = Math.min(
+              retriesElapsed * RETRY_SCALE_DURATION,
+              RETRY_TIMEOUT_MAX
+            );
+
+            this.logger.debug(`Scheduling re-try in ${nextRetryTimeout} ms.`);
+
+            retryTimeout = setTimeout(() => retries$.next(retriesElapsed), nextRetryTimeout);
+          },
+        };
+
+        let retryTimeout: NodeJS.Timeout;
+        return combineLatest([
+          this.coreStatus$.pipe(
+            tap(() => {
+              // If status or license change occurred before retry timeout we should cancel
+              // it and reset retry counter.
+              if (retryTimeout) {
+                clearTimeout(retryTimeout);
+              }
+
+              if (retries$.value > 0) {
+                retries$.next(0);
+              }
+            })
+          ),
+          retries$.asObservable().pipe(
+            // We shouldn't emit new value if retry counter is reset. This comparator isn't called for
+            // the initial value.
+            distinctUntilChanged((prev, curr) => prev === curr || curr === 0)
+          ),
+        ]).pipe(
+          filter(([isAvailable]) => isAvailable),
+          map(() => retryScheduler)
+        );
+      },
+    };
+  }
+
+  stop() {
+    if (this.clusterClient) {
+      this.clusterClient.close();
+      this.clusterClient = undefined;
+    }
+  }
+}
diff --git a/x-pack/plugins/security/server/elasticsearch/index.ts b/x-pack/plugins/security/server/elasticsearch/index.ts
new file mode 100644
index 000000000000000..793bdc1c6ad2612
--- /dev/null
+++ b/x-pack/plugins/security/server/elasticsearch/index.ts
@@ -0,0 +1,12 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+export {
+  ElasticsearchService,
+  ElasticsearchServiceSetup,
+  ElasticsearchServiceStart,
+  OnlineStatusRetryScheduler,
+} from './elasticsearch_service';
diff --git a/x-pack/plugins/security/server/plugin.ts b/x-pack/plugins/security/server/plugin.ts
index 555bd06e5549d1f..da2ef87053b960c 100644
--- a/x-pack/plugins/security/server/plugin.ts
+++ b/x-pack/plugins/security/server/plugin.ts
@@ -9,7 +9,6 @@ import { first, map } from 'rxjs/operators';
 import { TypeOf } from '@kbn/config-schema';
 import {
   deepFreeze,
-  ICustomClusterClient,
   CoreSetup,
   CoreStart,
   Logger,
@@ -29,8 +28,9 @@ import { defineRoutes } from './routes';
 import { SecurityLicenseService, SecurityLicense } from '../common/licensing';
 import { setupSavedObjects } from './saved_objects';
 import { AuditService, SecurityAuditLogger, AuditServiceSetup } from './audit';
-import { elasticsearchClientPlugin } from './elasticsearch/elasticsearch_client_plugin';
 import { SecurityFeatureUsageService, SecurityFeatureUsageServiceStart } from './feature_usage';
+import { SessionManagementService } from './session_management';
+import { ElasticsearchService } from './elasticsearch';
 
 export type SpacesService = Pick<
   SpacesPluginSetup['spacesService'],
@@ -81,7 +81,6 @@ export interface PluginStartDependencies {
  */
 export class Plugin {
   private readonly logger: Logger;
-  private clusterClient?: ICustomClusterClient;
   private spacesService?: SpacesService | symbol = Symbol('not accessed');
   private securityLicenseService?: SecurityLicenseService;
 
@@ -96,6 +95,12 @@ export class Plugin {
 
   private readonly auditService = new AuditService(this.initializerContext.logger.get('audit'));
   private readonly authorizationService = new AuthorizationService();
+  private readonly elasticsearchService = new ElasticsearchService(
+    this.initializerContext.logger.get('elasticsearch')
+  );
+  private readonly sessionManagementService = new SessionManagementService(
+    this.initializerContext.logger.get('session')
+  );
 
   private readonly getSpacesService = () => {
     // Changing property value from Symbol to undefined denotes the fact that property was accessed.
@@ -127,35 +132,44 @@ export class Plugin {
       .pipe(first())
       .toPromise();
 
-    this.clusterClient = core.elasticsearch.legacy.createClient('security', {
-      plugins: [elasticsearchClientPlugin],
-    });
-
     this.securityLicenseService = new SecurityLicenseService();
     const { license } = this.securityLicenseService.setup({
       license$: licensing.license$,
     });
 
+    const { clusterClient } = this.elasticsearchService.setup({
+      elasticsearch: core.elasticsearch,
+      license,
+      status: core.status,
+    });
+
     this.featureUsageService.setup({ featureUsage: licensing.featureUsage });
 
     const audit = this.auditService.setup({ license, config: config.audit });
     const auditLogger = new SecurityAuditLogger(audit.getLogger());
 
+    const { session } = this.sessionManagementService.setup({
+      auditLogger,
+      config,
+      clusterClient,
+      http: core.http,
+    });
+
     const authc = await setupAuthentication({
       auditLogger,
       getFeatureUsageService: this.getFeatureUsageService,
       http: core.http,
-      clusterClient: this.clusterClient,
+      clusterClient,
       config,
       license,
       loggers: this.initializerContext.logger,
+      session,
     });
 
     const authz = this.authorizationService.setup({
       http: core.http,
       capabilities: core.capabilities,
-      status: core.status,
-      clusterClient: this.clusterClient,
+      clusterClient,
       license,
       loggers: this.initializerContext.logger,
       kibanaIndexName: legacyConfig.kibana.index,
@@ -176,11 +190,12 @@ export class Plugin {
       basePath: core.http.basePath,
       httpResources: core.http.resources,
       logger: this.initializerContext.logger.get('routes'),
-      clusterClient: this.clusterClient,
+      clusterClient,
       config,
       authc,
       authz,
       license,
+      session,
       getFeatures: () =>
         core
           .getStartServices()
@@ -223,20 +238,20 @@ export class Plugin {
 
   public start(core: CoreStart, { features, licensing }: PluginStartDependencies) {
     this.logger.debug('Starting plugin');
+
     this.featureUsageServiceStart = this.featureUsageService.start({
       featureUsage: licensing.featureUsage,
     });
-    this.authorizationService.start({ features, clusterClient: this.clusterClient! });
+
+    const { clusterClient, watchOnlineStatus$ } = this.elasticsearchService.start();
+
+    this.sessionManagementService.start({ online$: watchOnlineStatus$() });
+    this.authorizationService.start({ features, clusterClient, online$: watchOnlineStatus$() });
   }
 
   public stop() {
     this.logger.debug('Stopping plugin');
 
-    if (this.clusterClient) {
-      this.clusterClient.close();
-      this.clusterClient = undefined;
-    }
-
     if (this.securityLicenseService) {
       this.securityLicenseService.stop();
       this.securityLicenseService = undefined;
@@ -245,8 +260,11 @@ export class Plugin {
     if (this.featureUsageServiceStart) {
       this.featureUsageServiceStart = undefined;
     }
+
     this.auditService.stop();
     this.authorizationService.stop();
+    this.elasticsearchService.stop();
+    this.sessionManagementService.stop();
   }
 
   private wasSpacesServiceAccessed() {
diff --git a/x-pack/plugins/security/server/routes/authentication/session.ts b/x-pack/plugins/security/server/routes/authentication/session.ts
index cdebc19d7cf8dbf..e1deed16ae449c2 100644
--- a/x-pack/plugins/security/server/routes/authentication/session.ts
+++ b/x-pack/plugins/security/server/routes/authentication/session.ts
@@ -4,12 +4,13 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
+import { SessionInfo } from '../../../common/types';
 import { RouteDefinitionParams } from '..';
 
 /**
  * Defines routes required for all authentication realms.
  */
-export function defineSessionRoutes({ router, logger, authc, basePath }: RouteDefinitionParams) {
+export function defineSessionRoutes({ router, logger, basePath, session }: RouteDefinitionParams) {
   router.get(
     {
       path: '/internal/security/session',
@@ -17,9 +18,21 @@ export function defineSessionRoutes({ router, logger, authc, basePath }: RouteDe
     },
     async (_context, request, response) => {
       try {
-        const sessionInfo = await authc.getSessionInfo(request);
-        // This is an authenticated request, so sessionInfo will always be non-null.
-        return response.ok({ body: sessionInfo! });
+        const sessionValue = await session.get(request);
+        return response.ok(
+          sessionValue
+            ? {
+                body: {
+                  // We can't rely on the client's system clock, so in addition to returning expiration timestamps, we also return
+                  // the current server time -- that way the client can calculate the relative time to expiration.
+                  now: Date.now(),
+                  idleTimeoutExpiration: sessionValue.idleTimeoutExpiration,
+                  lifespanExpiration: sessionValue.lifespanExpiration,
+                  provider: sessionValue.provider,
+                } as SessionInfo,
+              }
+            : {}
+        );
       } catch (err) {
         logger.error(`Error retrieving user session: ${err.message}`);
         return response.internalError();
diff --git a/x-pack/plugins/security/server/routes/index.ts b/x-pack/plugins/security/server/routes/index.ts
index 5721a2699d15ced..b0d961afde12924 100644
--- a/x-pack/plugins/security/server/routes/index.ts
+++ b/x-pack/plugins/security/server/routes/index.ts
@@ -25,6 +25,7 @@ import { defineUsersRoutes } from './users';
 import { defineRoleMappingRoutes } from './role_mapping';
 import { defineViewRoutes } from './views';
 import { SecurityFeatureUsageServiceStart } from '../feature_usage';
+import { Session } from '../session_management';
 
 /**
  * Describes parameters used to define HTTP routes.
@@ -38,6 +39,7 @@ export interface RouteDefinitionParams {
   config: ConfigType;
   authc: Authentication;
   authz: AuthorizationServiceSetup;
+  session: Session;
   license: SecurityLicense;
   getFeatures: () => Promise<Feature[]>;
   getFeatureUsageService: () => SecurityFeatureUsageServiceStart;
diff --git a/x-pack/plugins/security/server/routes/users/change_password.ts b/x-pack/plugins/security/server/routes/users/change_password.ts
index e915cd8759ff105..e149ac255065cff 100644
--- a/x-pack/plugins/security/server/routes/users/change_password.ts
+++ b/x-pack/plugins/security/server/routes/users/change_password.ts
@@ -16,6 +16,7 @@ import { RouteDefinitionParams } from '..';
 
 export function defineChangeUserPasswordRoutes({
   authc,
+  session,
   router,
   clusterClient,
 }: RouteDefinitionParams) {
@@ -37,7 +38,7 @@ export function defineChangeUserPasswordRoutes({
       const currentUser = authc.getCurrentUser(request);
       const isUserChangingOwnPassword =
         currentUser && currentUser.username === username && canUserChangePassword(currentUser);
-      const currentSession = isUserChangingOwnPassword ? await authc.getSessionInfo(request) : null;
+      const currentSession = isUserChangingOwnPassword ? await session.get(request) : null;
 
       // If user is changing their own password they should provide a proof of knowledge their
       // current password via sending it in `Authorization: Basic base64(username:current password)`
diff --git a/x-pack/plugins/security/server/routes/views/access_agreement.ts b/x-pack/plugins/security/server/routes/views/access_agreement.ts
index 49e1ff42a28a2a5..80a1c2a20cf599d 100644
--- a/x-pack/plugins/security/server/routes/views/access_agreement.ts
+++ b/x-pack/plugins/security/server/routes/views/access_agreement.ts
@@ -12,7 +12,7 @@ import { RouteDefinitionParams } from '..';
  * Defines routes required for the Access Agreement view.
  */
 export function defineAccessAgreementRoutes({
-  authc,
+  session,
   httpResources,
   license,
   config,
@@ -46,12 +46,12 @@ export function defineAccessAgreementRoutes({
       // authenticated with the help of HTTP authentication), that means we should safely check if
       // we have it and can get a corresponding configuration.
       try {
-        const session = await authc.getSessionInfo(request);
+        const sessionValue = await session.get(request);
         const accessAgreement =
-          (session &&
+          (sessionValue &&
             config.authc.providers[
-              session.provider.type as keyof ConfigType['authc']['providers']
-            ]?.[session.provider.name]?.accessAgreement?.message) ||
+              sessionValue.provider.type as keyof ConfigType['authc']['providers']
+            ]?.[sessionValue.provider.name]?.accessAgreement?.message) ||
           '';
 
         return response.ok({ body: { accessAgreement } });
diff --git a/x-pack/plugins/security/server/routes/views/logged_out.ts b/x-pack/plugins/security/server/routes/views/logged_out.ts
index 43c2f01b1b53d41..b35154e6a0f2a4d 100644
--- a/x-pack/plugins/security/server/routes/views/logged_out.ts
+++ b/x-pack/plugins/security/server/routes/views/logged_out.ts
@@ -17,7 +17,7 @@ import { RouteDefinitionParams } from '..';
  */
 export function defineLoggedOutRoutes({
   logger,
-  authc,
+  session,
   httpResources,
   basePath,
 }: RouteDefinitionParams) {
@@ -30,7 +30,7 @@ export function defineLoggedOutRoutes({
     async (context, request, response) => {
       // Authentication flow isn't triggered automatically for this route, so we should explicitly
       // check whether user has an active session already.
-      const isUserAlreadyLoggedIn = (await authc.getSessionInfo(request)) !== null;
+      const isUserAlreadyLoggedIn = (await session.get(request)) !== null;
       if (isUserAlreadyLoggedIn) {
         logger.debug('User is already authenticated, redirecting...');
         return response.redirected({
diff --git a/x-pack/plugins/security/server/saved_objects/mappings.json b/x-pack/plugins/security/server/saved_objects/mappings.json
new file mode 100644
index 000000000000000..27dfeda94e91c57
--- /dev/null
+++ b/x-pack/plugins/security/server/saved_objects/mappings.json
@@ -0,0 +1,32 @@
+{
+  "session": {
+    "dynamic": "strict",
+    "properties": {
+      "username_hash": {
+        "type": "keyword"
+      },
+      "roles": {
+        "type": "keyword"
+      },
+      "provider": {
+        "properties": {
+          "type": {
+            "type": "keyword"
+          },
+          "name": {
+            "type": "keyword"
+          }
+        }
+      },
+      "lifespanExpiration": {
+        "type": "date"
+      },
+      "accessAgreementAcknowledged": {
+        "type": "boolean"
+      },
+      "state": {
+        "type": "binary"
+      }
+    }
+  }
+}
diff --git a/x-pack/plugins/security/server/session_management/index.ts b/x-pack/plugins/security/server/session_management/index.ts
index 13d1f35d8e8d234..990ddac53881bfb 100644
--- a/x-pack/plugins/security/server/session_management/index.ts
+++ b/x-pack/plugins/security/server/session_management/index.ts
@@ -4,4 +4,9 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-export { Session, SessionValue } from './session';
+export { Session } from './session';
+export { SessionValue } from './session_value';
+export {
+  SessionManagementServiceSetup,
+  SessionManagementService,
+} from './session_management_service';
diff --git a/x-pack/plugins/security/server/session_management/session.ts b/x-pack/plugins/security/server/session_management/session.ts
index 732351a7a9ffab9..44edad773999d93 100644
--- a/x-pack/plugins/security/server/session_management/session.ts
+++ b/x-pack/plugins/security/server/session_management/session.ts
@@ -4,89 +4,22 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import {
-  IClusterClient,
-  KibanaRequest,
-  SessionStorageFactory,
-  Logger,
-  HttpServiceSetup,
-} from 'kibana/server';
-import { AuthenticationProvider } from '../../common/types';
+import nodeCrypto from '@elastic/node-crypto';
+import { promisify } from 'util';
+import { randomBytes } from 'crypto';
+import { KibanaRequest, Logger } from '../../../../../src/core/server';
 import { SecurityAuditLogger } from '../audit';
 import { ConfigType } from '../config';
-
-/**
- * The shape of the session value.
- */
-export interface SessionValue {
-  /**
-   * Name and type of the provider this session belongs to.
-   */
-  provider: AuthenticationProvider;
-
-  /**
-   * The Unix time in ms when the session should be considered expired. If `null`, session will stay
-   * active until the browser is closed.
-   */
-  idleTimeoutExpiration: number | null;
-
-  /**
-   * The Unix time in ms which is the max total lifespan of the session. If `null`, session expire
-   * time can be extended indefinitely.
-   */
-  lifespanExpiration: number | null;
-
-  /**
-   * Session value that is fed to the authentication provider. The shape is unknown upfront and
-   * entirely determined by the authentication provider that owns the current session.
-   */
-  state: unknown;
-
-  /**
-   * Cookie "Path" attribute that is validated against the current Kibana server configuration.
-   */
-  path: string;
-
-  /**
-   * Indicates whether user acknowledged access agreement or not.
-   */
-  accessAgreementAcknowledged?: boolean;
-}
+import { SessionValue } from './session_value';
+import { SessionIndex } from './session_index';
+import { SessionCookie } from './session_cookie';
 
 export interface SessionOptions {
   auditLogger: SecurityAuditLogger;
   logger: Logger;
-  clusterClient: IClusterClient;
-  http: Pick<HttpServiceSetup, 'basePath' | 'createCookieSessionStorageFactory'>;
-  config: ConfigType;
-}
-
-const isCookieSessionValueValid = (
-  sessionValue: SessionValue,
-  serverBasePath: string,
-  logger: Logger
-) => {
-  // ensure that this cookie was created with the current Kibana configuration
-  const { path, idleTimeoutExpiration, lifespanExpiration } = sessionValue;
-  if (path !== undefined && path !== serverBasePath) {
-    logger.debug(`Outdated session value with path "${sessionValue.path}"`);
-    return false;
-  }
-  // ensure that this cookie is not expired
-  if (idleTimeoutExpiration && idleTimeoutExpiration < Date.now()) {
-    return false;
-  } else if (lifespanExpiration && lifespanExpiration < Date.now()) {
-    return false;
-  }
-  return true;
-};
-
-/**
- * Determines if session value was created by the current Kibana version. Previous versions had a different session value format.
- * @param sessionValue The session value to check.
- */
-function isSupportedProviderSession(sessionValue: any): sessionValue is SessionValue {
-  return typeof sessionValue?.provider?.name === 'string';
+  sessionIndex: SessionIndex;
+  sessionCookie: SessionCookie;
+  config: Pick<ConfigType, 'encryptionKey' | 'authc' | 'session'>;
 }
 
 export class Session {
@@ -102,81 +35,218 @@ export class Session {
    */
   private readonly idleTimeout = this.options.config.session.idleTimeout;
 
+  /**
+   * Timeout after which idle timeout property is updated in the index. It's two times longer than
+   * configured idle timeout since index updates are costly and we want to minimize them.
+   */
+  private readonly idleIndexUpdateTimeout = this.options.config.session.idleTimeout
+    ? this.options.config.session.idleTimeout.asMilliseconds() * 2
+    : null;
+
   /**
    * Session max lifespan in ms. If `null` session may live indefinitely.
    */
   private readonly lifespan = this.options.config.session.lifespan;
 
   /**
-   * Which base path the HTTP server is hosted on.
+   * Used to encrypt and decrypt portion of the session value using configured encryption key.
    */
-  private readonly serverBasePath = this.options.http.basePath.serverBasePath || '/';
+  private readonly crypto = nodeCrypto({ encryptionKey: this.options.config.encryptionKey });
 
   /**
-   * Promise containing initialized cookie session storage factory.
+   * Promise-based version of the NodeJS native `randomBytes`.
    */
-  private readonly cookieSessionValueStorage: Promise<SessionStorageFactory<SessionValue>>;
-
-  constructor(private readonly options: SessionOptions) {
-    this.cookieSessionValueStorage = options.http.createCookieSessionStorageFactory({
-      encryptionKey: options.config.encryptionKey,
-      isSecure: options.config.secureCookies,
-      name: options.config.cookieName,
-      validate: (sessionValue: SessionValue | SessionValue[]) => {
-        for (const sess of Array.isArray(sessionValue) ? sessionValue : [sessionValue]) {
-          if (!isCookieSessionValueValid(sess, this.serverBasePath, options.logger)) {
-            return { isValid: false, path: sess.path };
-          }
-        }
-        return { isValid: true };
-      },
-    });
-  }
+  private readonly randomBytes = promisify(randomBytes);
+
+  constructor(private readonly options: SessionOptions) {}
 
   /**
-   * Extracts session value for the specified request. Under the hood it can
-   * clear session if it belongs to the provider that is no longer available or was created by legacy
-   * versions of Kibana.
+   * Extracts session value for the specified request. Under the hood it can clear session if it is
+   * invalid, created by the legacy versions of Kibana or belongs to the provider that is no longer
+   * available.
    * @param request Request instance to get session value for.
    */
   async get(request: KibanaRequest) {
-    const sessionStorage = (await this.cookieSessionValueStorage).asScoped(request);
-    const sessionValue = await sessionStorage.get();
+    const sessionCookieValue = await this.options.sessionCookie.get(request);
+    if (!sessionCookieValue) {
+      return null;
+    }
 
-    // If we detect that session is in incompatible format or for some reason we have a session
-    // stored for the provider that is not available anymore (e.g. when user was logged in with one
-    // provider, but then configuration has changed and that provider is no longer available), then
-    // we should clear session entirely.
     if (
-      sessionValue &&
-      (!isSupportedProviderSession(sessionValue) ||
-        this.providers.get(sessionValue.provider.name) !== sessionValue.provider.type)
+      (sessionCookieValue.idleTimeoutExpiration &&
+        sessionCookieValue.idleTimeoutExpiration < Date.now()) ||
+      (sessionCookieValue.lifespanExpiration && sessionCookieValue.lifespanExpiration < Date.now())
     ) {
-      sessionStorage.clear();
+      this.options.logger.debug('Session has expired and will be invalidated.');
+      await this.clear(request);
+      return null;
+    }
+
+    const sessionIndexValue = await this.options.sessionIndex.get(sessionCookieValue.sid);
+    if (!sessionIndexValue) {
+      this.options.logger.debug('Session value is not available.');
+      await this.clear(request);
       return null;
     }
 
-    return sessionValue;
+    // If we detect that for some reason we have a session stored for the provider that is not
+    // available anymore (e.g. when user was logged in with one provider, but then configuration has
+    // changed and that provider is no longer available), then we should clear session entirely.
+    if (this.providers.get(sessionIndexValue.provider.name) !== sessionIndexValue.provider.type) {
+      this.options.logger.warn(
+        `Session was created for "${sessionIndexValue.provider.name}/${sessionIndexValue.provider.type}" provider that is no longer configured or has a different type. Session will be invalidated.`
+      );
+      await this.clear(request);
+      return null;
+    }
+
+    try {
+      return {
+        ...sessionIndexValue,
+        state: await this.decryptState(sessionIndexValue.state as string, sessionCookieValue.aad),
+      };
+    } catch (err) {
+      await this.clear(request);
+      return null;
+    }
   }
 
   /**
-   * Creates or updates session value for the specified request.
-   * @param request Request instance to set session value for.
+   * Creates new session document in the session index encrypting sensitive state.
+   * @param request Request instance to create session value for.
    * @param sessionValue Session value parameters.
    */
-  async set(
+  async create(
     request: KibanaRequest,
-    sessionValue: Omit<SessionValue, 'path' | 'idleTimeoutExpiration'>
+    sessionValue: Omit<SessionValue, 'sid' | 'idleTimeoutExpiration' | 'lifespanExpiration'>
   ) {
-    const sessionValueToStore = {
+    // Do we want to partition these calls or merge in a single 512 call instead? Technically 512
+    // will be faster, and we'll occupy just one thread.
+    const [sid, aad] = await Promise.all([
+      this.randomBytes(256).then((sidBuffer) => sidBuffer.toString('base64')),
+      this.randomBytes(256).then((aadBuffer) => aadBuffer.toString('base64')),
+    ]);
+
+    const sessionExpirationInfo = this.calculateExpiry();
+    const createdSessionValue: SessionValue = {
       ...sessionValue,
-      ...this.calculateExpiry(sessionValue.lifespanExpiration),
-      path: this.serverBasePath,
+      ...sessionExpirationInfo,
+      sid,
+      state: await this.encryptState(sessionValue.state, aad),
+    };
+
+    // First try to store session in the index and only then in the cookie to make sure cookie is
+    // only updated if server side session is created successfully.
+    await this.options.sessionIndex.create(createdSessionValue);
+    await this.options.sessionCookie.set(request, { ...sessionExpirationInfo, sid, aad });
+
+    return {
+      ...createdSessionValue,
+      state: sessionValue.state,
     };
+  }
+
+  /**
+   * Creates or updates session value for the specified request.
+   * @param request Request instance to set session value for.
+   * @param sessionValue Session value parameters.
+   */
+  async update(request: KibanaRequest, sessionValue: SessionValue) {
+    const sessionCookieValue = await this.options.sessionCookie.get(request);
+    if (!sessionCookieValue) {
+      throw new Error('Session cannot be update since it doesnt exist.');
+    }
+
+    const sessionExpirationInfo = this.calculateExpiry(sessionCookieValue.lifespanExpiration);
+
+    // First try to store session in the index and only then in the cookie to make sure cookie is
+    // only updated if server side session is created successfully.
+    await this.options.sessionIndex.update({
+      ...sessionValue,
+      ...sessionExpirationInfo,
+      state: await this.encryptState(sessionValue.state, sessionCookieValue.aad),
+    });
+
+    await this.options.sessionCookie.set(request, {
+      ...sessionCookieValue,
+      ...sessionExpirationInfo,
+    });
+
+    return { ...sessionValue, ...sessionExpirationInfo };
+  }
+
+  /**
+   * Extends existing session.
+   * @param request Request instance to set session value for.
+   * @param sessionValue Session value parameters.
+   */
+  async extend(request: KibanaRequest, sessionValue: SessionValue) {
+    const sessionCookieValue = await this.options.sessionCookie.get(request);
+    if (!sessionCookieValue) {
+      throw new Error('Session cannot be extended since it doesnt exist.');
+    }
+
+    // We calculate actual expiration values based on the information extracted from the portion of
+    // the session value that is stored in the cookie since it always contains the most recent value.
+    const sessionExpirationInfo = this.calculateExpiry(sessionCookieValue.lifespanExpiration);
+    if (
+      sessionExpirationInfo.idleTimeoutExpiration === sessionValue.idleTimeoutExpiration &&
+      sessionExpirationInfo.lifespanExpiration === sessionValue.lifespanExpiration
+    ) {
+      return sessionValue;
+    }
+
+    // Session index updates are costly and should be minimized, but these are the cases when we
+    // should update session index:
+    let updateSessionIndex = false;
+    if (
+      (sessionExpirationInfo.idleTimeoutExpiration === null &&
+        sessionValue.idleTimeoutExpiration !== null) ||
+      (sessionExpirationInfo.idleTimeoutExpiration !== null &&
+        sessionValue.idleTimeoutExpiration === null)
+    ) {
+      // 1. If idle timeout wasn't configured when session was initially created and is configured
+      // now or vice versa.
+      this.options.logger.debug(
+        'Session idle timeout configuration has changed, session index will be updated.'
+      );
+      updateSessionIndex = true;
+    } else if (
+      (sessionExpirationInfo.lifespanExpiration === null &&
+        sessionValue.lifespanExpiration !== null) ||
+      (sessionExpirationInfo.lifespanExpiration !== null &&
+        sessionValue.lifespanExpiration === null)
+    ) {
+      // 2. If lifespan wasn't configured when session was initially created and is configured now
+      // or vice versa.
+      this.options.logger.debug(
+        'Session lifespan configuration has changed, session index will be updated.'
+      );
+      updateSessionIndex = true;
+    } else if (
+      this.idleIndexUpdateTimeout !== null &&
+      this.idleIndexUpdateTimeout <
+        sessionExpirationInfo.idleTimeoutExpiration! - sessionValue.idleTimeoutExpiration!
+    ) {
+      // 3. If idle timeout was updated a while ago.
+      this.options.logger.debug(
+        'Session idle timeout stored in the index is too old and will be updated.'
+      );
+      updateSessionIndex = true;
+    }
 
-    (await this.cookieSessionValueStorage).asScoped(request).set(sessionValueToStore);
+    // First try to store session in the index and only then in the cookie to make sure cookie is
+    // only updated if server side session is created successfully.
+    if (updateSessionIndex) {
+      await this.options.sessionIndex.update({ sid: sessionValue.sid, ...sessionExpirationInfo });
+    }
 
-    return sessionValueToStore;
+    await this.options.sessionCookie.set(request, {
+      ...sessionCookieValue,
+      ...sessionExpirationInfo,
+    });
+
+    return { ...sessionValue, ...sessionExpirationInfo };
   }
 
   /**
@@ -184,7 +254,51 @@ export class Session {
    * @param request Request instance to clear session value for.
    */
   async clear(request: KibanaRequest) {
-    (await this.cookieSessionValueStorage).asScoped(request).clear();
+    const sessionCookieValue = await this.options.sessionCookie.get(request);
+    if (!sessionCookieValue) {
+      return null;
+    }
+
+    await Promise.all([
+      this.options.sessionCookie.clear(request),
+      this.options.sessionIndex.clear(sessionCookieValue.sid),
+    ]);
+  }
+
+  /**
+   * Encrypts specified state.
+   * @param state State to encrypt.
+   * @param aad Additional authenticated data (AAD) to use for encryption.
+   */
+  private async encryptState(state: unknown, aad: string) {
+    if (state == null) {
+      return state;
+    }
+
+    try {
+      return await this.crypto.encrypt(JSON.stringify(state), aad);
+    } catch (err) {
+      this.options.logger.error(`Failed to encrypt session: ${err.message}`);
+      throw err;
+    }
+  }
+
+  /**
+   * Decrypts specified state.
+   * @param encryptedState State to decrypt.
+   * @param aad Additional authenticated data (AAD) used for encryption.
+   */
+  private async decryptState(encryptedState: string, aad: string) {
+    if (encryptedState == null) {
+      return encryptedState;
+    }
+
+    try {
+      return JSON.parse((await this.crypto.decrypt(encryptedState, aad)) as string);
+    } catch (err) {
+      this.options.logger.error(`Failed to decrypt session: ${err.message}`);
+      throw err;
+    }
   }
 
   private calculateExpiry(
diff --git a/x-pack/plugins/security/server/session_management/session_cookie.ts b/x-pack/plugins/security/server/session_management/session_cookie.ts
new file mode 100644
index 000000000000000..3ff2d3be855e0ce
--- /dev/null
+++ b/x-pack/plugins/security/server/session_management/session_cookie.ts
@@ -0,0 +1,136 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import {
+  HttpServiceSetup,
+  IBasePath,
+  KibanaRequest,
+  Logger,
+  SessionStorageFactory,
+} from '../../../../../src/core/server';
+import { ConfigType } from '../config';
+
+/**
+ * Represents shape of the session value stored in the cookie.
+ */
+export interface SessionCookieValue {
+  /**
+   * Unique session ID.
+   */
+  readonly sid: string;
+
+  /**
+   * Unique random value used as Additional authenticated data (AAD) while encrypting/decrypting
+   * sensitive or PII session content stored in the Elasticsearch index. This value is only stored
+   * in the user cookie.
+   */
+  readonly aad: string;
+
+  /**
+   * Cookie "Path" attribute that is validated against the current Kibana server configuration.
+   */
+  readonly path: string;
+
+  /**
+   * The Unix time in ms when the session should be considered expired. If `null`, session will stay
+   * active until the browser is closed.
+   */
+  readonly idleTimeoutExpiration: number | null;
+
+  /**
+   * The Unix time in ms which is the max total lifespan of the session. If `null`, session expire
+   * time can be extended indefinitely.
+   */
+  readonly lifespanExpiration: number | null;
+}
+
+export interface SessionCookieOptions {
+  logger: Logger;
+  basePath: IBasePath;
+  createCookieSessionStorageFactory: HttpServiceSetup['createCookieSessionStorageFactory'];
+  config: Pick<ConfigType, 'encryptionKey' | 'secureCookies' | 'cookieName'>;
+}
+
+export class SessionCookie {
+  /**
+   * Which base path the HTTP server is hosted on.
+   */
+  private readonly serverBasePath = this.options.basePath.serverBasePath || '/';
+
+  /**
+   * Promise containing initialized cookie session storage factory.
+   */
+  private readonly cookieSessionValueStorage: Promise<SessionStorageFactory<SessionCookieValue>>;
+
+  constructor(private readonly options: SessionCookieOptions) {
+    this.cookieSessionValueStorage = options.createCookieSessionStorageFactory({
+      encryptionKey: options.config.encryptionKey,
+      isSecure: options.config.secureCookies,
+      name: options.config.cookieName,
+      validate: (sessionValue: SessionCookieValue | SessionCookieValue[]) => {
+        // ensure that this cookie was created with the current Kibana configuration
+        const invalidSessionValue = (Array.isArray(sessionValue)
+          ? sessionValue
+          : [sessionValue]
+        ).find((sess) => sess.path !== undefined && sess.path !== this.serverBasePath);
+
+        // TODO: We should notify session index about that too.
+        if (invalidSessionValue) {
+          options.logger.debug(`Outdated session value with path "${invalidSessionValue.path}"`);
+          return { isValid: false, path: invalidSessionValue.path };
+        }
+
+        return { isValid: true };
+      },
+    });
+  }
+
+  /**
+   * Extracts session value for the specified request.
+   * @param request Request instance to get session value for.
+   */
+  async get(request: KibanaRequest) {
+    const sessionStorage = (await this.cookieSessionValueStorage).asScoped(request);
+    const sessionValue = await sessionStorage.get();
+
+    // If we detect that cookie session value is in incompatible format, then we should clear such
+    // cookie.
+    if (sessionValue && !SessionCookie.isSupportedSessionValue(sessionValue)) {
+      sessionStorage.clear();
+      return null;
+    }
+
+    return sessionValue;
+  }
+
+  /**
+   * Creates or updates session value for the specified request.
+   * @param request Request instance to set session value for.
+   * @param sessionValue Session value parameters.
+   */
+  async set(request: KibanaRequest, sessionValue: Omit<SessionCookieValue, 'path'>) {
+    (await this.cookieSessionValueStorage).asScoped(request).set({
+      ...sessionValue,
+      path: this.serverBasePath,
+    });
+  }
+
+  /**
+   * Clears session value for the specified request.
+   * @param request Request instance to clear session value for.
+   */
+  async clear(request: KibanaRequest) {
+    (await this.cookieSessionValueStorage).asScoped(request).clear();
+  }
+
+  /**
+   * Determines if session value was created by the current Kibana version. Previous versions had a different session value format.
+   * @param sessionValue The session value to check.
+   */
+  private static isSupportedSessionValue(sessionValue: any): sessionValue is SessionCookieValue {
+    return typeof sessionValue?.sid === 'string' && typeof sessionValue?.aad === 'string';
+  }
+}
diff --git a/x-pack/plugins/security/server/session_management/session_index.ts b/x-pack/plugins/security/server/session_management/session_index.ts
new file mode 100644
index 000000000000000..f6c70deed29af96
--- /dev/null
+++ b/x-pack/plugins/security/server/session_management/session_index.ts
@@ -0,0 +1,200 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { IClusterClient, Logger } from '../../../../../src/core/server';
+import { ConfigType } from '../config';
+import { SessionValue } from './session_value';
+
+export interface SessionIndexOptions {
+  readonly clusterClient: IClusterClient;
+  readonly config: Pick<ConfigType, 'session'>;
+  readonly logger: Logger;
+}
+
+export class SessionIndex {
+  /**
+   * Name of the Elasticsearch index that is used to store user session information.
+   */
+  private static INDEX_NAME = '.kibana_security_session';
+
+  /**
+   * Timeout after which session with the expired idle timeout _may_ be removed from the index
+   * during regular cleanup routine. It's intentionally larger than `idleIndexUpdateTimeout`
+   * configured in `Session` to be sure that the may be safely cleaned up.
+   */
+  private readonly idleIndexCleanupTimeout = this.options.config.session.idleTimeout
+    ? this.options.config.session.idleTimeout.asMilliseconds() * 3
+    : null;
+
+  constructor(private readonly options: SessionIndexOptions) {}
+
+  /**
+   * Creates new document for the session value.
+   * @param sessionValue Session value.
+   */
+  async create(sessionValue: SessionValue) {
+    const { sid, ...sessionValueToStore } = sessionValue;
+    try {
+      await this.options.clusterClient.callAsInternalUser('create', {
+        id: sid,
+        index: SessionIndex.INDEX_NAME,
+        body: sessionValueToStore,
+        refresh: 'wait_for',
+      });
+    } catch (err) {
+      this.options.logger.error(`Failed to create session: ${err.message}`);
+      throw err;
+    }
+  }
+
+  /**
+   * Makes a partial update of the existing session value.
+   * @param sessionValue Session value.
+   */
+  async update(sessionValue: { sid: string } & Partial<SessionValue>) {
+    const { sid, ...sessionValueToStore } = sessionValue;
+    try {
+      await this.options.clusterClient.callAsInternalUser('update', {
+        id: sid,
+        index: SessionIndex.INDEX_NAME,
+        body: { doc: sessionValueToStore },
+        refresh: 'wait_for',
+      });
+    } catch (err) {
+      this.options.logger.error(`Failed to update session: ${err.message}`);
+      throw err;
+    }
+  }
+
+  /**
+   * Retrieves session value with the specified ID from the index. If session value isn't found
+   * `null` will be returned.
+   * @param sid Session ID.
+   */
+  async get(sid: string) {
+    try {
+      const response = await this.options.clusterClient.callAsInternalUser('get', {
+        id: sid,
+        ignore: [404],
+        index: SessionIndex.INDEX_NAME,
+      });
+
+      if (response.status === 404) {
+        return null;
+      }
+
+      return { sid, ...response._source } as SessionValue;
+    } catch (err) {
+      this.options.logger.error(`Failed to retrieve session: ${err.message}`);
+      throw err;
+    }
+  }
+
+  /**
+   * Clears session value with the specified ID.
+   * @param sid Session ID to clear.
+   */
+  async clear(sid: string) {
+    try {
+      const now = Date.now();
+
+      // Always try to delete session with the specified ID and with expired lifespan (even if it's
+      // not configured right now).
+      const deleteQueries: object[] = [
+        { term: { _id: sid } },
+        { range: { lifespanExpiration: { lte: now } } },
+      ];
+
+      // If lifespan is configured we should remove sessions that were created without it if any.
+      if (this.options.config.session.lifespan) {
+        deleteQueries.push({ bool: { must_not: { exists: { field: 'lifespanExpiration' } } } });
+      }
+
+      // If idle timeout is configured we should delete all sessions without specified idle timeout
+      // or if that session hasn't been updated for a while meaning that session is expired.
+      if (this.idleIndexCleanupTimeout) {
+        deleteQueries.push(
+          { range: { idleTimeoutExpiration: { lte: now - this.idleIndexCleanupTimeout } } },
+          { bool: { must_not: { exists: { field: 'idleTimeoutExpiration' } } } }
+        );
+      } else {
+        // Otherwise just delete all expired sessions that were previously created with the idle
+        // timeout if any.
+        deleteQueries.push({ range: { idleTimeoutExpiration: { lte: now } } });
+      }
+
+      await this.options.clusterClient.callAsInternalUser('deleteByQuery', {
+        index: SessionIndex.INDEX_NAME,
+        refresh: 'wait_for',
+        body: { conflicts: 'proceed', query: { bool: { should: deleteQueries } } },
+      });
+    } catch (err) {
+      this.options.logger.error(`Failed to clear session: ${err.message}`);
+      throw err;
+    }
+  }
+
+  /**
+   * Initializes index that is used to store session values.
+   */
+  async initialize() {
+    // TODO: Check the version in _meta field.
+    // TODO: Migrate if version changed
+    // TODO: We should be able to recreate index on the fly if it's deleted for some reason.
+
+    this.options.logger.debug('Initializing session index.');
+
+    try {
+      const result = await this.options.clusterClient.callAsInternalUser('indices.get', {
+        ignore: [404],
+        index: SessionIndex.INDEX_NAME,
+      });
+
+      const indexExists = result?.status !== 404;
+      if (indexExists) {
+        this.options.logger.debug('Session index exists, no further action is necessary.');
+        return;
+      }
+    } catch (err) {
+      this.options.logger.error(`Failed to check if session index exists: ${err.message}`);
+      throw err;
+    }
+
+    this.options.logger.debug('Session index does not exist and will be created.');
+
+    try {
+      await this.options.clusterClient.callAsInternalUser('indices.create', {
+        body: {
+          mappings: {
+            dynamic: 'strict',
+            properties: {
+              username_hash: { type: 'keyword' },
+              roles: { type: 'keyword' },
+              provider: { properties: { type: { type: 'keyword' }, name: { type: 'keyword' } } },
+              idleTimeoutExpiration: { type: 'date' },
+              lifespanExpiration: { type: 'date' },
+              accessAgreementAcknowledged: { type: 'boolean' },
+              state: { type: 'binary' },
+            },
+          },
+          settings: {
+            number_of_shards: 1,
+            number_of_replicas: 0,
+            auto_expand_replicas: '0-1',
+            'index.priority': 1000,
+            'index.refresh_interval': '1s',
+          },
+        },
+        index: SessionIndex.INDEX_NAME,
+      });
+    } catch (err) {
+      this.options.logger.error(`Failed to create session index: ${err.message}`);
+      throw err;
+    }
+
+    this.options.logger.debug('Successfully created session index.');
+  }
+}
diff --git a/x-pack/plugins/security/server/session_management/session_management_service.ts b/x-pack/plugins/security/server/session_management/session_management_service.ts
new file mode 100644
index 000000000000000..29a38a59103d513
--- /dev/null
+++ b/x-pack/plugins/security/server/session_management/session_management_service.ts
@@ -0,0 +1,83 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { Observable, Subscription } from 'rxjs';
+import { HttpServiceSetup, IClusterClient, Logger } from '../../../../../src/core/server';
+import { SecurityAuditLogger } from '../audit';
+import { ConfigType } from '../config';
+import { OnlineStatusRetryScheduler } from '../elasticsearch';
+import { SessionCookie } from './session_cookie';
+import { SessionIndex } from './session_index';
+import { Session } from './session';
+
+export interface SessionManagementServiceSetupParams {
+  readonly auditLogger: SecurityAuditLogger;
+  readonly http: Pick<HttpServiceSetup, 'basePath' | 'createCookieSessionStorageFactory'>;
+  readonly config: ConfigType;
+  readonly clusterClient: IClusterClient;
+}
+
+export interface SessionManagementServiceStartParams {
+  readonly online$: Observable<OnlineStatusRetryScheduler>;
+}
+
+export interface SessionManagementServiceSetup {
+  readonly session: Session;
+}
+
+export class SessionManagementService {
+  private statusSubscription?: Subscription;
+  private sessionIndex!: SessionIndex;
+
+  constructor(private readonly logger: Logger) {}
+
+  setup({
+    auditLogger,
+    config,
+    clusterClient,
+    http,
+  }: SessionManagementServiceSetupParams): SessionManagementServiceSetup {
+    const sessionCookie = new SessionCookie({
+      config,
+      createCookieSessionStorageFactory: http.createCookieSessionStorageFactory,
+      basePath: http.basePath,
+      logger: this.logger.get('cookie'),
+    });
+
+    this.sessionIndex = new SessionIndex({
+      config,
+      clusterClient,
+      logger: this.logger.get('index'),
+    });
+
+    return {
+      session: new Session({
+        auditLogger,
+        logger: this.logger,
+        sessionCookie,
+        sessionIndex: this.sessionIndex,
+        config,
+      }),
+    };
+  }
+
+  start({ online$ }: SessionManagementServiceStartParams) {
+    this.statusSubscription = online$.subscribe(async ({ scheduleRetry }) => {
+      try {
+        await this.sessionIndex.initialize();
+      } catch (err) {
+        scheduleRetry();
+      }
+    });
+  }
+
+  stop() {
+    if (this.statusSubscription !== undefined) {
+      this.statusSubscription.unsubscribe();
+      this.statusSubscription = undefined;
+    }
+  }
+}
diff --git a/x-pack/plugins/security/server/session_management/session_value.ts b/x-pack/plugins/security/server/session_management/session_value.ts
new file mode 100644
index 000000000000000..d305f95306ffb61
--- /dev/null
+++ b/x-pack/plugins/security/server/session_management/session_value.ts
@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { AuthenticationProvider } from '../../common/types';
+
+/**
+ * The shape of the session value stored in the Elasticsearch index.
+ */
+export interface SessionValue {
+  /**
+   * Unique session ID.
+   */
+  readonly sid: string;
+
+  /**
+   * Name and type of the provider this session belongs to.
+   */
+  readonly provider: AuthenticationProvider;
+
+  /**
+   * The Unix time in ms when the session should be considered expired. If `null`, session will stay
+   * active until the browser is closed.
+   */
+  readonly idleTimeoutExpiration: number | null;
+
+  /**
+   * The Unix time in ms which is the max total lifespan of the session. If `null`, session expire
+   * time can be extended indefinitely.
+   */
+  readonly lifespanExpiration: number | null;
+
+  /**
+   * Session value that is fed to the authentication provider. The shape is unknown upfront and
+   * entirely determined by the authentication provider that owns the current session.
+   */
+  readonly state: unknown;
+
+  /**
+   * Indicates whether user acknowledged access agreement or not.
+   */
+  readonly accessAgreementAcknowledged?: boolean;
+}