System subject

server/src/main/java/org/opensearch/cluster/service/ClusterApplierService.java

@@ -61,8 +61,8 @@
 import org.opensearch.common.util.concurrent.OpenSearchExecutors;
 import org.opensearch.common.util.concurrent.PrioritizedOpenSearchThreadPoolExecutor;
 import org.opensearch.common.util.concurrent.ThreadContext;
-import org.opensearch.common.util.concurrent.ThreadContextAccess;
 import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException;
+import org.opensearch.identity.SystemSubject;
 import org.opensearch.telemetry.metrics.noop.NoopMetricsRegistry;
 import org.opensearch.telemetry.metrics.tags.Tags;
 import org.opensearch.threadpool.Scheduler;
@@ -396,31 +396,33 @@ private void submitStateUpdateTask(
         }
         final ThreadContext threadContext = threadPool.getThreadContext();
         final Supplier<ThreadContext.StoredContext> supplier = threadContext.newRestorableContext(true);
-        try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
-            ThreadContextAccess.doPrivilegedVoid(threadContext::markAsSystemContext);
-            final UpdateTask updateTask = new UpdateTask(
-                config.priority(),
-                source,
-                new SafeClusterApplyListener(listener, supplier, logger),
-                executor
-            );
-            if (config.timeout() != null) {
-                threadPoolExecutor.execute(
-                    updateTask,
-                    config.timeout(),
-                    () -> threadPool.generic()
-                        .execute(() -> listener.onFailure(source, new ProcessClusterEventTimeoutException(config.timeout(), source)))
+        SystemSubject.getInstance().runAs(() -> {
+            try {
+                final UpdateTask updateTask = new UpdateTask(
+                    config.priority(),
+                    source,
+                    new SafeClusterApplyListener(listener, supplier, logger),
+                    executor
                 );
-            } else {
-                threadPoolExecutor.execute(updateTask);
-            }
-        } catch (OpenSearchRejectedExecutionException e) {
-            // ignore cases where we are shutting down..., there is really nothing interesting
-            // to be done here...
-            if (!lifecycle.stoppedOrClosed()) {
-                throw e;
+                if (config.timeout() != null) {
+                    threadPoolExecutor.execute(
+                        updateTask,
+                        config.timeout(),
+                        () -> threadPool.generic()
+                            .execute(() -> listener.onFailure(source, new ProcessClusterEventTimeoutException(config.timeout(), source)))
+                    );
+                } else {
+                    threadPoolExecutor.execute(updateTask);
+                }
+            } catch (OpenSearchRejectedExecutionException e) {
+                // ignore cases where we are shutting down..., there is really nothing interesting
+                // to be done here...
+                if (!lifecycle.stoppedOrClosed()) {
+                    throw e;
+                }
             }
-        }
+            return null;
+        });
     }
 
     /** asserts that the current thread is <b>NOT</b> the cluster state update thread */

server/src/main/java/org/opensearch/cluster/service/MasterService.java

@@ -66,11 +66,11 @@
 import org.opensearch.common.util.concurrent.OpenSearchExecutors;
 import org.opensearch.common.util.concurrent.PrioritizedOpenSearchThreadPoolExecutor;
 import org.opensearch.common.util.concurrent.ThreadContext;
-import org.opensearch.common.util.concurrent.ThreadContextAccess;
 import org.opensearch.core.Assertions;
 import org.opensearch.core.common.text.Text;
 import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException;
 import org.opensearch.discovery.Discovery;
+import org.opensearch.identity.SystemSubject;
 import org.opensearch.node.Node;
 import org.opensearch.telemetry.metrics.noop.NoopMetricsRegistry;
 import org.opensearch.telemetry.metrics.tags.Tags;
@@ -1022,21 +1022,23 @@ public <T> void submitStateUpdateTasks(
         }
         final ThreadContext threadContext = threadPool.getThreadContext();
         final Supplier<ThreadContext.StoredContext> supplier = threadContext.newRestorableContext(true);
-        try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
-            ThreadContextAccess.doPrivilegedVoid(threadContext::markAsSystemContext);
-
-            List<Batcher.UpdateTask> safeTasks = tasks.entrySet()
-                .stream()
-                .map(e -> taskBatcher.new UpdateTask(config.priority(), source, e.getKey(), safe(e.getValue(), supplier), executor))
-                .collect(Collectors.toList());
-            taskBatcher.submitTasks(safeTasks, config.timeout());
-        } catch (OpenSearchRejectedExecutionException e) {
-            // ignore cases where we are shutting down..., there is really nothing interesting
-            // to be done here...
-            if (!lifecycle.stoppedOrClosed()) {
-                throw e;
+
+        SystemSubject.getInstance().runAs(() -> {
+            try {
+                List<Batcher.UpdateTask> safeTasks = tasks.entrySet()
+                    .stream()
+                    .map(e -> taskBatcher.new UpdateTask(config.priority(), source, e.getKey(), safe(e.getValue(), supplier), executor))
+                    .collect(Collectors.toList());
+                taskBatcher.submitTasks(safeTasks, config.timeout());
+            } catch (OpenSearchRejectedExecutionException e) {
+                // ignore cases where we are shutting down..., there is really nothing interesting
+                // to be done here...
+                if (!lifecycle.stoppedOrClosed()) {
+                    throw e;
+                }
             }
-        }
+            return null;
+        });
     }
 
     public ClusterStateStats getClusterStateStats() {

server/src/main/java/org/opensearch/http/AbstractHttpServerTransport.java

@@ -52,6 +52,7 @@
 import org.opensearch.core.common.transport.TransportAddress;
 import org.opensearch.core.common.unit.ByteSizeValue;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.identity.SystemSubject;
 import org.opensearch.rest.RestChannel;
 import org.opensearch.rest.RestRequest;
 import org.opensearch.telemetry.tracing.Span;
@@ -383,9 +384,9 @@ public void incomingRequest(final HttpRequest httpRequest, final HttpChannel htt
 
     // Visible for testing
     void dispatchRequest(final RestRequest restRequest, final RestChannel channel, final Throwable badRequestCause) {
-        RestChannel traceableRestChannel = channel;
         final ThreadContext threadContext = threadPool.getThreadContext();
-        try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
+        SystemSubject.getInstance().runAs(() -> {
+            RestChannel traceableRestChannel = channel;
             final Span span = tracer.startSpan(SpanBuilder.from(restRequest));
             try (final SpanScope spanScope = tracer.withSpanInScope(span)) {
                 if (channel != null) {
@@ -397,8 +398,8 @@ void dispatchRequest(final RestRequest restRequest, final RestChannel channel, f
                     dispatcher.dispatchRequest(restRequest, traceableRestChannel, threadContext);
                 }
             }
-        }
-
+            return null;
+        });
     }
 
     private void handleIncomingRequest(final HttpRequest httpRequest, final HttpChannel httpChannel, final Exception exception) {

server/src/main/java/org/opensearch/identity/IdentityService.java

@@ -32,6 +32,7 @@ public class IdentityService {
 
     public IdentityService(final Settings settings, final ThreadPool threadPool, final List<IdentityPlugin> identityPlugins) {
         this.settings = settings;
+        SystemSubject.getInstance().initialize(threadPool);
 
         if (identityPlugins.size() == 0) {
             log.debug("Identity plugins size is 0");

server/src/main/java/org/opensearch/identity/SystemSubject.java

@@ -0,0 +1,69 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.identity;
+
+import org.opensearch.common.annotation.InternalApi;
+import org.opensearch.common.util.concurrent.ThreadContext;
+import org.opensearch.common.util.concurrent.ThreadContextAccess;
+import org.opensearch.threadpool.ThreadPool;
+
+import java.security.Principal;
+import java.util.concurrent.Callable;
+
+/**
+ * A special reserved {@link Subject} that represents the internal system subject
+ *
+ * @opensearch.internal
+ */
+@InternalApi
+public class SystemSubject implements Subject {
+    private static final SystemSubject INSTANCE = new SystemSubject();
+
+    private static final SystemPrincipal SYSTEM_PRINCIPAL = new SystemPrincipal();
+
+    private ThreadPool threadPool;
+
+    private SystemSubject() {}
+
+    public static SystemSubject getInstance() {
+        return SystemSubject.INSTANCE;
+    }
+
+    void initialize(ThreadPool threadPool) {
+        this.threadPool = threadPool;
+    }
+
+    @Override
+    public Principal getPrincipal() {
+        return SYSTEM_PRINCIPAL;
+    }
+
+    @Override
+    public <T> T runAs(Callable<T> callable) throws RuntimeException {
+        ThreadContext threadContext = threadPool.getThreadContext();
+        try (ThreadContext.StoredContext ctx = threadPool.getThreadContext().stashContext()) {
+            ThreadContextAccess.doPrivilegedVoid(threadContext::markAsSystemContext);
+            try {
+                return callable.call();
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+    private static class SystemPrincipal implements Principal {
+
+        private SystemPrincipal() {}
+
+        @Override
+        public String getName() {
+            return "system";
+        }
+    }
+}

server/src/main/java/org/opensearch/index/seqno/GlobalCheckpointSyncAction.java

@@ -43,11 +43,10 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.concurrent.ThreadContext;
-import org.opensearch.common.util.concurrent.ThreadContextAccess;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.index.shard.ShardId;
+import org.opensearch.identity.SystemSubject;
 import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.shard.IndexShardClosedException;
 import org.opensearch.index.translog.Translog;
@@ -97,15 +96,14 @@ public GlobalCheckpointSyncAction(
     }
 
     public void updateGlobalCheckpointForShard(final ShardId shardId) {
-        final ThreadContext threadContext = threadPool.getThreadContext();
-        try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
-            ThreadContextAccess.doPrivilegedVoid(threadContext::markAsSystemContext);
+        SystemSubject.getInstance().runAs(() -> {
             execute(new Request(shardId), ActionListener.wrap(r -> {}, e -> {
                 if (ExceptionsHelper.unwrap(e, AlreadyClosedException.class, IndexShardClosedException.class) == null) {
                     logger.info(new ParameterizedMessage("{} global checkpoint sync failed", shardId), e);
                 }
             }));
-        }
+            return null;
+        });
     }
 
     @Override

server/src/main/java/org/opensearch/index/seqno/RetentionLeaseBackgroundSyncAction.java

@@ -47,13 +47,12 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.concurrent.ThreadContext;
-import org.opensearch.common.util.concurrent.ThreadContextAccess;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.core.tasks.TaskId;
+import org.opensearch.identity.SystemSubject;
 import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.shard.IndexShardClosedException;
@@ -120,10 +119,7 @@ protected void doExecute(Task task, Request request, ActionListener<ReplicationR
     }
 
     final void backgroundSync(ShardId shardId, String primaryAllocationId, long primaryTerm, RetentionLeases retentionLeases) {
-        final ThreadContext threadContext = threadPool.getThreadContext();
-        try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
-            // we have to execute under the system context so that if security is enabled the sync is authorized
-            ThreadContextAccess.doPrivilegedVoid(threadContext::markAsSystemContext);
+        SystemSubject.getInstance().runAs(() -> {
             final Request request = new Request(shardId, retentionLeases);
             final ReplicationTask task = (ReplicationTask) taskManager.register("transport", "retention_lease_background_sync", request);
             transportService.sendChildRequest(
@@ -170,7 +166,8 @@ public void handleException(TransportException e) {
                     }
                 }
             );
-        }
+            return null;
+        });
     }
 
     @Override

server/src/main/java/org/opensearch/index/seqno/RetentionLeaseSyncAction.java

@@ -49,13 +49,12 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.concurrent.ThreadContext;
-import org.opensearch.common.util.concurrent.ThreadContextAccess;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.core.tasks.TaskId;
+import org.opensearch.identity.SystemSubject;
 import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.index.IndexingPressureService;
 import org.opensearch.index.shard.IndexShard;
@@ -135,10 +134,7 @@ final void sync(
         RetentionLeases retentionLeases,
         ActionListener<ReplicationResponse> listener
     ) {
-        final ThreadContext threadContext = threadPool.getThreadContext();
-        try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
-            // we have to execute under the system context so that if security is enabled the sync is authorized
-            ThreadContextAccess.doPrivilegedVoid(threadContext::markAsSystemContext);
+        SystemSubject.getInstance().runAs(() -> {
             final Request request = new Request(shardId, retentionLeases);
             final ReplicationTask task = (ReplicationTask) taskManager.register("transport", "retention_lease_sync", request);
             transportService.sendChildRequest(
@@ -181,7 +177,8 @@ public void handleException(TransportException e) {
                     }
                 }
             );
-        }
+            return null;
+        });
     }
 
     @Override

server/src/main/java/org/opensearch/indices/replication/checkpoint/PublishCheckpointAction.java

@@ -23,10 +23,9 @@
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.concurrent.ThreadContext;
-import org.opensearch.common.util.concurrent.ThreadContextAccess;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.identity.SystemSubject;
 import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.shard.IndexShardClosedException;
@@ -111,10 +110,7 @@ public ReplicationMode getReplicationMode(IndexShard indexShard) {
     final void publish(IndexShard indexShard, ReplicationCheckpoint checkpoint) {
         String primaryAllocationId = indexShard.routingEntry().allocationId().getId();
         long primaryTerm = indexShard.getPendingPrimaryTerm();
-        final ThreadContext threadContext = threadPool.getThreadContext();
-        try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
-            // we have to execute under the system context so that if security is enabled the sync is authorized
-            ThreadContextAccess.doPrivilegedVoid(threadContext::markAsSystemContext);
+        SystemSubject.getInstance().runAs(() -> {
             PublishCheckpointRequest request = new PublishCheckpointRequest(checkpoint);
             final ReplicationTask task = (ReplicationTask) taskManager.register("transport", "segrep_publish_checkpoint", request);
             final ReplicationTimer timer = new ReplicationTimer();
@@ -182,7 +178,8 @@ public void handleException(TransportException e) {
                     checkpoint
                 )
             );
-        }
+            return null;
+        });
     }
 
     @Override